From 54fe8e43f8a337afc396d03c4d4277fa0fd23f37 Mon Sep 17 00:00:00 2001 From: pengys5 <8082209@qq.com> Date: Thu, 27 Jul 2017 09:45:45 +0800 Subject: [PATCH] Instance register, heartBeat, recover finish --- .../agentregister/instance/IInstanceDAO.java | 8 -- .../instance/InstanceIDService.java | 11 ++- .../worker/register/IdAutoIncrement.java | 24 +++++ .../ApplicationRegisterSerialWorker.java | 18 +--- .../application/dao/ApplicationEsDAO.java | 39 +-------- .../register/instance/InstanceDataDefine.java | 30 +++++-- .../InstanceRegisterRemoteWorker.java | 4 +- .../InstanceRegisterSerialWorker.java | 87 +++++++++++++++++++ .../register/instance/dao/IInstanceDAO.java | 18 ++++ .../register/instance/dao/InstanceEsDAO.java | 84 ++++++++++++++++++ .../register/instance/dao/InstanceH2DAO.java | 25 ++++++ .../resources/META-INF/defines/es_dao.define | 3 +- .../resources/META-INF/defines/h2_dao.define | 3 +- .../local_async_worker_provider.define | 4 +- .../elasticsearch/ElasticSearchClient.java | 10 +++ .../storage/elasticsearch/dao/EsDAO.java | 45 ++++++++++ 16 files changed, 340 insertions(+), 73 deletions(-) delete mode 100644 apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/IInstanceDAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/IdAutoIncrement.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceRegisterSerialWorker.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/IInstanceDAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/InstanceEsDAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/InstanceH2DAO.java diff --git a/apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/IInstanceDAO.java b/apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/IInstanceDAO.java deleted file mode 100644 index 7cdb17369..000000000 --- a/apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/IInstanceDAO.java +++ /dev/null @@ -1,8 +0,0 @@ -package org.skywalking.apm.collector.agentregister.instance; - -/** - * @author pengys5 - */ -public interface IInstanceDAO { - int getInstanceId(String agentUUID); -} diff --git a/apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/InstanceIDService.java b/apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/InstanceIDService.java index 9a74f3118..4ac42d8bd 100644 --- a/apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/InstanceIDService.java +++ b/apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/InstanceIDService.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentregister.instance; import org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationRegisterRemoteWorker; import org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceDataDefine; +import org.skywalking.apm.collector.agentstream.worker.register.instance.dao.IInstanceDAO; import org.skywalking.apm.collector.core.framework.CollectorContextHelper; import org.skywalking.apm.collector.storage.dao.DAOContainer; import org.skywalking.apm.collector.stream.StreamModuleContext; @@ -20,11 +21,11 @@ public class InstanceIDService { public int getOrCreate(int applicationId, String agentUUID, long registerTime) { IInstanceDAO dao = (IInstanceDAO)DAOContainer.INSTANCE.get(IInstanceDAO.class.getName()); - int instanceId = dao.getInstanceId(agentUUID); + int instanceId = dao.getInstanceId(applicationId, agentUUID); if (instanceId == 0) { StreamModuleContext context = (StreamModuleContext)CollectorContextHelper.INSTANCE.getContext(StreamModuleGroupDefine.GROUP_NAME); - InstanceDataDefine.Instance instance = new InstanceDataDefine.Instance(agentUUID, applicationId, agentUUID, registerTime, 0); + InstanceDataDefine.Instance instance = new InstanceDataDefine.Instance("0", applicationId, agentUUID, registerTime, 0); try { context.getClusterWorkerContext().lookup(ApplicationRegisterRemoteWorker.WorkerRole.INSTANCE).tell(instance); } catch (WorkerNotFoundException | WorkerInvokeException e) { @@ -35,10 +36,14 @@ public class InstanceIDService { } public void heartBeat(int instanceId, long heartbeatTime) { - + IInstanceDAO dao = (IInstanceDAO)DAOContainer.INSTANCE.get(IInstanceDAO.class.getName()); + dao.updateHeartbeatTime(instanceId, heartbeatTime); } public void recover(int instanceId, int applicationId, long registerTime) { + IInstanceDAO dao = (IInstanceDAO)DAOContainer.INSTANCE.get(IInstanceDAO.class.getName()); + InstanceDataDefine.Instance instance = new InstanceDataDefine.Instance(String.valueOf(instanceId), applicationId, "", registerTime, instanceId); + dao.save(instance); } } diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/IdAutoIncrement.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/IdAutoIncrement.java new file mode 100644 index 000000000..d9864e4e7 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/IdAutoIncrement.java @@ -0,0 +1,24 @@ +package org.skywalking.apm.collector.agentstream.worker.register; + +/** + * @author pengys5 + */ +public enum IdAutoIncrement { + INSTANCE; + + public int increment(int min, int max) { + int instanceId; + if (min == max) { + instanceId = -1; + } else if (min + max == 0) { + instanceId = max + 1; + } else if (min + max > 0) { + instanceId = min - 1; + } else if (max < 0) { + instanceId = 1; + } else { + instanceId = max + 1; + } + return instanceId; + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/application/ApplicationRegisterSerialWorker.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/application/ApplicationRegisterSerialWorker.java index ff01be792..66c6c0e2a 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/application/ApplicationRegisterSerialWorker.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/application/ApplicationRegisterSerialWorker.java @@ -1,5 +1,6 @@ package org.skywalking.apm.collector.agentstream.worker.register.application; +import org.skywalking.apm.collector.agentstream.worker.register.IdAutoIncrement; import org.skywalking.apm.collector.agentstream.worker.register.application.dao.IApplicationDAO; import org.skywalking.apm.collector.storage.dao.DAOContainer; import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorker; @@ -41,20 +42,9 @@ public class ApplicationRegisterSerialWorker extends AbstractLocalAsyncWorker { application.setId("1"); } else { int max = dao.getMaxApplicationId(); - int instanceId; - if (min == max) { - instanceId = -1; - } else if (min + max == 0) { - instanceId = max + 1; - } else if (min + max > 0) { - instanceId = min - 1; - } else if (max < 0) { - instanceId = 1; - } else { - instanceId = max + 1; - } - application.setApplicationId(instanceId); - application.setId(String.valueOf(instanceId)); + int applicationId = IdAutoIncrement.INSTANCE.increment(min, max); + application.setApplicationId(applicationId); + application.setId(String.valueOf(applicationId)); } dao.save(application); } diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/application/dao/ApplicationEsDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/application/dao/ApplicationEsDAO.java index 7f12f50c3..ebe4360a8 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/application/dao/ApplicationEsDAO.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/application/dao/ApplicationEsDAO.java @@ -9,11 +9,6 @@ import org.elasticsearch.action.search.SearchType; import org.elasticsearch.action.support.WriteRequest; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; -import org.elasticsearch.search.aggregations.AggregationBuilders; -import org.elasticsearch.search.aggregations.metrics.max.Max; -import org.elasticsearch.search.aggregations.metrics.max.MaxAggregationBuilder; -import org.elasticsearch.search.aggregations.metrics.min.Min; -import org.elasticsearch.search.aggregations.metrics.min.MinAggregationBuilder; import org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationDataDefine; import org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationTable; import org.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; @@ -47,41 +42,11 @@ public class ApplicationEsDAO extends EsDAO implements IApplicationDAO { } @Override public int getMaxApplicationId() { - ElasticSearchClient client = getClient(); - SearchRequestBuilder searchRequestBuilder = client.prepareSearch(ApplicationTable.TABLE); - searchRequestBuilder.setTypes("type"); - searchRequestBuilder.setSize(0); - MaxAggregationBuilder aggregation = AggregationBuilders.max("agg").field(ApplicationTable.COLUMN_APPLICATION_ID); - searchRequestBuilder.addAggregation(aggregation); - - SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); - Max agg = searchResponse.getAggregations().get("agg"); - - int id = (int)agg.getValue(); - if (id == Integer.MAX_VALUE || id == Integer.MIN_VALUE) { - return 0; - } else { - return id; - } + return getMaxId(ApplicationTable.TABLE, ApplicationTable.COLUMN_APPLICATION_ID); } @Override public int getMinApplicationId() { - ElasticSearchClient client = getClient(); - SearchRequestBuilder searchRequestBuilder = client.prepareSearch(ApplicationTable.TABLE); - searchRequestBuilder.setTypes("type"); - searchRequestBuilder.setSize(0); - MinAggregationBuilder aggregation = AggregationBuilders.min("agg").field(ApplicationTable.COLUMN_APPLICATION_ID); - searchRequestBuilder.addAggregation(aggregation); - - SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); - Min agg = searchResponse.getAggregations().get("agg"); - - int id = (int)agg.getValue(); - if (id == Integer.MAX_VALUE || id == Integer.MIN_VALUE) { - return 0; - } else { - return id; - } + return getMinId(ApplicationTable.TABLE, ApplicationTable.COLUMN_APPLICATION_ID); } @Override public void save(ApplicationDataDefine.Application application) { diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceDataDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceDataDefine.java index 446a47f8e..6470d4520 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceDataDefine.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceDataDefine.java @@ -51,11 +51,11 @@ public class InstanceDataDefine extends DataDefine { } public static class Instance { - private final String id; - private final int applicationId; - private final String agentUUID; - private final long registerTime; - private final int instanceId; + private String id; + private int applicationId; + private String agentUUID; + private long registerTime; + private int instanceId; public Instance(String id, int applicationId, String agentUUID, long registerTime, int instanceId) { this.id = id; @@ -84,5 +84,25 @@ public class InstanceDataDefine extends DataDefine { public int getInstanceId() { return instanceId; } + + public void setId(String id) { + this.id = id; + } + + public void setApplicationId(int applicationId) { + this.applicationId = applicationId; + } + + public void setAgentUUID(String agentUUID) { + this.agentUUID = agentUUID; + } + + public void setRegisterTime(long registerTime) { + this.registerTime = registerTime; + } + + public void setInstanceId(int instanceId) { + this.instanceId = instanceId; + } } } diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceRegisterRemoteWorker.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceRegisterRemoteWorker.java index 3e7cd93ec..c6a85dd3e 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceRegisterRemoteWorker.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceRegisterRemoteWorker.java @@ -1,6 +1,5 @@ package org.skywalking.apm.collector.agentstream.worker.register.instance; -import org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationDataDefine; import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorker; import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorkerProvider; import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext; @@ -30,6 +29,7 @@ public class InstanceRegisterRemoteWorker extends AbstractRemoteWorker { @Override protected void onWork(Object message) throws WorkerException { InstanceDataDefine.Instance instance = (InstanceDataDefine.Instance)message; logger.debug("application id: {}, agentUUID: {}, register time: {}", instance.getApplicationId(), instance.getAgentUUID(), instance.getRegisterTime()); + getClusterContext().lookup(InstanceRegisterSerialWorker.WorkerRole.INSTANCE).tell(instance); } public static class Factory extends AbstractRemoteWorkerProvider { @@ -58,7 +58,7 @@ public class InstanceRegisterRemoteWorker extends AbstractRemoteWorker { } @Override public DataDefine dataDefine() { - return new ApplicationDataDefine(); + return new InstanceDataDefine(); } } } diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceRegisterSerialWorker.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceRegisterSerialWorker.java new file mode 100644 index 000000000..9bfabe44e --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/InstanceRegisterSerialWorker.java @@ -0,0 +1,87 @@ +package org.skywalking.apm.collector.agentstream.worker.register.instance; + +import org.skywalking.apm.collector.agentstream.worker.register.IdAutoIncrement; +import org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationDataDefine; +import org.skywalking.apm.collector.agentstream.worker.register.instance.dao.IInstanceDAO; +import org.skywalking.apm.collector.storage.dao.DAOContainer; +import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorker; +import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorkerProvider; +import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext; +import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException; +import org.skywalking.apm.collector.stream.worker.Role; +import org.skywalking.apm.collector.stream.worker.WorkerException; +import org.skywalking.apm.collector.stream.worker.impl.data.DataDefine; +import org.skywalking.apm.collector.stream.worker.selector.ForeverFirstSelector; +import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class InstanceRegisterSerialWorker extends AbstractLocalAsyncWorker { + + private final Logger logger = LoggerFactory.getLogger(InstanceRegisterSerialWorker.class); + + public InstanceRegisterSerialWorker(Role role, ClusterWorkerContext clusterContext) { + super(role, clusterContext); + } + + @Override public void preStart() throws ProviderNotFoundException { + super.preStart(); + } + + @Override protected void onWork(Object message) throws WorkerException { + if (message instanceof InstanceDataDefine.Instance) { + InstanceDataDefine.Instance instance = (InstanceDataDefine.Instance)message; + logger.debug("register instance, application id: {}, agentUUID: {}", instance.getApplicationId(), instance.getAgentUUID()); + + IInstanceDAO dao = (IInstanceDAO)DAOContainer.INSTANCE.get(IInstanceDAO.class.getName()); + int min = dao.getMinInstanceId(); + if (min == 0) { + instance.setId("1"); + instance.setInstanceId(1); + } else { + int max = dao.getMaxInstanceId(); + int instanceId = IdAutoIncrement.INSTANCE.increment(min, max); + instance.setId(String.valueOf(instanceId)); + instance.setInstanceId(instanceId); + } + dao.save(instance); + } + } + + public static class Factory extends AbstractLocalAsyncWorkerProvider { + @Override + public Role role() { + return WorkerRole.INSTANCE; + } + + @Override + public InstanceRegisterSerialWorker workerInstance(ClusterWorkerContext clusterContext) { + return new InstanceRegisterSerialWorker(role(), clusterContext); + } + + @Override public int queueSize() { + return 256; + } + } + + public enum WorkerRole implements Role { + INSTANCE; + + @Override + public String roleName() { + return InstanceRegisterSerialWorker.class.getSimpleName(); + } + + @Override + public WorkerSelector workerSelector() { + return new ForeverFirstSelector(); + } + + @Override public DataDefine dataDefine() { + return new ApplicationDataDefine(); + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/IInstanceDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/IInstanceDAO.java new file mode 100644 index 000000000..f0c311146 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/IInstanceDAO.java @@ -0,0 +1,18 @@ +package org.skywalking.apm.collector.agentstream.worker.register.instance.dao; + +import org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceDataDefine; + +/** + * @author pengys5 + */ +public interface IInstanceDAO { + int getInstanceId(int applicationId, String agentUUID); + + int getMaxInstanceId(); + + int getMinInstanceId(); + + void save(InstanceDataDefine.Instance instance); + + void updateHeartbeatTime(int instanceId, long heartbeatTime); +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/InstanceEsDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/InstanceEsDAO.java new file mode 100644 index 000000000..237f50009 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/InstanceEsDAO.java @@ -0,0 +1,84 @@ +package org.skywalking.apm.collector.agentstream.worker.register.instance.dao; + +import java.util.HashMap; +import java.util.Map; +import org.elasticsearch.action.index.IndexResponse; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.search.SearchType; +import org.elasticsearch.action.support.WriteRequest; +import org.elasticsearch.action.update.UpdateRequest; +import org.elasticsearch.index.query.BoolQueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; +import org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceDataDefine; +import org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceTable; +import org.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; +import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class InstanceEsDAO extends EsDAO implements IInstanceDAO { + + private final Logger logger = LoggerFactory.getLogger(InstanceEsDAO.class); + + @Override public int getInstanceId(int applicationId, String agentUUID) { + ElasticSearchClient client = getClient(); + + SearchRequestBuilder searchRequestBuilder = client.prepareSearch(InstanceTable.TABLE); + searchRequestBuilder.setTypes("type"); + searchRequestBuilder.setSearchType(SearchType.QUERY_THEN_FETCH); + BoolQueryBuilder builder = QueryBuilders.boolQuery(); + builder.must().add(QueryBuilders.termQuery(InstanceTable.COLUMN_APPLICATION_ID, applicationId)); + builder.must().add(QueryBuilders.termQuery(InstanceTable.COLUMN_AGENTUUID, agentUUID)); + searchRequestBuilder.setQuery(builder); + searchRequestBuilder.setSize(1); + + SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); + if (searchResponse.getHits().totalHits > 0) { + SearchHit searchHit = searchResponse.getHits().iterator().next(); + int instanceId = (int)searchHit.getSource().get(InstanceTable.COLUMN_INSTANCE_ID); + return instanceId; + } + return 0; + } + + @Override public int getMaxInstanceId() { + return getMaxId(InstanceTable.TABLE, InstanceTable.COLUMN_INSTANCE_ID); + } + + @Override public int getMinInstanceId() { + return getMinId(InstanceTable.TABLE, InstanceTable.COLUMN_INSTANCE_ID); + } + + @Override public void save(InstanceDataDefine.Instance instance) { + logger.debug("save instance register info, application id: {}, agentUUID: {}", instance.getApplicationId(), instance.getAgentUUID()); + ElasticSearchClient client = getClient(); + Map source = new HashMap(); + source.put(InstanceTable.COLUMN_INSTANCE_ID, instance.getInstanceId()); + source.put(InstanceTable.COLUMN_APPLICATION_ID, instance.getApplicationId()); + source.put(InstanceTable.COLUMN_AGENTUUID, instance.getAgentUUID()); + source.put(InstanceTable.COLUMN_REGISTER_TIME, instance.getRegisterTime()); + + IndexResponse response = client.prepareIndex(InstanceTable.TABLE, instance.getId()).setSource(source).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get(); + logger.debug("save instance register info, application id: {}, agentUUID: {}, status: {}", instance.getApplicationId(), instance.getAgentUUID(), response.status().name()); + } + + @Override public void updateHeartbeatTime(int instanceId, long heartbeatTime) { + ElasticSearchClient client = getClient(); + UpdateRequest updateRequest = new UpdateRequest(); + updateRequest.index(InstanceTable.TABLE); + updateRequest.type("type"); + updateRequest.id(String.valueOf(instanceId)); + updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); + + Map source = new HashMap(); + source.put(InstanceTable.COLUMN_HEARTBEAT_TIME, heartbeatTime); + + updateRequest.doc(source); + client.update(updateRequest); + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/InstanceH2DAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/InstanceH2DAO.java new file mode 100644 index 000000000..d3bda6437 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/register/instance/dao/InstanceH2DAO.java @@ -0,0 +1,25 @@ +package org.skywalking.apm.collector.agentstream.worker.register.instance.dao; + +import org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceDataDefine; +import org.skywalking.apm.collector.storage.h2.dao.H2DAO; + +/** + * @author pengys5 + */ +public class InstanceH2DAO extends H2DAO implements IInstanceDAO { + @Override public int getInstanceId(int applicationId, String agentUUID) { + return 0; + } + + @Override public int getMaxInstanceId() { + return 0; + } + + @Override public int getMinInstanceId() { + return 0; + } + + @Override public void save(InstanceDataDefine.Instance instance) { + + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define index 2a346cdbf..c4c309747 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define @@ -1 +1,2 @@ -org.skywalking.apm.collector.agentstream.worker.register.application.dao.ApplicationEsDAO \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.register.application.dao.ApplicationEsDAO +org.skywalking.apm.collector.agentstream.worker.register.instance.dao.InstanceEsDAO \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define index c243fe509..65950d42f 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define @@ -1 +1,2 @@ -org.skywalking.apm.collector.agentstream.worker.register.application.dao.ApplicationH2DAO \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.register.application.dao.ApplicationH2DAO +org.skywalking.apm.collector.agentstream.worker.register.instance.dao.InstanceH2DAO \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_async_worker_provider.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_async_worker_provider.define index 418ecfce1..b272a20ce 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_async_worker_provider.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_async_worker_provider.define @@ -1,3 +1,3 @@ org.skywalking.apm.collector.agentstream.worker.node.component.NodeComponentAggWorker$Factory - -org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationRegisterSerialWorker$Factory \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationRegisterSerialWorker$Factory +org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceRegisterSerialWorker$Factory \ No newline at end of file diff --git a/apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/elasticsearch/ElasticSearchClient.java b/apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/elasticsearch/ElasticSearchClient.java index 303222e16..27218ce35 100644 --- a/apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/elasticsearch/ElasticSearchClient.java +++ b/apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/elasticsearch/ElasticSearchClient.java @@ -4,11 +4,13 @@ import java.net.InetAddress; import java.net.UnknownHostException; import java.util.LinkedList; import java.util.List; +import java.util.concurrent.ExecutionException; import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse; import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.client.IndicesAdminClient; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.InetSocketTransportAddress; @@ -109,4 +111,12 @@ public class ElasticSearchClient implements Client { public IndexRequestBuilder prepareIndex(String indexName, String id) { return client.prepareIndex(indexName, "type", id); } + + public void update(UpdateRequest updateRequest) { + try { + client.update(updateRequest).get(); + } catch (InterruptedException | ExecutionException e) { + logger.error(e.getMessage(), e); + } + } } diff --git a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/dao/EsDAO.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/dao/EsDAO.java index a1a15ac0f..f6eb18a85 100644 --- a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/dao/EsDAO.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/dao/EsDAO.java @@ -1,5 +1,12 @@ package org.skywalking.apm.collector.storage.elasticsearch.dao; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.search.aggregations.AggregationBuilders; +import org.elasticsearch.search.aggregations.metrics.max.Max; +import org.elasticsearch.search.aggregations.metrics.max.MaxAggregationBuilder; +import org.elasticsearch.search.aggregations.metrics.min.Min; +import org.elasticsearch.search.aggregations.metrics.min.MinAggregationBuilder; import org.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; import org.skywalking.apm.collector.storage.dao.DAO; @@ -7,4 +14,42 @@ import org.skywalking.apm.collector.storage.dao.DAO; * @author pengys5 */ public abstract class EsDAO extends DAO { + + public final int getMaxId(String indexName, String columnName) { + ElasticSearchClient client = getClient(); + SearchRequestBuilder searchRequestBuilder = client.prepareSearch(indexName); + searchRequestBuilder.setTypes("type"); + searchRequestBuilder.setSize(0); + MaxAggregationBuilder aggregation = AggregationBuilders.max("agg").field(columnName); + searchRequestBuilder.addAggregation(aggregation); + + SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); + Max agg = searchResponse.getAggregations().get("agg"); + + int id = (int)agg.getValue(); + if (id == Integer.MAX_VALUE || id == Integer.MIN_VALUE) { + return 0; + } else { + return id; + } + } + + public final int getMinId(String indexName, String columnName) { + ElasticSearchClient client = getClient(); + SearchRequestBuilder searchRequestBuilder = client.prepareSearch(indexName); + searchRequestBuilder.setTypes("type"); + searchRequestBuilder.setSize(0); + MinAggregationBuilder aggregation = AggregationBuilders.min("agg").field(columnName); + searchRequestBuilder.addAggregation(aggregation); + + SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); + Min agg = searchResponse.getAggregations().get("agg"); + + int id = (int)agg.getValue(); + if (id == Integer.MAX_VALUE || id == Integer.MIN_VALUE) { + return 0; + } else { + return id; + } + } }