Service metric pyramid aggregate test successful.

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceDayMetricTransformNode.java

@@ -36,9 +36,10 @@ public class ServiceDayMetricTransformNode implements NodeProcessor<ServiceMetri
 
     @Override public void process(ServiceMetric serviceMetric, Next<ServiceMetric> next) {
         long timeBucket = TimeBucketUtils.INSTANCE.minuteToDay(serviceMetric.getTimeBucket());
-        serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
-        serviceMetric.setTimeBucket(timeBucket);
 
-        next.execute(serviceMetric);
+        ServiceMetric newServiceMetric = ServiceMetricCopy.copy(serviceMetric);
+        newServiceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
+        newServiceMetric.setTimeBucket(timeBucket);
+        next.execute(newServiceMetric);
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceHourMetricTransformNode.java

@@ -36,9 +36,10 @@ public class ServiceHourMetricTransformNode implements NodeProcessor<ServiceMetr
 
     @Override public void process(ServiceMetric serviceMetric, Next<ServiceMetric> next) {
         long timeBucket = TimeBucketUtils.INSTANCE.minuteToHour(serviceMetric.getTimeBucket());
-        serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
-        serviceMetric.setTimeBucket(timeBucket);
 
-        next.execute(serviceMetric);
+        ServiceMetric newServiceMetric = ServiceMetricCopy.copy(serviceMetric);
+        newServiceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
+        newServiceMetric.setTimeBucket(timeBucket);
+        next.execute(newServiceMetric);
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricCopy.java

+/*
+ * 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.analysis.metric.provider.worker.service.metric;
+
+import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ServiceMetricCopy {
+
+    public static ServiceMetric copy(ServiceMetric serviceMetric) {
+        ServiceMetric newServiceMetric = new ServiceMetric();
+        newServiceMetric.setId(serviceMetric.getId());
+        newServiceMetric.setMetricId(serviceMetric.getMetricId());
+        newServiceMetric.setSourceValue(serviceMetric.getSourceValue());
+
+        newServiceMetric.setApplicationId(serviceMetric.getApplicationId());
+        newServiceMetric.setInstanceId(serviceMetric.getInstanceId());
+        newServiceMetric.setServiceId(serviceMetric.getServiceId());
+
+        newServiceMetric.setTransactionCalls(serviceMetric.getTransactionCalls());
+        newServiceMetric.setTransactionDurationSum(serviceMetric.getTransactionDurationSum());
+        newServiceMetric.setTransactionErrorCalls(serviceMetric.getTransactionErrorCalls());
+        newServiceMetric.setTransactionErrorDurationSum(serviceMetric.getTransactionErrorDurationSum());
+
+        newServiceMetric.setBusinessTransactionCalls(serviceMetric.getBusinessTransactionCalls());
+        newServiceMetric.setBusinessTransactionDurationSum(serviceMetric.getBusinessTransactionDurationSum());
+        newServiceMetric.setBusinessTransactionErrorCalls(serviceMetric.getBusinessTransactionErrorCalls());
+        newServiceMetric.setBusinessTransactionErrorDurationSum(serviceMetric.getBusinessTransactionErrorDurationSum());
+
+        newServiceMetric.setMqTransactionCalls(serviceMetric.getMqTransactionCalls());
+        newServiceMetric.setMqTransactionDurationSum(serviceMetric.getMqTransactionDurationSum());
+        newServiceMetric.setMqTransactionErrorCalls(serviceMetric.getMqTransactionErrorCalls());
+        newServiceMetric.setMqTransactionErrorDurationSum(serviceMetric.getMqTransactionErrorDurationSum());
+
+        newServiceMetric.setTimeBucket(serviceMetric.getTimeBucket());
+
+        return newServiceMetric;
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricGraph.java

@@ -50,8 +50,8 @@ public class ServiceMetricGraph {
 
         Graph<ServiceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(MetricGraphIdDefine.SERVICE_METRIC_GRAPH_ID, ServiceReferenceMetric.class);
 
-        Node<ServiceMetric, ServiceMetric> remoteNode = graph.addNode(new ServiceMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener))
-            .addNext(new ServiceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, MetricGraphIdDefine.SERVICE_METRIC_GRAPH_ID).create(workerCreateListener));
+        Node<ServiceMetric, ServiceMetric> remoteNode = graph.addNode(new ServiceMetricMinuteAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ServiceMinuteMetricRemoteWorker.Factory(moduleManager, remoteSenderService, MetricGraphIdDefine.SERVICE_METRIC_GRAPH_ID).create(workerCreateListener));
 
         remoteNode.addNext(new ServiceMinuteMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
 
@@ -64,7 +64,7 @@ public class ServiceMetricGraph {
         remoteNode.addNext(new ServiceMonthMetricTransformNode())
             .addNext(new ServiceMonthMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
 
-//        link(graph);
+        link(graph);
     }
 
     private void link(Graph<ServiceReferenceMetric> graph) {

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricAggregationWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricMinuteAggregationWorker.java

@@ -29,9 +29,9 @@ import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenc
 /**
  * @author peng-yongsheng
  */
-public class ServiceMetricAggregationWorker extends AggregationWorker<ServiceReferenceMetric, ServiceMetric> {
+public class ServiceMetricMinuteAggregationWorker extends AggregationWorker<ServiceReferenceMetric, ServiceMetric> {
 
-    public ServiceMetricAggregationWorker(ModuleManager moduleManager) {
+    public ServiceMetricMinuteAggregationWorker(ModuleManager moduleManager) {
         super(moduleManager);
     }
 
@@ -76,14 +76,14 @@ public class ServiceMetricAggregationWorker extends AggregationWorker<ServiceRef
         return serviceMetric;
     }
 
-    public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceReferenceMetric, ServiceMetric, ServiceMetricAggregationWorker> {
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceReferenceMetric, ServiceMetric, ServiceMetricMinuteAggregationWorker> {
 
         public Factory(ModuleManager moduleManager) {
             super(moduleManager);
         }
 
-        @Override public ServiceMetricAggregationWorker workerInstance(ModuleManager moduleManager) {
-            return new ServiceMetricAggregationWorker(moduleManager);
+        @Override public ServiceMetricMinuteAggregationWorker workerInstance(ModuleManager moduleManager) {
+            return new ServiceMetricMinuteAggregationWorker(moduleManager);
         }
 
         @Override public int queueSize() {

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMinuteMetricRemoteWorker.java

@@ -30,9 +30,9 @@ import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
 /**
  * @author peng-yongsheng
  */
-public class ServiceMetricRemoteWorker extends AbstractRemoteWorker<ServiceMetric, ServiceMetric> {
+public class ServiceMinuteMetricRemoteWorker extends AbstractRemoteWorker<ServiceMetric, ServiceMetric> {
 
-    public ServiceMetricRemoteWorker(ModuleManager moduleManager) {
+    public ServiceMinuteMetricRemoteWorker(ModuleManager moduleManager) {
         super(moduleManager);
     }
 
@@ -48,13 +48,13 @@ public class ServiceMetricRemoteWorker extends AbstractRemoteWorker<ServiceMetri
         return Selector.HashCode;
     }
 
-    public static class Factory extends AbstractRemoteWorkerProvider<ServiceMetric, ServiceMetric, ServiceMetricRemoteWorker> {
+    public static class Factory extends AbstractRemoteWorkerProvider<ServiceMetric, ServiceMetric, ServiceMinuteMetricRemoteWorker> {
         public Factory(ModuleManager moduleManager, RemoteSenderService remoteSenderService, int graphId) {
             super(moduleManager, remoteSenderService, graphId);
         }
 
-        @Override public ServiceMetricRemoteWorker workerInstance(ModuleManager moduleManager) {
-            return new ServiceMetricRemoteWorker(moduleManager);
+        @Override public ServiceMinuteMetricRemoteWorker workerInstance(ModuleManager moduleManager) {
+            return new ServiceMinuteMetricRemoteWorker(moduleManager);
         }
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMonthMetricTransformNode.java

@@ -36,9 +36,10 @@ public class ServiceMonthMetricTransformNode implements NodeProcessor<ServiceMet
 
     @Override public void process(ServiceMetric serviceMetric, Next<ServiceMetric> next) {
         long timeBucket = TimeBucketUtils.INSTANCE.minuteToMonth(serviceMetric.getTimeBucket());
-        serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
-        serviceMetric.setTimeBucket(timeBucket);
 
-        next.execute(serviceMetric);
+        ServiceMetric newServiceMetric = ServiceMetricCopy.copy(serviceMetric);
+        newServiceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
+        newServiceMetric.setTimeBucket(timeBucket);
+        next.execute(newServiceMetric);
     }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java

@@ -79,7 +79,10 @@ import org.apache.skywalking.apm.collector.storage.dao.register.IApplicationRegi
 import org.apache.skywalking.apm.collector.storage.dao.register.IInstanceRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.register.INetworkAddressRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.register.IServiceNameRegisterDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceDayMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceHourMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMonthMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.srmp.IServiceReferenceDayMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.srmp.IServiceReferenceHourMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.srmp.IServiceReferenceMinuteMetricPersistenceDAO;
@@ -151,9 +154,9 @@ public class StorageModule extends Module {
         classes.add(ISegmentPersistenceDAO.class);
 
         classes.add(IServiceMinuteMetricPersistenceDAO.class);
-//        classes.add(IServiceHourMetricPersistenceDAO.class);
-//        classes.add(IServiceDayMetricPersistenceDAO.class);
-//        classes.add(IServiceMonthMetricPersistenceDAO.class);
+        classes.add(IServiceHourMetricPersistenceDAO.class);
+        classes.add(IServiceDayMetricPersistenceDAO.class);
+        classes.add(IServiceMonthMetricPersistenceDAO.class);
 
         classes.add(IServiceReferenceMinuteMetricPersistenceDAO.class);
         classes.add(IServiceReferenceHourMetricPersistenceDAO.class);

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/service/ServiceMetric.java

@@ -31,6 +31,7 @@ public class ServiceMetric extends StreamData implements Metric {
 
     private static final Column[] STRING_COLUMNS = {
         new Column(ServiceMetricTable.COLUMN_ID, new NonOperation()),
+        new Column(ServiceMetricTable.COLUMN_METRIC_ID, new NonOperation()),
     };
 
     private static final Column[] LONG_COLUMNS = {
@@ -76,11 +77,11 @@ public class ServiceMetric extends StreamData implements Metric {
     }
 
     @Override public String getMetricId() {
-        return getId();
+        return getDataString(1);
     }
 
     @Override public void setMetricId(String metricId) {
-        setId(metricId);
+        setDataString(1, metricId);
     }
 
     @Override

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java

@@ -88,7 +88,10 @@ import org.apache.skywalking.apm.collector.storage.dao.register.IApplicationRegi
 import org.apache.skywalking.apm.collector.storage.dao.register.IInstanceRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.register.INetworkAddressRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.register.IServiceNameRegisterDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceDayMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceHourMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMonthMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.srmp.IServiceReferenceDayMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.srmp.IServiceReferenceHourMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.srmp.IServiceReferenceMinuteMetricPersistenceDAO;
@@ -152,7 +155,10 @@ import org.apache.skywalking.apm.collector.storage.es.dao.register.ApplicationEs
 import org.apache.skywalking.apm.collector.storage.es.dao.register.InstanceEsRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.register.NetworkAddressRegisterEsDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.register.ServiceNameEsRegisterDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.smp.ServiceDayMetricEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.smp.ServiceHourMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.smp.ServiceMinuteMetricEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.smp.ServiceMonthMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.srmp.ServiceReferenceDayMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.srmp.ServiceReferenceHourMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.srmp.ServiceReferenceMinuteMetricEsPersistenceDAO;
@@ -274,7 +280,11 @@ public class StorageModuleEsProvider extends ModuleProvider {
 
         this.registerServiceImplementation(ISegmentCostPersistenceDAO.class, new SegmentCostEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(ISegmentPersistenceDAO.class, new SegmentEsPersistenceDAO(elasticSearchClient));
+
         this.registerServiceImplementation(IServiceMinuteMetricPersistenceDAO.class, new ServiceMinuteMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IServiceHourMetricPersistenceDAO.class, new ServiceHourMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IServiceDayMetricPersistenceDAO.class, new ServiceDayMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IServiceMonthMetricPersistenceDAO.class, new ServiceMonthMetricEsPersistenceDAO(elasticSearchClient));
 
         this.registerServiceImplementation(IServiceReferenceMinuteMetricPersistenceDAO.class, new ServiceReferenceMinuteMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IServiceReferenceHourMetricPersistenceDAO.class, new ServiceReferenceHourMetricEsPersistenceDAO(elasticSearchClient));

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceDayMetricEsPersistenceDAO.java

@@ -21,7 +21,7 @@ package org.apache.skywalking.apm.collector.storage.es.dao.smp;
 import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
 import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
 import org.apache.skywalking.apm.collector.core.util.Const;
-import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceDayMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
 import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable;
 import org.elasticsearch.action.index.IndexRequestBuilder;
@@ -30,7 +30,7 @@ import org.elasticsearch.action.update.UpdateRequestBuilder;
 /**
  * @author peng-yongsheng
  */
-public class ServiceDayMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceMinuteMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
+public class ServiceDayMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceDayMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
 
     public ServiceDayMetricEsPersistenceDAO(ElasticSearchClient client) {
         super(client);

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceHourMetricEsPersistenceDAO.java

@@ -21,7 +21,7 @@ package org.apache.skywalking.apm.collector.storage.es.dao.smp;
 import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
 import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
 import org.apache.skywalking.apm.collector.core.util.Const;
-import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceHourMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
 import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable;
 import org.elasticsearch.action.index.IndexRequestBuilder;
@@ -30,7 +30,7 @@ import org.elasticsearch.action.update.UpdateRequestBuilder;
 /**
  * @author peng-yongsheng
  */
-public class ServiceHourMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceMinuteMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
+public class ServiceHourMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceHourMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
 
     public ServiceHourMetricEsPersistenceDAO(ElasticSearchClient client) {
         super(client);

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMonthMetricEsPersistenceDAO.java

@@ -21,7 +21,7 @@ package org.apache.skywalking.apm.collector.storage.es.dao.smp;
 import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
 import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
 import org.apache.skywalking.apm.collector.core.util.Const;
-import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMonthMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
 import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable;
 import org.elasticsearch.action.index.IndexRequestBuilder;
@@ -30,7 +30,7 @@ import org.elasticsearch.action.update.UpdateRequestBuilder;
 /**
  * @author peng-yongsheng
  */
-public class ServiceMonthMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceMinuteMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
+public class ServiceMonthMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceMonthMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
 
     public ServiceMonthMetricEsPersistenceDAO(ElasticSearchClient client) {
         super(client);

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/ServiceMetricEsTableDefine.java → apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/smp/AbstractServiceMetricEsTableDefine.java

@@ -16,8 +16,7 @@
  *
  */
 
-
-package org.apache.skywalking.apm.collector.storage.es.define;
+package org.apache.skywalking.apm.collector.storage.es.define.smp;
 
 import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchColumnDefine;
 import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchTableDefine;
@@ -26,18 +25,17 @@ import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTa
 /**
  * @author peng-yongsheng
  */
-public class ServiceMetricEsTableDefine extends ElasticSearchTableDefine {
-
-    public ServiceMetricEsTableDefine() {
-        super(ServiceMetricTable.TABLE);
-    }
+public abstract class AbstractServiceMetricEsTableDefine extends ElasticSearchTableDefine {
 
-    @Override public int refreshInterval() {
-        return 2;
+    public AbstractServiceMetricEsTableDefine(String name) {
+        super(name);
     }
 
-    @Override public void initialize() {
+    @Override public final void initialize() {
+        addColumn(new ElasticSearchColumnDefine(ServiceMetricTable.COLUMN_ID, ElasticSearchColumnDefine.Type.Keyword.name()));
+        addColumn(new ElasticSearchColumnDefine(ServiceMetricTable.COLUMN_METRIC_ID, ElasticSearchColumnDefine.Type.Keyword.name()));
         addColumn(new ElasticSearchColumnDefine(ServiceMetricTable.COLUMN_SERVICE_ID, ElasticSearchColumnDefine.Type.Integer.name()));
+        addColumn(new ElasticSearchColumnDefine(ServiceMetricTable.COLUMN_SOURCE_VALUE, ElasticSearchColumnDefine.Type.Integer.name()));
 
         addColumn(new ElasticSearchColumnDefine(ServiceMetricTable.COLUMN_TRANSACTION_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
         addColumn(new ElasticSearchColumnDefine(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, ElasticSearchColumnDefine.Type.Long.name()));

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/smp/ServiceDayMetricEsTableDefine.java

+/*
+ * 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.define.smp;
+
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ServiceDayMetricEsTableDefine extends AbstractServiceMetricEsTableDefine {
+
+    public ServiceDayMetricEsTableDefine() {
+        super(ServiceMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Day.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 2;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/smp/ServiceHourMetricEsTableDefine.java

+/*
+ * 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.define.smp;
+
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ServiceHourMetricEsTableDefine extends AbstractServiceMetricEsTableDefine {
+
+    public ServiceHourMetricEsTableDefine() {
+        super(ServiceMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Hour.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 2;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/smp/ServiceMinuteMetricEsTableDefine.java

+/*
+ * 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.define.smp;
+
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ServiceMinuteMetricEsTableDefine extends AbstractServiceMetricEsTableDefine {
+
+    public ServiceMinuteMetricEsTableDefine() {
+        super(ServiceMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Minute.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 2;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/smp/ServiceMonthMetricEsTableDefine.java

+/*
+ * 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.define.smp;
+
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ServiceMonthMetricEsTableDefine extends AbstractServiceMetricEsTableDefine {
+
+    public ServiceMonthMetricEsTableDefine() {
+        super(ServiceMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Month.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 2;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/resources/META-INF/defines/storage.define

@@ -23,6 +23,11 @@ org.apache.skywalking.apm.collector.storage.es.define.srmp.ServiceReferenceHourM
 org.apache.skywalking.apm.collector.storage.es.define.srmp.ServiceReferenceDayMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.srmp.ServiceReferenceMonthMetricEsTableDefine
 
+org.apache.skywalking.apm.collector.storage.es.define.smp.ServiceMinuteMetricEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.smp.ServiceHourMetricEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.smp.ServiceDayMetricEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.smp.ServiceMonthMetricEsTableDefine
+
 org.apache.skywalking.apm.collector.storage.es.define.GlobalTraceEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.SegmentEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.SegmentCostEsTableDefine
\ No newline at end of file