diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/base/dao/IPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/base/dao/IPersistenceDAO.java index 0a476d3de..86034e892 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/base/dao/IPersistenceDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/base/dao/IPersistenceDAO.java @@ -16,20 +16,19 @@ * */ - package org.apache.skywalking.apm.collector.storage.base.dao; -import org.apache.skywalking.apm.collector.core.data.AbstractData; +import org.apache.skywalking.apm.collector.core.data.StreamData; /** * @author peng-yongsheng */ -public interface IPersistenceDAO extends DAO { - DataImpl get(String id); +public interface IPersistenceDAO extends DAO { + STREAM_DATA get(String id); - Insert prepareBatchInsert(DataImpl data); + Insert prepareBatchInsert(STREAM_DATA data); - Update prepareBatchUpdate(DataImpl data); + Update prepareBatchUpdate(STREAM_DATA data); void deleteHistory(Long startTimestamp, Long endTimestamp); } diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IInstanceMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IInstanceMetricPersistenceDAO.java index 009eef728..dc6fa9fb2 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IInstanceMetricPersistenceDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IInstanceMetricPersistenceDAO.java @@ -24,5 +24,5 @@ import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric /** * @author peng-yongsheng */ -public interface IInstanceMetricPersistenceDAO extends IPersistenceDAO { +public interface IInstanceMetricPersistenceDAO extends IPersistenceDAO { } diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationComponent.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationComponent.java index 23bf9e1e1..0e4187dd1 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationComponent.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationComponent.java @@ -16,18 +16,17 @@ * */ - package org.apache.skywalking.apm.collector.storage.table.application; import org.apache.skywalking.apm.collector.core.data.Column; -import org.apache.skywalking.apm.collector.core.data.AbstractData; +import org.apache.skywalking.apm.collector.core.data.StreamData; import org.apache.skywalking.apm.collector.core.data.operator.CoverOperation; import org.apache.skywalking.apm.collector.core.data.operator.NonOperation; /** * @author peng-yongsheng */ -public class ApplicationComponent extends AbstractData { +public class ApplicationComponent extends StreamData { private static final Column[] STRING_COLUMNS = { new Column(ApplicationComponentTable.COLUMN_ID, new NonOperation()), @@ -45,8 +44,24 @@ public class ApplicationComponent extends AbstractData { private static final Column[] BOOLEAN_COLUMNS = {}; private static final Column[] BYTE_COLUMNS = {}; - public ApplicationComponent(String id) { - super(id, STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS); + public ApplicationComponent() { + super(STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS); + } + + @Override public String getId() { + return null; + } + + @Override public void setId(String id) { + + } + + @Override public String getMetricId() { + return null; + } + + @Override public void setMetricId(String metricId) { + } public Long getTimeBucket() { diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/instance/InstanceMetric.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/instance/InstanceMetric.java index 9dbf3a355..348742310 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/instance/InstanceMetric.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/instance/InstanceMetric.java @@ -19,7 +19,7 @@ package org.apache.skywalking.apm.collector.storage.table.instance; import org.apache.skywalking.apm.collector.core.data.Column; -import org.apache.skywalking.apm.collector.core.data.AbstractData; +import org.apache.skywalking.apm.collector.core.data.StreamData; import org.apache.skywalking.apm.collector.core.data.operator.AddOperation; import org.apache.skywalking.apm.collector.core.data.operator.CoverOperation; import org.apache.skywalking.apm.collector.core.data.operator.NonOperation; @@ -28,10 +28,11 @@ import org.apache.skywalking.apm.collector.storage.table.Metric; /** * @author peng-yongsheng */ -public class InstanceMetric extends AbstractData implements Metric { +public class InstanceMetric extends StreamData implements Metric { private static final Column[] STRING_COLUMNS = { new Column(InstanceMetricTable.COLUMN_ID, new NonOperation()), + new Column(InstanceMetricTable.COLUMN_METRIC_ID, new NonOperation()), }; private static final Column[] LONG_COLUMNS = { @@ -62,8 +63,24 @@ public class InstanceMetric extends AbstractData implements Metric { private static final Column[] BOOLEAN_COLUMNS = {}; private static final Column[] BYTE_COLUMNS = {}; - public InstanceMetric(String id) { - super(id, STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS); + public InstanceMetric() { + super(STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS); + } + + @Override public String getId() { + return getDataString(0); + } + + @Override public void setId(String id) { + setDataString(0, id); + } + + @Override public String getMetricId() { + return getDataString(1); + } + + @Override public void setMetricId(String metricId) { + setDataString(1, metricId); } @Override diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/AbstractPersistenceEsDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/AbstractPersistenceEsDAO.java new file mode 100644 index 000000000..31a508ff2 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/AbstractPersistenceEsDAO.java @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.storage.es.base.dao; + +import java.util.Map; +import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; +import org.apache.skywalking.apm.collector.core.data.StreamData; +import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils; +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.elasticsearch.action.get.GetResponse; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.index.reindex.BulkByScrollResponse; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public abstract class AbstractPersistenceEsDAO extends EsDAO implements IPersistenceDAO { + + private final Logger logger = LoggerFactory.getLogger(AbstractPersistenceEsDAO.class); + + protected AbstractPersistenceEsDAO(ElasticSearchClient client) { + super(client); + } + + protected abstract String tableName(); + + protected abstract STREAM_DATA esDataToStreamData(Map source); + + @Override public final STREAM_DATA get(String id) { + GetResponse getResponse = getClient().prepareGet(tableName(), id).get(); + if (getResponse.isExists()) { + return esDataToStreamData(getResponse.getSource()); + } else { + return null; + } + } + + protected abstract Map esStreamDataToEsData(STREAM_DATA streamData); + + @Override public final IndexRequestBuilder prepareBatchInsert(STREAM_DATA streamData) { + Map source = esStreamDataToEsData(streamData); + return getClient().prepareIndex(tableName(), streamData.getId()).setSource(source); + } + + @Override public final UpdateRequestBuilder prepareBatchUpdate(STREAM_DATA streamData) { + Map source = esStreamDataToEsData(streamData); + return getClient().prepareUpdate(tableName(), streamData.getId()).setDoc(source); + } + + protected abstract String timeBucketColumnNameForDelete(); + + @Override public final void deleteHistory(Long startTimestamp, Long endTimestamp) { + long startTimeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(startTimestamp); + long endTimeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(endTimestamp); + BulkByScrollResponse response = getClient().prepareDelete() + .filter(QueryBuilders.rangeQuery(timeBucketColumnNameForDelete()).gte(startTimeBucket).lte(endTimeBucket)) + .source(tableName()) + .get(); + + long deleted = response.getDeleted(); + logger.info("Delete {} rows history from {} index.", deleted, tableName()); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/BatchEsDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/BatchEsDAO.java index 9c94701f4..0e7e146f3 100644 --- a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/BatchEsDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/BatchEsDAO.java @@ -16,17 +16,16 @@ * */ - package org.apache.skywalking.apm.collector.storage.es.base.dao; import java.util.List; import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; import org.apache.skywalking.apm.collector.core.util.CollectionUtils; +import org.apache.skywalking.apm.collector.storage.base.dao.IBatchDAO; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.apache.skywalking.apm.collector.storage.base.dao.IBatchDAO; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,15 +45,14 @@ public class BatchEsDAO extends EsDAO implements IBatchDAO { logger.debug("bulk data size: {}", batchCollection.size()); if (CollectionUtils.isNotEmpty(batchCollection)) { - for (int i = 0; i < batchCollection.size(); i++) { - Object builder = batchCollection.get(i); + batchCollection.forEach(builder -> { if (builder instanceof IndexRequestBuilder) { bulkRequest.add((IndexRequestBuilder)builder); } if (builder instanceof UpdateRequestBuilder) { bulkRequest.add((UpdateRequestBuilder)builder); } - } + }); BulkResponse bulkResponse = bulkRequest.execute().actionGet(); if (bulkResponse.hasFailures()) { diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/EsDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/EsDAO.java index 95ea3e4d8..0b10629e6 100644 --- a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/EsDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/base/dao/EsDAO.java @@ -16,9 +16,10 @@ * */ - package org.apache.skywalking.apm.collector.storage.es.base.dao; +import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; +import org.apache.skywalking.apm.collector.storage.base.dao.AbstractDAO; import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.search.aggregations.AggregationBuilders; @@ -26,8 +27,6 @@ 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.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; -import org.apache.skywalking.apm.collector.storage.base.dao.AbstractDAO; /** * @author peng-yongsheng @@ -38,7 +37,7 @@ public abstract class EsDAO extends AbstractDAO { super(client); } - public final int getMaxId(String indexName, String columnName) { + protected final int getMaxId(String indexName, String columnName) { ElasticSearchClient client = getClient(); SearchRequestBuilder searchRequestBuilder = client.prepareSearch(indexName); searchRequestBuilder.setTypes("type"); @@ -57,7 +56,7 @@ public abstract class EsDAO extends AbstractDAO { } } - public final int getMinId(String indexName, String columnName) { + protected final int getMinId(String indexName, String columnName) { ElasticSearchClient client = getClient(); SearchRequestBuilder searchRequestBuilder = client.prepareSearch(indexName); searchRequestBuilder.setTypes("type"); diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/InstanceMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/InstanceMetricEsPersistenceDAO.java index 6ed0d38b2..bc22aed4e 100644 --- a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/InstanceMetricEsPersistenceDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/InstanceMetricEsPersistenceDAO.java @@ -21,124 +21,82 @@ package org.apache.skywalking.apm.collector.storage.es.dao; import java.util.HashMap; import java.util.Map; import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; -import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils; import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricPersistenceDAO; -import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO; +import org.apache.skywalking.apm.collector.storage.es.base.dao.AbstractPersistenceEsDAO; import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric; import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetricTable; -import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.index.reindex.BulkByScrollResponse; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * @author peng-yongsheng */ -public class InstanceMetricEsPersistenceDAO extends EsDAO implements IInstanceMetricPersistenceDAO { - - private final Logger logger = LoggerFactory.getLogger(InstanceMetricEsPersistenceDAO.class); +public class InstanceMetricEsPersistenceDAO extends AbstractPersistenceEsDAO implements IInstanceMetricPersistenceDAO { public InstanceMetricEsPersistenceDAO(ElasticSearchClient client) { super(client); } - @Override public InstanceMetric get(String id) { - GetResponse getResponse = getClient().prepareGet(InstanceMetricTable.TABLE, id).get(); - if (getResponse.isExists()) { - logger.debug("getId: {} is exist", id); - InstanceMetric instanceMetric = new InstanceMetric(id); - Map source = getResponse.getSource(); - instanceMetric.setApplicationId((Integer)source.get(InstanceMetricTable.COLUMN_APPLICATION_ID)); - instanceMetric.setInstanceId((Integer)source.get(InstanceMetricTable.COLUMN_INSTANCE_ID)); - instanceMetric.setSourceValue((Integer)source.get(InstanceMetricTable.COLUMN_SOURCE_VALUE)); - - instanceMetric.setTransactionCalls(((Number)source.get(InstanceMetricTable.COLUMN_TRANSACTION_CALLS)).longValue()); - instanceMetric.setTransactionErrorCalls(((Number)source.get(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS)).longValue()); - instanceMetric.setTransactionDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_TRANSACTION_DURATION_SUM)).longValue()); - instanceMetric.setTransactionErrorDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - instanceMetric.setBusinessTransactionCalls(((Number)source.get(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS)).longValue()); - instanceMetric.setBusinessTransactionErrorCalls(((Number)source.get(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS)).longValue()); - instanceMetric.setBusinessTransactionDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM)).longValue()); - instanceMetric.setBusinessTransactionErrorDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - instanceMetric.setMqTransactionCalls(((Number)source.get(InstanceMetricTable.COLUMN_MQ_TRANSACTION_CALLS)).longValue()); - instanceMetric.setMqTransactionErrorCalls(((Number)source.get(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS)).longValue()); - instanceMetric.setMqTransactionDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM)).longValue()); - instanceMetric.setMqTransactionErrorDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - instanceMetric.setTimeBucket(((Number)source.get(InstanceMetricTable.COLUMN_TIME_BUCKET)).longValue()); - - return instanceMetric; - } else { - return null; - } + @Override protected String tableName() { + return InstanceMetricTable.TABLE; } - @Override public IndexRequestBuilder prepareBatchInsert(InstanceMetric data) { + @Override protected String timeBucketColumnNameForDelete() { + return InstanceMetricTable.COLUMN_TIME_BUCKET; + } + + @Override protected InstanceMetric esDataToStreamData(Map source) { + InstanceMetric instanceMetric = new InstanceMetric(); + instanceMetric.setId((String)source.get(InstanceMetricTable.COLUMN_ID)); + instanceMetric.setMetricId((String)source.get(InstanceMetricTable.COLUMN_METRIC_ID)); + instanceMetric.setApplicationId((Integer)source.get(InstanceMetricTable.COLUMN_APPLICATION_ID)); + instanceMetric.setInstanceId((Integer)source.get(InstanceMetricTable.COLUMN_INSTANCE_ID)); + instanceMetric.setSourceValue((Integer)source.get(InstanceMetricTable.COLUMN_SOURCE_VALUE)); + + instanceMetric.setTransactionCalls(((Number)source.get(InstanceMetricTable.COLUMN_TRANSACTION_CALLS)).longValue()); + instanceMetric.setTransactionErrorCalls(((Number)source.get(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS)).longValue()); + instanceMetric.setTransactionDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_TRANSACTION_DURATION_SUM)).longValue()); + instanceMetric.setTransactionErrorDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + instanceMetric.setBusinessTransactionCalls(((Number)source.get(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS)).longValue()); + instanceMetric.setBusinessTransactionErrorCalls(((Number)source.get(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS)).longValue()); + instanceMetric.setBusinessTransactionDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM)).longValue()); + instanceMetric.setBusinessTransactionErrorDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + instanceMetric.setMqTransactionCalls(((Number)source.get(InstanceMetricTable.COLUMN_MQ_TRANSACTION_CALLS)).longValue()); + instanceMetric.setMqTransactionErrorCalls(((Number)source.get(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS)).longValue()); + instanceMetric.setMqTransactionDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM)).longValue()); + instanceMetric.setMqTransactionErrorDurationSum(((Number)source.get(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + instanceMetric.setTimeBucket(((Number)source.get(InstanceMetricTable.COLUMN_TIME_BUCKET)).longValue()); + return null; + } + + @Override protected Map esStreamDataToEsData(InstanceMetric streamData) { Map source = new HashMap<>(); - source.put(InstanceMetricTable.COLUMN_APPLICATION_ID, data.getApplicationId()); - source.put(InstanceMetricTable.COLUMN_INSTANCE_ID, data.getInstanceId()); - source.put(InstanceMetricTable.COLUMN_SOURCE_VALUE, data.getSourceValue()); + source.put(InstanceMetricTable.COLUMN_METRIC_ID, streamData.getMetricId()); - source.put(InstanceMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls()); - source.put(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls()); - source.put(InstanceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum()); - source.put(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum()); + source.put(InstanceMetricTable.COLUMN_APPLICATION_ID, streamData.getApplicationId()); + source.put(InstanceMetricTable.COLUMN_INSTANCE_ID, streamData.getInstanceId()); + source.put(InstanceMetricTable.COLUMN_SOURCE_VALUE, streamData.getSourceValue()); - source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls()); - source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls()); - source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum()); - source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum()); + source.put(InstanceMetricTable.COLUMN_TRANSACTION_CALLS, streamData.getTransactionCalls()); + source.put(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, streamData.getTransactionErrorCalls()); + source.put(InstanceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, streamData.getTransactionDurationSum()); + source.put(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, streamData.getTransactionErrorDurationSum()); - source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls()); - source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls()); - source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum()); - source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum()); + source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, streamData.getBusinessTransactionCalls()); + source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, streamData.getBusinessTransactionErrorCalls()); + source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, streamData.getBusinessTransactionDurationSum()); + source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, streamData.getBusinessTransactionErrorDurationSum()); - source.put(InstanceMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket()); + source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, streamData.getMqTransactionCalls()); + source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, streamData.getMqTransactionErrorCalls()); + source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, streamData.getMqTransactionDurationSum()); + source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, streamData.getMqTransactionErrorDurationSum()); - return getClient().prepareIndex(InstanceMetricTable.TABLE, data.getId()).setSource(source); - } + source.put(InstanceMetricTable.COLUMN_TIME_BUCKET, streamData.getTimeBucket()); - @Override public UpdateRequestBuilder prepareBatchUpdate(InstanceMetric data) { - Map source = new HashMap<>(); - source.put(InstanceMetricTable.COLUMN_APPLICATION_ID, data.getApplicationId()); - source.put(InstanceMetricTable.COLUMN_INSTANCE_ID, data.getInstanceId()); - source.put(InstanceMetricTable.COLUMN_SOURCE_VALUE, data.getSourceValue()); - - source.put(InstanceMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls()); - source.put(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls()); - source.put(InstanceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum()); - source.put(InstanceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum()); - - source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls()); - source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls()); - source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum()); - source.put(InstanceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum()); - - source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls()); - source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls()); - source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum()); - source.put(InstanceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum()); - - source.put(InstanceMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket()); - - return getClient().prepareUpdate(InstanceMetricTable.TABLE, data.getId()).setDoc(source); - } - - @Override public void deleteHistory(Long startTimestamp, Long endTimestamp) { - long startTimeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(startTimestamp); - long endTimeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(endTimestamp); - BulkByScrollResponse response = getClient().prepareDelete() - .filter(QueryBuilders.rangeQuery(InstanceMetricTable.COLUMN_TIME_BUCKET).gte(startTimeBucket).lte(endTimeBucket)) - .source(InstanceMetricTable.TABLE) - .get(); - - long deleted = response.getDeleted(); - logger.info("Delete {} rows history from {} index.", deleted, InstanceMetricTable.TABLE); + return source; } }