diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractLocalSyncWorker.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractLocalSyncWorker.java index 77843adbe..09817d04f 100644 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractLocalSyncWorker.java +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractLocalSyncWorker.java @@ -7,10 +7,4 @@ public abstract class AbstractLocalSyncWorker extends AbstractLocalWorker { public AbstractLocalSyncWorker(Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) { super(role, clusterContext, selfContext); } - - @Override - final public void work(Object message) throws Exception { - } - - public abstract Object onWork(Object message) throws Exception; } diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/Promise.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/Promise.java deleted file mode 100644 index 7e5ff7adf..000000000 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/Promise.java +++ /dev/null @@ -1,18 +0,0 @@ -package com.a.eye.skywalking.collector.actor; - -/** - * @author pengys5 - */ -public class Promise { - private boolean isTold = false; - private Object value; - - protected void completed(Object value) { - this.value = value; - isTold = true; - } - - public boolean isTold() { - return isTold; - } -} diff --git a/skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/actor/TestLocalSyncWorker.java b/skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/actor/TestLocalSyncWorker.java index 5c496e10f..f6fa40e38 100644 --- a/skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/actor/TestLocalSyncWorker.java +++ b/skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/actor/TestLocalSyncWorker.java @@ -18,13 +18,12 @@ public class TestLocalSyncWorker extends AbstractLocalSyncWorker { } @Override - public Object onWork(Object message) throws Exception { + public void work(Object message) throws Exception { if (message.equals("TellLocalWorker")) { System.out.println("hello! "); } else { System.out.println("unhandled"); } - return "Hello"; } public static class Factory extends AbstractLocalSyncWorkerProvider { diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/CollectorBootStartUp.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/CollectorBootStartUp.java index c5b422892..45e682b4e 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/CollectorBootStartUp.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/CollectorBootStartUp.java @@ -4,7 +4,7 @@ import com.a.eye.skywalking.collector.cluster.ClusterConfig; import com.a.eye.skywalking.collector.cluster.ClusterConfigInitializer; import com.a.eye.skywalking.logging.LogManager; import com.a.eye.skywalking.logging.log4j2.Log4j2Resolver; -import com.a.eye.skywalking.collector.worker.httpserver.HttpServer; +import com.a.eye.skywalking.collector.worker.httpserver.WebServer; import com.typesafe.config.Config; import com.typesafe.config.ConfigFactory; @@ -32,7 +32,7 @@ public class CollectorBootStartUp { // ActorSystem system = ActorSystem.create(ClusterConfig.Cluster.appname, config); // WorkersCreator.INSTANCE.boot(system); - HttpServer.INSTANCE.boot(); + WebServer.INSTANCE.boot(); // EsClient.boot(); } } diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMain.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMain.java index d28522454..b2ab4ea5e 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMain.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMain.java @@ -38,7 +38,7 @@ public class ApplicationMain extends AbstractLocalSyncWorker { } @Override - public Object onWork(Object message) throws Exception { + public void work(Object message) throws Exception { if (message instanceof TraceSegmentReceiver.TraceSegmentTimeSlice) { logger.debug("begin translate TraceSegment Object to JsonObject"); TraceSegmentReceiver.TraceSegmentTimeSlice traceSegment = (TraceSegmentReceiver.TraceSegmentTimeSlice) message; @@ -50,7 +50,6 @@ public class ApplicationMain extends AbstractLocalSyncWorker { sendToResponseCostPersistence(traceSegment); sendToResponseSummaryPersistence(traceSegment); } - return null; } public static class Factory extends AbstractLocalSyncWorkerProvider { diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/ApplicationRefMain.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/ApplicationRefMain.java index fcc72ce44..062c93629 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/ApplicationRefMain.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/ApplicationRefMain.java @@ -26,7 +26,7 @@ public class ApplicationRefMain extends AbstractLocalSyncWorker { } @Override - public Object onWork(Object message) throws Exception { + public void work(Object message) throws Exception { TraceSegmentReceiver.TraceSegmentTimeSlice traceSegment = (TraceSegmentReceiver.TraceSegmentTimeSlice) message; TraceSegment segment = traceSegment.getTraceSegment(); @@ -40,7 +40,6 @@ public class ApplicationRefMain extends AbstractLocalSyncWorker { getSelfContext().lookup(DAGNodeRefAnalysis.Role.INSTANCE).tell(nodeRef); } } - return null; } public static class Factory extends AbstractLocalSyncWorkerProvider { diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/Controller.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/Controller.java index 56535702a..b7a307248 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/Controller.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/Controller.java @@ -16,8 +16,4 @@ public abstract class Controller { protected abstract String path(); protected abstract JsonElement execute(Map parms); - - protected void tell(Role role, Object message) throws Exception { -// targetMember.beTold(message); - } } \ No newline at end of file diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/HttpServer.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/WebServer.java similarity index 93% rename from skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/HttpServer.java rename to skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/WebServer.java index fd570763a..7b2f2d7c4 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/HttpServer.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/httpserver/WebServer.java @@ -11,10 +11,10 @@ import java.util.Map; /** * @author pengys5 */ -public enum HttpServer { +public enum WebServer { INSTANCE; - private Logger logger = LogManager.getFormatterLogger(HttpServer.class); + private Logger logger = LogManager.getFormatterLogger(WebServer.class); public void boot() throws Exception { NanoHttpServer server = new NanoHttpServer(7001); diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/AbstractIndex.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/AbstractIndex.java new file mode 100644 index 000000000..61e1f0aba --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/AbstractIndex.java @@ -0,0 +1,72 @@ +package com.a.eye.skywalking.collector.worker.storage.index; + +import com.a.eye.skywalking.collector.worker.storage.EsClient; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; +import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse; +import org.elasticsearch.client.IndicesAdminClient; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.common.xcontent.XContentFactory; +import org.elasticsearch.index.IndexNotFoundException; + +import java.io.IOException; + +/** + * @author pengys5 + */ +public abstract class AbstractIndex { + + private Logger logger = LogManager.getFormatterLogger(AbstractIndex.class); + + final public XContentBuilder createSettingBuilder() throws IOException { + XContentBuilder settingsBuilder = XContentFactory.jsonBuilder() + .startObject() + .field("index.number_of_shards", 2) + .field("index.number_of_replicas", 0) + .endObject(); + return settingsBuilder; + } + + public abstract XContentBuilder createMappingBuilder() throws IOException; + + final public boolean createIndex() { + // settings + String settingSource = ""; + + // mapping + XContentBuilder mappingBuilder = null; + try { + XContentBuilder settingsBuilder = createSettingBuilder(); + + settingSource = settingsBuilder.string(); + + mappingBuilder = createMappingBuilder(); + logger.info("mapping builder str: %s", mappingBuilder.string()); + } catch (Exception e) { + logger.error("create %s index type of %s mapping builder error", index(), type()); + } + Settings settings = Settings.builder().loadFromSource(settingSource).build(); + IndicesAdminClient client = EsClient.getClient().admin().indices(); + CreateIndexResponse response = client.prepareCreate(index()).setSettings(settings).addMapping(type(), mappingBuilder).get(); + logger.info("create %s index with type of %s finished, isAcknowledged: %s", index(), type(), response.isAcknowledged()); + return response.isAcknowledged(); + } + + final public boolean deleteIndex() { + IndicesAdminClient client = EsClient.getClient().admin().indices(); + try { + DeleteIndexResponse response = client.prepareDelete(index()).get(); + logger.info("delete %s index finished, isAcknowledged: %s", index(), response.isAcknowledged()); + return response.isAcknowledged(); + } catch (IndexNotFoundException e) { + logger.info("%s index not found", index()); + } + return false; + } + + public abstract String index(); + + public abstract String type(); +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/ApplicationIndexWithDagNodeType.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/ApplicationIndexWithDagNodeType.java new file mode 100644 index 000000000..e9e6b4820 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/ApplicationIndexWithDagNodeType.java @@ -0,0 +1,55 @@ +package com.a.eye.skywalking.collector.worker.storage.index; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.common.xcontent.XContentFactory; + +import java.io.IOException; + +/** + * @author pengys5 + */ +public class ApplicationIndexWithDagNodeType extends AbstractIndex { + + private Logger logger = LogManager.getFormatterLogger(ApplicationIndexWithDagNodeType.class); + + public static final String Index = "application"; + public static final String Type = "dag_node"; + + @Override + public String index() { + return Index; + } + + @Override + public String type() { + return Type; + } + + @Override + public XContentBuilder createMappingBuilder() throws IOException { + XContentBuilder mappingBuilder = XContentFactory.jsonBuilder() + .startObject() + .startObject("properties") + .startObject("code") + .field("type", "string") + .field("index", "not_analyzed") + .endObject() + .startObject("layer") + .field("type", "string") + .field("index", "not_analyzed") + .endObject() + .startObject("component") + .field("type", "string") + .field("index", "not_analyzed") + .endObject() + .startObject("timeSlice") + .field("type", "long") + .field("index", "not_analyzed") + .endObject() + .endObject() + .endObject(); + return mappingBuilder; + } +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/ApplicationIndexWithNodeInstType.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/ApplicationIndexWithNodeInstType.java new file mode 100644 index 000000000..a23252407 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/storage/index/ApplicationIndexWithNodeInstType.java @@ -0,0 +1,52 @@ +package com.a.eye.skywalking.collector.worker.storage.index; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.common.xcontent.XContentFactory; + +import java.io.IOException; + +/** + * @author pengys5 + */ +public class ApplicationIndexWithNodeInstType extends AbstractIndex { + + private static Logger logger = LogManager.getFormatterLogger(ApplicationIndexWithNodeInstType.class); + + public static final String Index = "application"; + public static final String Type = "node_instance"; + + @Override + public String index() { + return Index; + } + + @Override + public String type() { + return Type; + } + + @Override + public XContentBuilder createMappingBuilder() throws IOException { + XContentBuilder mappingBuilder = XContentFactory.jsonBuilder() + .startObject() + .startObject("properties") + .startObject("code") + .field("type", "string") + .field("fielddata", true) + .field("index", "not_analyzed") + .endObject() + .startObject("address") + .field("type", "string") + .field("index", "not_analyzed") + .endObject() + .startObject("timeSlice") + .field("type", "long_range") + .field("index", "not_analyzed") + .endObject() + .endObject() + .endObject(); + return mappingBuilder; + } +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/controller/tracedag/TraceDagLoadController.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/controller/tracedag/TraceDagLoadController.java new file mode 100644 index 000000000..5579e0fb4 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/controller/tracedag/TraceDagLoadController.java @@ -0,0 +1,31 @@ +package com.a.eye.skywalking.collector.worker.web.controller.tracedag; + +import com.a.eye.skywalking.collector.worker.httpserver.Controller; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import fi.iki.elonen.NanoHTTPD; + +import java.util.Map; + +/** + * @author pengys5 + */ +public class TraceDagLoadController extends Controller { + + @Override + protected NanoHTTPD.Method httpMethod() { + return NanoHTTPD.Method.GET; + } + + @Override + protected String path() { + return "/traceDagLoad"; + } + + @Override + public JsonElement execute(Map parms) { + JsonObject jsonObject = new JsonObject(); + jsonObject.addProperty("test", "aaaa"); + return jsonObject; + } +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/controller/tracedag/TraceDagUpdateController.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/controller/tracedag/TraceDagUpdateController.java new file mode 100644 index 000000000..ca6a35e79 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/controller/tracedag/TraceDagUpdateController.java @@ -0,0 +1,7 @@ +package com.a.eye.skywalking.collector.worker.web.controller.tracedag; + +/** + * @author pengys5 + */ +public class TraceDagUpdateController { +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/persistence/ApplicationPersistence.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/persistence/ApplicationPersistence.java new file mode 100644 index 000000000..ae92bdce1 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/persistence/ApplicationPersistence.java @@ -0,0 +1,54 @@ +package com.a.eye.skywalking.collector.worker.web.persistence; + +import com.a.eye.skywalking.collector.worker.storage.EsClient; +import com.a.eye.skywalking.collector.worker.storage.index.ApplicationIndexWithDagNodeType; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.core.config.builder.api.FilterComponentBuilder; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.search.SearchType; +import org.elasticsearch.common.xcontent.XContentBuilder; +import org.elasticsearch.common.xcontent.XContentFactory; +import org.elasticsearch.index.query.ConstantScoreQueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.aggregations.Aggregation; +import org.elasticsearch.search.aggregations.AggregationBuilders; +import org.elasticsearch.search.aggregations.bucket.global.GlobalAggregationBuilder; +import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder; +import org.elasticsearch.search.builder.SearchSourceBuilder; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +/** + * @author pengys5 + */ +public class ApplicationPersistence { + + private Logger logger = LogManager.getFormatterLogger(ApplicationPersistence.class); + + public void searchDagNode(long startTimeSlice, long endTimeSlice){ + SearchRequestBuilder searchRequestBuilder = EsClient.getClient().prepareSearch(ApplicationIndexWithDagNodeType.Index); + searchRequestBuilder.setTypes(ApplicationIndexWithDagNodeType.Type); + searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH); + +// searchRequestBuilder.setQuery(QueryBuilders.rangeQuery("timeSlice").gte(startTimeSlice).lte(endTimeSlice)); + ConstantScoreQueryBuilder constantScoreQueryBuilder = QueryBuilders.constantScoreQuery(QueryBuilders.rangeQuery("timeSlice").gte(startTimeSlice).lte(endTimeSlice)); + searchRequestBuilder.setQuery(constantScoreQueryBuilder); + +//// GlobalAggregationBuilder aggregationBuilder = AggregationBuilders.global("agg").subAggregation(AggregationBuilders.terms("distinct_code").field("code")); +// TermsAggregationBuilder aggregationBuilder = AggregationBuilders.terms("distinct_code").field("code"); + searchRequestBuilder.addAggregation(AggregationBuilders.terms("distinct_code").field("code")); + SearchResponse response = searchRequestBuilder.execute().actionGet(); + List aggregationList = response.getAggregations().asList(); + logger.debug("dag node list size: %s", aggregationList.size()); + + for (Aggregation aggregation : aggregationList) { + for (Map.Entry entry : aggregation.getMetaData().entrySet()) { + logger.debug("code: %s, count: %s", entry.getKey(), entry.getValue()); + } + } + } +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/persistence/NodeInstancePersistence.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/persistence/NodeInstancePersistence.java new file mode 100644 index 000000000..6a5907a32 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/web/persistence/NodeInstancePersistence.java @@ -0,0 +1,7 @@ +package com.a.eye.skywalking.collector.worker.web.persistence; + +/** + * @author pengys5 + */ +public class NodeInstancePersistence { +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/resources/log4j2.xml b/skywalking-collector/skywalking-collector-worker/src/main/resources/log4j2.xml index e8fc5bbdb..ef61f4b3f 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/resources/log4j2.xml +++ b/skywalking-collector/skywalking-collector-worker/src/main/resources/log4j2.xml @@ -1,13 +1,19 @@ - - - - - - - - - - - + + + + + + + + + + + + + + + + + diff --git a/skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/web/persistence/ApplicationPersistenceTestCase.java b/skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/web/persistence/ApplicationPersistenceTestCase.java new file mode 100644 index 000000000..d986c81c3 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/web/persistence/ApplicationPersistenceTestCase.java @@ -0,0 +1,50 @@ +package com.a.eye.skywalking.collector.worker.web.persistence; + +import com.a.eye.skywalking.collector.worker.storage.EsClient; +import com.a.eye.skywalking.collector.worker.storage.index.ApplicationIndexWithDagNodeType; +import org.elasticsearch.action.index.IndexResponse; +import org.elasticsearch.rest.RestStatus; +import org.junit.Before; +import org.junit.Test; + +import java.net.UnknownHostException; +import java.util.HashMap; +import java.util.Map; + +/** + * @author pengys5 + */ +public class ApplicationPersistenceTestCase { + + @Before + public void initIndex() throws UnknownHostException { + EsClient.boot(); + ApplicationIndexWithDagNodeType index = new ApplicationIndexWithDagNodeType(); + index.deleteIndex(); + index.createIndex(); + insertData(); + } + + @Test + public void testSearchDagNode() { + ApplicationPersistence persistence = new ApplicationPersistence(); + persistence.searchDagNode(201703101200l, 201703101259l); + } + + private void insertData() { + long minute = 201703101200l; + for (int i = 0; i < 60; i++) { + Map json = new HashMap(); + json.put("code", "DubboServer_MySQL"); + json.put("timeSlice", minute); + json.put("component", "Dubbo"); + json.put("layer", "http"); + + String _id = minute + "-DubboServer_MySQL"; + IndexResponse response = EsClient.getClient().prepareIndex(ApplicationIndexWithDagNodeType.Index, ApplicationIndexWithDagNodeType.Type, _id).setSource(json).get(); + RestStatus status = response.status(); + status.getStatus(); + minute++; + } + } +}