You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2015/03/17 00:37:44 UTC
[16/50] incubator-usergrid git commit: Added a master only elastic
search node
Added a master only elastic search node
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/e58597be
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/e58597be
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/e58597be
Branch: refs/heads/two-dot-o
Commit: e58597be99d18ed1f3bfa4387746bdc0254ff122
Parents: b3e42dd
Author: Todd Nine <tn...@apigee.com>
Authored: Wed Mar 11 10:14:09 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Wed Mar 11 10:14:09 2015 -0600
----------------------------------------------------------------------
.../dist/init_instance/install_elasticsearch.sh | 12 +-
.../main/groovy/configure_elasticsearch.groovy | 70 ++--
stack/awscluster/ugcluster-cf.json | 326 ++++++++++++++++++-
.../events/EntityDeletedHandler.java | 2 +-
.../events/EntityVersionCreatedHandler.java | 2 +-
.../usergrid/persistence/index/EntityIndex.java | 26 +-
.../index/impl/EsEntityIndexImpl.java | 166 +++++-----
.../index/impl/EsIndexBufferConsumerImpl.java | 5 +-
8 files changed, 476 insertions(+), 133 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e58597be/stack/awscluster/src/main/dist/init_instance/install_elasticsearch.sh
----------------------------------------------------------------------
diff --git a/stack/awscluster/src/main/dist/init_instance/install_elasticsearch.sh b/stack/awscluster/src/main/dist/init_instance/install_elasticsearch.sh
index 06af849..b2018e1 100644
--- a/stack/awscluster/src/main/dist/init_instance/install_elasticsearch.sh
+++ b/stack/awscluster/src/main/dist/init_instance/install_elasticsearch.sh
@@ -42,9 +42,17 @@ echo "vm.swappiness = 0" >> /etc/sysctl.conf
sysctl -p
# No need to do this, elasticsearch nodes are also cassandra nodes
+
cd /usr/share/usergrid/scripts
-groovy registry_register.groovy elasticsearch
-groovy wait_for_instances.groovy elasticsearch ${ES_NUM_SERVERS}
+
+#If we're the master, register ourselves and move on, if we're not, wait for the master to come up
+if [ "$ES_MASTER" = "true" ]; then
+ groovy registry_register.groovy elasticsearch_master
+else
+ groovy registry_register.groovy elasticsearch
+ groovy wait_for_instances.groovy elasticsearch_master 1
+fi
+
# leave room for Cassandra: use about one half of RAM for heap
case `(curl http://169.254.169.254/latest/meta-data/instance-type)` in
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e58597be/stack/awscluster/src/main/groovy/configure_elasticsearch.groovy
----------------------------------------------------------------------
diff --git a/stack/awscluster/src/main/groovy/configure_elasticsearch.groovy b/stack/awscluster/src/main/groovy/configure_elasticsearch.groovy
index 7bee82c..19c969d 100644
--- a/stack/awscluster/src/main/groovy/configure_elasticsearch.groovy
+++ b/stack/awscluster/src/main/groovy/configure_elasticsearch.groovy
@@ -18,10 +18,10 @@
-//
-// configure_elasticsearch.groovy
-//
-// Emits Elasticsearch config file based on environment and Elasticsearch node
+//
+// configure_elasticsearch.groovy
+//
+// Emits Elasticsearch config file based on environment and Elasticsearch node
// registry in SimpleDB
//
@@ -29,13 +29,18 @@
String hostName = (String)System.getenv().get("PUBLIC_HOSTNAME")
def clusterName = (String)System.getenv().get("ES_CLUSTER_NAME")
+def isMaster = ((String)System.getenv().get("ES_MASTER")).toBoolean()
+
int esNumServers = ((String)System.getenv().get("ES_NUM_SERVERS")).toInteger()
-int quorum = esNumServers/2+1;
+///int quorum = esNumServers/2+1;
+
+//TODO get this from the number of master nodes
+int quorum = 1
NodeRegistry registry = new NodeRegistry();
// build seed list by listing all Elasticsearch nodes found in SimpleDB domain with our stackName
-def selectResult = registry.searchNode('elasticsearch')
+def selectResult = registry.searchNode('elasticsearch_master')
def esnodes = ""
def sep = ""
for (hostname in selectResult) {
@@ -43,13 +48,19 @@ for (hostname in selectResult) {
sep = ","
}
+
+def nodeData = !isMaster
+def nodeMaster = isMaster
+
+
+
def elasticSearchConfig = """
cluster.name: ${clusterName}
discovery.zen.minimum_master_nodes: ${quorum}
discovery.zen.ping.multicast.enabled: false
discovery.zen.ping.unicast.hosts: [${esnodes}]
node:
- name: ${hostName}
+ name: ${hostName}
network:
host: ${hostName}
path:
@@ -65,21 +76,20 @@ es.logger.level: INFO
# Temporarily removing. We don't know better :)
# http://www.elasticsearch.org/guide/en/elasticsearch/guide/current/_don_8217_t_touch_these_settings.html#_threadpools
#
-#threadpool:
-# index:
-# type: fixed
-# size: 160
-# queue_size: 0
-# bulk:
-# type: fixed
-# size: 5000
-# size: 16
-# queue_size: 100
-# search:
-# size: 10000
-# size: 48
-# type: fixed
-# queue_size: 100
+threadpool:
+ index:
+ type: fixed
+ size: 160
+ queue_size: 1000
+ bulk:
+ type: fixed
+ size: 160
+ size: 16
+ queue_size: 1000
+ search:
+ size: 320
+ type: fixed
+ queue_size: 1000
action.auto_create_index: false
@@ -108,10 +118,18 @@ cluster.routing.allocation.disk.watermark.low: .97
cluster.routing.allocation.disk.watermark.high: .99
#Set streaming high water marks so reboots don't kill our service
-cluster.routing.allocation.node_concurrent_recoveries: 4
-cluster.routing.allocation.node_initial_primaries_recoveries: 18
-indices.recovery.concurrent_streams: 4
-indices.recovery.max_bytes_per_sec: 40mb
+cluster.routing.allocation.node_concurrent_recoveries: 40
+cluster.routing.allocation.node_initial_primaries_recoveries: 40
+indices.recovery.concurrent_streams: 16
+indices.recovery.max_bytes_per_sec: 300mb
+
+
+##############################
+# Master or data node options
+#############################
+
+node.data: ${nodeData}
+node.master: ${nodeMaster}
###############
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e58597be/stack/awscluster/ugcluster-cf.json
----------------------------------------------------------------------
diff --git a/stack/awscluster/ugcluster-cf.json b/stack/awscluster/ugcluster-cf.json
index 7926929..0c2a9d2 100644
--- a/stack/awscluster/ugcluster-cf.json
+++ b/stack/awscluster/ugcluster-cf.json
@@ -109,6 +109,14 @@
"Default": "6",
"MinValue": "2"
},
+ "ESNumMasterServers": {
+ "Description": "Number of ES master servers to start.",
+ "Type": "Number",
+ "Default": "1",
+ "MinValue": "1"
+ },
+
+
"ESInstanceType": {
"Description": "Instance type for ES servers",
"Type": "String",
@@ -947,7 +955,7 @@
}
}
},
- "ESAutoScalingLaunchConfiguration":{
+ "ESMasterAutoScalingLaunchConfiguration":{
"Type":"AWS::AutoScaling::LaunchConfiguration",
"Properties":{
"UserData":{
@@ -979,7 +987,10 @@
"\n",
"export ES_CLUSTER_NAME=", { "Ref":"ESClusterName" }, "\n",
"export ES_NUM_SERVERS=", { "Ref":"ESNumServers" }, "\n",
+ "export ES_MASTER=true", "\n",
+
"export GRAPHITE_NUM_SERVERS=", { "Ref":"GraphiteNumServers" }, "\n",
+
"\n",
"export RELEASE_BUCKET=", { "Ref":"ReleaseBucket" }, "\n",
"\n",
@@ -1062,7 +1073,7 @@
]
}
},
- "ESAutoScalingGroup": {
+ "ESMasterAutoScalingGroup": {
"Type": "AWS::AutoScaling::AutoScalingGroup",
"Version": "2009-05-15",
"Properties": {
@@ -1105,13 +1116,13 @@
}
],
"LaunchConfigurationName": {
- "Ref": "ESAutoScalingLaunchConfiguration"
+ "Ref": "ESMasterAutoScalingLaunchConfiguration"
},
"MinSize": {
- "Ref": "ESNumServers"
+ "Ref": "ESNumMasterServers"
},
"MaxSize": {
- "Ref": "ESNumServers"
+ "Ref": "ESNumMasterServers"
},
"NotificationConfiguration": {
"TopicARN": {
@@ -1251,6 +1262,311 @@
]
}
},
+ "ESAutoScalingLaunchConfiguration":{
+ "Type":"AWS::AutoScaling::LaunchConfiguration",
+ "Properties":{
+ "UserData":{
+ "Fn::Base64":{
+ "Fn::Join":[
+ "",
+ [
+ "#!/bin/bash -ex\n",
+ "# ES NODE STARTUP \n",
+ "exec >/var/log/usergrid-bootstrap.log 2>&1\n",
+ "\n",
+ "mkdir -p /usr/share/usergrid\n",
+ "\n",
+ "# create script that sets our environment variables\n",
+ "cat >/etc/profile.d/usergrid-env.sh <<EOF\n",
+ "alias sudo='sudo -E'\n", "\n",
+ "export TYPE=es\n",
+ "export STACK_NAME=", { "Ref":"AWS::StackName" }, "\n", "\n",
+ "export PUBLIC_HOSTNAME=`(curl -s http://169.254.169.254/latest/meta-data/public-hostname)`\n",
+ "export INTERNAL_HOSTNAME=`(curl http://169.254.169.254/latest/meta-data/local-ipv4)`\n",
+ "\n",
+ "export EC2_INSTANCE_ID=`ec2metadata --instance-id`\n",
+ "export EC2_REGION=", { "Ref":"AWS::Region" },
+ "\n",
+ "export EC2_URL=https://ec2.amazonaws.com/\n",
+ "\n",
+ "export ES_SECURITY_GROUP_NAME=", { "Ref":"ESSecurityGroup" }, "\n",
+ "\n",
+ "\n",
+ "export ES_CLUSTER_NAME=", { "Ref":"ESClusterName" }, "\n",
+ "export ES_NUM_SERVERS=", { "Ref":"ESNumServers" }, "\n",
+ "export ES_MASTER=false","\n",
+ "export GRAPHITE_NUM_SERVERS=", { "Ref":"GraphiteNumServers" }, "\n",
+ "\n",
+ "export RELEASE_BUCKET=", { "Ref":"ReleaseBucket" }, "\n",
+ "\n",
+ "EOF\n",
+ "\n",
+ "# put AWS creds in environment\n",
+ "cat >/etc/profile.d/aws-credentials.sh <<EOF\n",
+ "export AWS_ACCESS_KEY=", { "Ref":"ESKey" }, "\n",
+ "export AWS_SECRET_KEY=", { "Fn::GetAtt":[ "ESKey", "SecretAccessKey" ] }, "\n",
+ "EOF\n",
+ "\n",
+ "# setup s3cmd (will be installed by init script) \n",
+ "cat >/etc/s3cfg <<EOF\n",
+ "access_key=", { "Ref":"ESKey" }, "\n",
+ "secret_key=", { "Fn::GetAtt":[ "ESKey", "SecretAccessKey" ] },
+ "\n",
+ "EOF\n",
+ "chmod 644 /etc/s3cfg\n",
+ "ln -s /etc/s3cfg ~ubuntu/.s3cfg\n",
+ "ln -s /etc/s3cfg ~root/.s3cfg\n",
+ "\n",
+ "# download usergrid and init script bundle from S3\n",
+ "wget -O- -q http://s3tools.org/repo/deb-all/stable/s3tools.key | apt-key add -\n",
+ "wget -O/etc/apt/sources.list.d/s3tools.list http://s3tools.org/repo/deb-all/stable/s3tools.list\n",
+ "apt-get update\n",
+ "apt-get -y install s3cmd\n",
+ "cd /usr/share/usergrid\n",
+ "s3cmd --config=/etc/s3cfg get s3://", { "Ref":"ReleaseBucket" }, "/awscluster-1.0-SNAPSHOT-any.tar.gz\n",
+ "tar xvf awscluster-1.0-SNAPSHOT-any.tar.gz\n",
+ "rm -fr awscluster-1.0-SNAPSHOT-any.tar.gz\n",
+ "chmod 755 ./init_instance/*.sh\n",
+ "cd ./init_instance\n",
+ "# init as an ES node \n",
+ "sh ./init_es_server.sh\n"
+ ]
+ ]
+ }
+ },
+ "KeyName": {
+ "Ref": "KeyPair"
+ },
+ "ImageId": {
+ "Fn::FindInMap": [
+ "AWSRegionArch2AMI",
+ {
+ "Ref": "AWS::Region"
+ },
+ {
+ "Fn::FindInMap": [
+ "AWSInstanceType2Arch",
+ {
+ "Ref": "CassInstanceType"
+ },
+ "Arch"
+ ]
+ }
+ ]
+ },
+ "InstanceType": {
+ "Ref": "ESInstanceType"
+ },
+ "IamInstanceProfile": {
+ "Ref": "RootInstanceProfile"
+ },
+ "SecurityGroups": [
+ {
+ "Ref": "ESSecurityGroup"
+ }
+ ],
+
+ "BlockDeviceMappings": [
+ {
+ "DeviceName": "/dev/sdb",
+ "VirtualName": "ephemeral0"
+ },
+ {
+ "DeviceName": "/dev/sdc",
+ "VirtualName": "ephemeral1"
+ }
+ ]
+ }
+ },
+ "ESAutoScalingGroup": {
+ "Type": "AWS::AutoScaling::AutoScalingGroup",
+ "Version": "2009-05-15",
+ "Properties": {
+ "AvailabilityZones": [
+ {
+ "Fn::FindInMap": [
+ "FourAZs",
+ {
+ "Ref": "AWS::Region"
+ },
+ "AZ1"
+ ]
+ },
+ {
+ "Fn::FindInMap": [
+ "FourAZs",
+ {
+ "Ref": "AWS::Region"
+ },
+ "AZ2"
+ ]
+ },
+ {
+ "Fn::FindInMap": [
+ "FourAZs",
+ {
+ "Ref": "AWS::Region"
+ },
+ "AZ3"
+ ]
+ },
+ {
+ "Fn::FindInMap": [
+ "FourAZs",
+ {
+ "Ref": "AWS::Region"
+ },
+ "AZ4"
+ ]
+ }
+ ],
+ "LaunchConfigurationName": {
+ "Ref": "ESAutoScalingLaunchConfiguration"
+ },
+ "MinSize": {
+ "Ref": "ESNumServers"
+ },
+ "MaxSize": {
+ "Ref": "ESNumServers"
+ },
+ "NotificationConfiguration": {
+ "TopicARN": {
+ "Ref": "NotificationTopic"
+ },
+ "NotificationTypes": [
+ "autoscaling:EC2_INSTANCE_LAUNCH",
+ "autoscaling:EC2_INSTANCE_LAUNCH_ERROR",
+ "autoscaling:EC2_INSTANCE_TERMINATE",
+ "autoscaling:EC2_INSTANCE_TERMINATE_ERROR"
+ ]
+ }
+ }
+ },
+ "RestAutoScalingLaunchConfiguration":{
+ "Type":"AWS::AutoScaling::LaunchConfiguration",
+ "Properties":{
+ "UserData":{
+ "Fn::Base64":{
+ "Fn::Join":[
+ "",
+ [
+ "#!/bin/bash -ex\n",
+ "# REST SERVER STARTUP \n",
+ "exec >/var/log/usergrid-bootstrap.log 2>&1\n",
+ "\n",
+ "mkdir -p /usr/share/usergrid\n",
+ "\n",
+ "# create script that sets our environment variables\n",
+ "cat >/etc/profile.d/usergrid-env.sh <<EOF\n",
+ "alias sudo='sudo -E'\n",
+ "\n",
+ "export TYPE=rest\n",
+ "export STACK_NAME=", { "Ref":"AWS::StackName" }, "\n",
+ "export YOURKIT=", { "Ref":"InstallYourkit" }, "\n",
+ "export DNS_NAME=", { "Ref":"DnsSubDomain" }, "\n",
+ "export DNS_DOMAIN=", { "Ref":"DnsDomain" }, "\n",
+ "export PUBLIC_HOSTNAME=`(curl -s http://169.254.169.254/latest/meta-data/public-hostname)`\n",
+ "export INTERNAL_HOSTNAME=`(curl http://169.254.169.254/latest/meta-data/local-ipv4)`\n",
+ "export ELB_NAME=", { "Ref":"RestElasticLoadBalancer" }, "\n",
+ "\n",
+ "export EC2_INSTANCE_ID=`ec2metadata --instance-id`\n",
+ "export EC2_REGION=", { "Ref":"AWS::Region" }, "\n",
+ "export EC2_URL=https://ec2.amazonaws.com/\n", "\n",
+ "export REST_SECURITY_GROUP_NAME=", { "Ref":"RestSecurityGroup" }, "\n",
+ "export DB_SECURITY_GROUP_NAME=", { "Ref":"CassSecurityGroup" }, "\n",
+ "\n",
+ "export CASSANDRA_CLUSTER_NAME=", { "Ref":"CassClusterName" }, "\n",
+ "export CASSANDRA_KEYSPACE_NAME=usergrid", "\n",
+ "export CASSANDRA_NUM_SERVERS=", { "Ref":"CassNumServers" }, "\n",
+ "export GRAPHITE_NUM_SERVERS=", { "Ref":"GraphiteNumServers" }, "\n",
+ "export TOMCAT_NUM_SERVERS=", { "Ref":"RestMinServers" }, "\n",
+ "\n",
+ "export CASSANDRA_REPLICATION_FACTOR=", { "Ref":"CassReplicationFactor" }, "\n",
+ "\n",
+ "export CASSANDRA_READ_CONSISTENCY=", { "Ref":"CassReadConsistency" }, "\n",
+ "\n",
+ "export CASSANDRA_WRITE_CONSISTENCY=", { "Ref":"CassWriteConsistency" }, "\n",
+ "\n",
+
+ "export ES_CLUSTER_NAME=", { "Ref":"ESClusterName" }, "\n",
+ "export ES_NUM_SERVERS=", { "Ref":"ESNumServers" }, "\n",
+ "\n",
+ "export RELEASE_BUCKET=", { "Ref":"ReleaseBucket" }, "\n",
+ "\n",
+ "export NUM_THREAD_PROC=", { "Ref":"TomcatThreadsPerCore" }, "\n",
+ "\n",
+ "export SUPER_USER_EMAIL=", { "Ref":"SuperUserEmail" }, "\n",
+ "export TEST_ADMIN_USER_EMAIL=", { "Ref":"TestAdminUserEmail" }, "\n",
+ "\n",
+ "EOF\n",
+ "\n",
+ "# put AWS creds in environment\n",
+ "cat >/etc/profile.d/aws-credentials.sh <<EOF\n",
+ "export AWS_ACCESS_KEY=", { "Ref":"RestKey" }, "\n",
+ "export AWS_SECRET_KEY=", { "Fn::GetAtt":[ "RestKey", "SecretAccessKey" ] }, "\n",
+ "EOF\n",
+ "\n",
+ "# setup s3cmd (will be installed by init script) \n",
+ "cat >/etc/s3cfg <<EOF\n",
+ "access_key=", { "Ref":"RestKey" }, "\n",
+ "secret_key=", { "Fn::GetAtt":[ "RestKey", "SecretAccessKey" ] }, "\n",
+ "EOF\n",
+ "chmod 644 /etc/s3cfg\n",
+ "ln -s /etc/s3cfg ~ubuntu/.s3cfg\n",
+ "ln -s /etc/s3cfg ~root/.s3cfg\n",
+ "\n",
+ "# download usergrid and init script bundle from S3\n",
+ "wget -O- -q http://s3tools.org/repo/deb-all/stable/s3tools.key | apt-key add -\n",
+ "wget -O/etc/apt/sources.list.d/s3tools.list http://s3tools.org/repo/deb-all/stable/s3tools.list\n",
+ "apt-get update\n",
+ "apt-get -y install s3cmd\n",
+ "cd /usr/share/usergrid\n",
+ "s3cmd --config=/etc/s3cfg get s3://", {"Ref": "ReleaseBucket"}, "/awscluster-1.0-SNAPSHOT-any.tar.gz\n",
+ "s3cmd --config=/etc/s3cfg get s3://", {"Ref": "ReleaseBucket"}, "/ROOT.war\n",
+ "tar xvf awscluster-1.0-SNAPSHOT-any.tar.gz\n",
+ "rm -fr awscluster-1.0-SNAPSHOT-any.tar.gz\n",
+ "mv ROOT.war webapps/ROOT.war\n",
+ "chmod 755 ./init_instance/*.sh\n",
+ "cd ./init_instance\n",
+ "# Init as a REST intance \n",
+ "sh ./init_rest_server.sh\n"
+ ]
+ ]
+ }
+ },
+ "KeyName":{
+ "Ref":"KeyPair"
+ },
+ "ImageId":{
+ "Fn::FindInMap":[
+ "AWSRegionArch2AMI",
+ {
+ "Ref":"AWS::Region"
+ },
+ {
+ "Fn::FindInMap":[
+ "AWSInstanceType2Arch",
+ {
+ "Ref":"RestInstanceType"
+ },
+ "Arch"
+ ]
+ }
+ ]
+ },
+ "InstanceType":{
+ "Ref":"RestInstanceType"
+ },
+ "IamInstanceProfile":{
+ "Ref":"RootInstanceProfile"
+ },
+ "SecurityGroups":[
+ {
+ "Ref":"RestSecurityGroup"
+ }
+ ]
+ }
+ },
"RestAutoScalingGroup": {
"Type": "AWS::AutoScaling::AutoScalingGroup",
"Version": "2009-05-15",
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e58597be/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityDeletedHandler.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityDeletedHandler.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityDeletedHandler.java
index ad4ed32..33cc988 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityDeletedHandler.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityDeletedHandler.java
@@ -63,6 +63,6 @@ public class EntityDeletedHandler implements EntityDeleted {
CpEntityManagerFactory cpemf = (CpEntityManagerFactory)emf;
final EntityIndex ei = cpemf.getManagerCache().getEntityIndex(scope);
- ei.deleteAllVersionsOfEntity( entityId );
+// ei.deleteAllVersionsOfEntity( entityId );
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e58597be/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionCreatedHandler.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionCreatedHandler.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionCreatedHandler.java
index 3b7163c..590df61 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionCreatedHandler.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/events/EntityVersionCreatedHandler.java
@@ -64,6 +64,6 @@ public class EntityVersionCreatedHandler implements EntityVersionCreated {
CpEntityManagerFactory cpemf = (CpEntityManagerFactory)emf;
final EntityIndex ei = cpemf.getManagerCache().getEntityIndex(scope);
- ei.deletePreviousVersions( entity.getId(), entity.getVersion() );
+// ei.deletePreviousVersions( entity.getId(), entity.getVersion() );
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e58597be/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
index 3edde4d..aa2bf8e 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/EntityIndex.java
@@ -74,19 +74,19 @@ public interface EntityIndex {
*/
public CandidateResults getEntityVersions(final IndexScope indexScope, final Id id);
- /**
- * Create a delete method that deletes by Id. This will delete all documents from ES with the same entity Id,
- * effectively removing all versions of an entity from all index scopes
- * @param entityId The entityId to remove
- */
- public Future deleteAllVersionsOfEntity(final Id entityId );
-
- /**
- * Takes all the previous versions of the current entity and deletes all previous versions
- * @param id The id to remove
- * @param version The max version to retain
- */
- public Future deletePreviousVersions(final Id id, final UUID version);
+// /**
+// * Create a delete method that deletes by Id. This will delete all documents from ES with the same entity Id,
+// * effectively removing all versions of an entity from all index scopes
+// * @param entityId The entityId to remove
+// */
+// public Future deleteAllVersionsOfEntity(final Id entityId );
+//
+// /**
+// * Takes all the previous versions of the current entity and deletes all previous versions
+// * @param id The id to remove
+// * @param version The max version to retain
+// */
+// public Future deletePreviousVersions(final Id id, final UUID version);
/**
* Refresh the index.
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e58597be/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
index 99d2b84..99643da 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java
@@ -662,89 +662,89 @@ public class EsEntityIndexImpl implements AliasedEntityIndex {
}
- @Override
- public ListenableActionFuture deleteAllVersionsOfEntity(final Id entityId ) {
-
- String idString = IndexingUtils.idString(entityId).toLowerCase();
-
- final TermQueryBuilder tqb = QueryBuilders.termQuery(ENTITYID_ID_FIELDNAME, idString);
-
- //Added For Graphite Metrics
- final Timer.Context timeDeleteAllVersions =allVersionsTimer.time();
- final Timer.Context timeDeleteAllVersionsFuture = allVersionsTimerFuture.time();
- final ListenableActionFuture<DeleteByQueryResponse> response = esProvider.getClient()
- .prepareDeleteByQuery( alias.getWriteAlias() ).setQuery( tqb ).execute();
-
- response.addListener( new ActionListener<DeleteByQueryResponse>() {
-
- @Override
- public void onResponse( DeleteByQueryResponse response) {
- timeDeleteAllVersions.stop();
- logger
- .debug( "Deleted entity {}:{} from all index scopes with response status = {}", entityId.getType(),
- entityId.getUuid(), response.status().toString() );
-
- checkDeleteByQueryResponse(tqb, response);
- }
-
-
- @Override
- public void onFailure( Throwable e ) {
- timeDeleteAllVersions.stop();
- logger.error( "Deleted entity {}:{} from all index scopes with error {}", entityId.getType(),
- entityId.getUuid(), e);
-
-
- }
- });
- timeDeleteAllVersionsFuture.stop();
- return response;
- }
-
-
- @Override
- public ListenableActionFuture deletePreviousVersions(final Id entityId, final UUID version) {
-
- String idString = IndexingUtils.idString( entityId ).toLowerCase();
-
- final FilteredQueryBuilder fqb = QueryBuilders.filteredQuery(
- QueryBuilders.termQuery(ENTITYID_ID_FIELDNAME, idString),
- FilterBuilders.rangeFilter(ENTITY_VERSION_FIELDNAME).lt(version.timestamp())
- );
-
- //Added For Graphite Metrics
- //Checks the time from the execute to the response below
- final Timer.Context timeDeletePreviousVersions = deletePreviousTimer.time();
- final Timer.Context timeDeletePreviousVersionFuture = deletePreviousTimerFuture.time();
- final ListenableActionFuture<DeleteByQueryResponse> response = esProvider.getClient()
- .prepareDeleteByQuery(alias.getWriteAlias()).setQuery(fqb).execute();
-
- //Added For Graphite Metrics
- response.addListener(new ActionListener<DeleteByQueryResponse>() {
- @Override
- public void onResponse(DeleteByQueryResponse response) {
- timeDeletePreviousVersions.stop();
- //error message needs to be retooled so that it describes the entity more throughly
- logger
- .debug("Deleted entity {}:{} with version {} from all " + "index scopes with response status = {}",
- entityId.getType(), entityId.getUuid(), version, response.status().toString());
-
- checkDeleteByQueryResponse( fqb, response );
- }
-
-
- @Override
- public void onFailure( Throwable e ) {
- timeDeletePreviousVersions.stop();
- logger.error( "Deleted entity {}:{} from all index scopes with error {}", entityId.getType(),
- entityId.getUuid(), e );
- }
- } );
-
- timeDeletePreviousVersionFuture.stop();
-
- return response;
- }
+// @Override
+// public ListenableActionFuture deleteAllVersionsOfEntity(final Id entityId ) {
+// String idString = IndexingUtils.idString(entityId).toLowerCase();
+//
+// final TermQueryBuilder tqb = QueryBuilders.termQuery(ENTITYID_ID_FIELDNAME, idString);
+//
+// //Added For Graphite Metrics
+// final Timer.Context timeDeleteAllVersions =allVersionsTimer.time();
+// final Timer.Context timeDeleteAllVersionsFuture = allVersionsTimerFuture.time();
+//
+// final ListenableActionFuture<DeleteByQueryResponse> response = esProvider.getClient()
+// .prepareDeleteByQuery( alias.getWriteAlias() ).setQuery( tqb ).execute();
+//
+// response.addListener( new ActionListener<DeleteByQueryResponse>() {
+//
+// @Override
+// public void onResponse( DeleteByQueryResponse response) {
+// timeDeleteAllVersions.stop();
+// logger
+// .debug( "Deleted entity {}:{} from all index scopes with response status = {}", entityId.getType(),
+// entityId.getUuid(), response.status().toString() );
+//
+// checkDeleteByQueryResponse(tqb, response);
+// }
+//
+//
+// @Override
+// public void onFailure( Throwable e ) {
+// timeDeleteAllVersions.stop();
+// logger.error( "Deleted entity {}:{} from all index scopes with error {}", entityId.getType(),
+// entityId.getUuid(), e);
+//
+//
+// }
+// });
+// timeDeleteAllVersionsFuture.stop();
+// return response;
+// }
+//
+//
+// @Override
+// public ListenableActionFuture deletePreviousVersions(final Id entityId, final UUID version) {
+//
+// String idString = IndexingUtils.idString( entityId ).toLowerCase();
+//
+// final FilteredQueryBuilder fqb = QueryBuilders.filteredQuery(
+// QueryBuilders.termQuery(ENTITYID_ID_FIELDNAME, idString),
+// FilterBuilders.rangeFilter(ENTITY_VERSION_FIELDNAME).lt(version.timestamp())
+// );
+//
+// //Added For Graphite Metrics
+// //Checks the time from the execute to the response below
+// final Timer.Context timeDeletePreviousVersions = deletePreviousTimer.time();
+// final Timer.Context timeDeletePreviousVersionFuture = deletePreviousTimerFuture.time();
+// final ListenableActionFuture<DeleteByQueryResponse> response = esProvider.getClient()
+// .prepareDeleteByQuery(alias.getWriteAlias()).setQuery(fqb).execute();
+//
+// //Added For Graphite Metrics
+// response.addListener(new ActionListener<DeleteByQueryResponse>() {
+// @Override
+// public void onResponse(DeleteByQueryResponse response) {
+// timeDeletePreviousVersions.stop();
+// //error message needs to be retooled so that it describes the entity more throughly
+// logger
+// .debug("Deleted entity {}:{} with version {} from all " + "index scopes with response status = {}",
+// entityId.getType(), entityId.getUuid(), version, response.status().toString());
+//
+// checkDeleteByQueryResponse( fqb, response );
+// }
+//
+//
+// @Override
+// public void onFailure( Throwable e ) {
+// timeDeletePreviousVersions.stop();
+// logger.error( "Deleted entity {}:{} from all index scopes with error {}", entityId.getType(),
+// entityId.getUuid(), e );
+// }
+// } );
+//
+// timeDeletePreviousVersionFuture.stop();
+//
+// return response;
+// }
/**
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e58597be/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
index 3fc3e77..2762c18 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
@@ -72,8 +72,6 @@ public class EsIndexBufferConsumerImpl implements IndexBufferConsumer {
//the actively running subscription
private Subscription subscription;
- private Observable<List<IndexOperationMessage>> consumer;
-
private Object mutex = new Object();
@Inject
@@ -203,6 +201,9 @@ public class EsIndexBufferConsumerImpl implements IndexBufferConsumer {
final Observable<IndexRequest> index = Observable.from( indexOperationMessage.getIndexRequests() );
final Observable<DeIndexRequest> deIndex = Observable.from( indexOperationMessage.getDeIndexRequests() );
+ indexSizeCounter.inc(indexOperationMessage.getDeIndexRequests().size());
+ indexSizeCounter.inc(indexOperationMessage.getIndexRequests().size());
+
return Observable.merge( index, deIndex );
}
} )