Aggregate response time distribution metric and persistent into database. (#1104)

apm-collector/apm-collector-agent/agent-grpc/agent-grpc-provider/src/test/java/org/apache/skywalking/apm/collector/agent/grpc/provider/handler/mock/ProviderMock.java

@@ -85,7 +85,7 @@ class ProviderMock {
         span.setStartTime(startTimestamp + 510);
         span.setEndTime(startTimestamp + 1490);
         span.setComponentId(ComponentsDefine.MONGODB.getId());
-        span.setIsError(false);
+        span.setIsError(true);
 
         if (isPrepare) {
             span.setOperationName("mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]");

apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/MetricGraphIdDefine.java

@@ -32,7 +32,7 @@ public class MetricGraphIdDefine {
 
     public static final int APPLICATION_COMPONENT_GRAPH_ID = 406;
     public static final int APPLICATION_MAPPING_GRAPH_ID = 407;
-    public static final int SERVICE_MAPPING_GRAPH_ID = 408;
+    public static final int RESPONSE_TIME_DISTRIBUTION_GRAPH_ID = 408;
     public static final int GLOBAL_TRACE_GRAPH_ID = 409;
     public static final int SEGMENT_DURATION_GRAPH_ID = 410;
     public static final int INSTANCE_MAPPING_GRAPH_ID = 411;

apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/MetricWorkerIdDefine.java

@@ -112,6 +112,16 @@ public class MetricWorkerIdDefine {
     public static final int APPLICATION_COMPONENT_MONTH_PERSISTENCE_WORKER_ID = 4907;
     public static final int APPLICATION_COMPONENT_MONTH_TRANSFORM_NODE_ID = 4908;
 
+    public static final int RESPONSE_TIME_DISTRIBUTION_MINUTE_AGGREGATION_WORKER_ID = 4040;
+    public static final int RESPONSE_TIME_DISTRIBUTION_MINUTE_REMOTE_WORKER_ID = 4041;
+    public static final int RESPONSE_TIME_DISTRIBUTION_MINUTE_PERSISTENCE_WORKER_ID = 4042;
+    public static final int RESPONSE_TIME_DISTRIBUTION_HOUR_PERSISTENCE_WORKER_ID = 4043;
+    public static final int RESPONSE_TIME_DISTRIBUTION_HOUR_TRANSFORM_NODE_ID = 4044;
+    public static final int RESPONSE_TIME_DISTRIBUTION_DAY_PERSISTENCE_WORKER_ID = 4045;
+    public static final int RESPONSE_TIME_DISTRIBUTION_DAY_TRANSFORM_NODE_ID = 4046;
+    public static final int RESPONSE_TIME_DISTRIBUTION_MONTH_PERSISTENCE_WORKER_ID = 4047;
+    public static final int RESPONSE_TIME_DISTRIBUTION_MONTH_TRANSFORM_NODE_ID = 4048;
+
     public static final int GLOBAL_TRACE_PERSISTENCE_WORKER_ID = 427;
     public static final int SEGMENT_DURATION_PERSISTENCE_WORKER_ID = 428;
 

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java

@@ -29,6 +29,8 @@ import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.appli
 import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.refmetric.ApplicationReferenceMetricGraph;
 import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTraceGraph;
 import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTraceSpanListener;
+import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.std.ResponseTimeDistributionGraph;
+import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.std.ResponseTimeDistributionSpanListener;
 import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.heartbeat.InstanceHeartBeatPersistenceGraph;
 import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.mapping.InstanceMappingGraph;
 import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.mapping.InstanceMappingSpanListener;
@@ -56,6 +58,7 @@ import org.apache.skywalking.apm.collector.storage.table.application.Application
 import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping;
 import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric;
 import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
 import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMapping;
 import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric;
 import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
@@ -118,6 +121,7 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
         segmentParserListenerRegister.register(new InstanceMappingSpanListener.Factory());
         segmentParserListenerRegister.register(new GlobalTraceSpanListener.Factory());
         segmentParserListenerRegister.register(new SegmentDurationSpanListener.Factory());
+        segmentParserListenerRegister.register(new ResponseTimeDistributionSpanListener.Factory());
     }
 
     private void graphCreate(WorkerCreateListener workerCreateListener) {
@@ -151,6 +155,9 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
         GlobalTraceGraph globalTraceGraph = new GlobalTraceGraph(getManager(), workerCreateListener);
         globalTraceGraph.create();
 
+        ResponseTimeDistributionGraph responseTimeDistributionGraph = new ResponseTimeDistributionGraph(getManager(), workerCreateListener);
+        responseTimeDistributionGraph.create();
+
         SegmentDurationGraph segmentDurationGraph = new SegmentDurationGraph(getManager(), workerCreateListener);
         segmentDurationGraph.create();
 
@@ -169,5 +176,6 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
         remoteDataRegisterService.register(InstanceReferenceMetric.class, new InstanceReferenceMetric.InstanceCreator());
         remoteDataRegisterService.register(ServiceMetric.class, new ServiceMetric.InstanceCreator());
         remoteDataRegisterService.register(ServiceReferenceMetric.class, new ServiceReferenceMetric.InstanceCreator());
+        remoteDataRegisterService.register(ResponseTimeDistribution.class, new ResponseTimeDistribution.InstanceCreator());
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionCopy.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.global.std;
+
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionCopy {
+
+    public static ResponseTimeDistribution copy(ResponseTimeDistribution responseTimeDistribution) {
+        ResponseTimeDistribution newResponseTimeDistribution = new ResponseTimeDistribution();
+        newResponseTimeDistribution.setId(responseTimeDistribution.getId());
+        newResponseTimeDistribution.setMetricId(responseTimeDistribution.getMetricId());
+        newResponseTimeDistribution.setTimeBucket(responseTimeDistribution.getTimeBucket());
+
+        newResponseTimeDistribution.setStep(responseTimeDistribution.getStep());
+        newResponseTimeDistribution.setCalls(responseTimeDistribution.getCalls());
+        newResponseTimeDistribution.setSuccessCalls(responseTimeDistribution.getSuccessCalls());
+        newResponseTimeDistribution.setErrorCalls(responseTimeDistribution.getErrorCalls());
+        return newResponseTimeDistribution;
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionDayPersistenceWorker.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorkerProvider;
+import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionDayPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionDayPersistenceWorker extends PersistenceWorker<ResponseTimeDistribution> {
+
+    private ResponseTimeDistributionDayPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_DAY_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, ResponseTimeDistribution> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(IResponseTimeDistributionDayPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<ResponseTimeDistribution, ResponseTimeDistributionDayPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public ResponseTimeDistributionDayPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new ResponseTimeDistributionDayPersistenceWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+
+    @GraphComputingMetric(name = "/persistence/onWork/" + ResponseTimeDistributionTable.TABLE + "/day")
+    @Override protected void onWork(ResponseTimeDistribution input) {
+        super.onWork(input);
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionDayTransformNode.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.core.graph.Next;
+import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionDayTransformNode implements NodeProcessor<ResponseTimeDistribution, ResponseTimeDistribution> {
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_DAY_TRANSFORM_NODE_ID;
+    }
+
+    @Override
+    public void process(ResponseTimeDistribution responseTimeDistribution, Next<ResponseTimeDistribution> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.minuteToDay(responseTimeDistribution.getTimeBucket());
+
+        ResponseTimeDistribution newResponseTimeDistribution = ResponseTimeDistributionCopy.copy(responseTimeDistribution);
+        newResponseTimeDistribution.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + responseTimeDistribution.getMetricId());
+        newResponseTimeDistribution.setTimeBucket(timeBucket);
+        next.execute(newResponseTimeDistribution);
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionGraph.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricGraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
+import org.apache.skywalking.apm.collector.core.graph.GraphManager;
+import org.apache.skywalking.apm.collector.core.graph.Node;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionGraph {
+
+    private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
+
+    public ResponseTimeDistributionGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
+        this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
+    }
+
+    public void create() {
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        Node<ResponseTimeDistribution, ResponseTimeDistribution> remoteNode = GraphManager.INSTANCE.createIfAbsent(MetricGraphIdDefine.RESPONSE_TIME_DISTRIBUTION_GRAPH_ID, ResponseTimeDistribution.class)
+            .addNode(new ResponseTimeDistributionMinuteAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ResponseTimeDistributionMinuteRemoteWorker.Factory(moduleManager, remoteSenderService, MetricGraphIdDefine.RESPONSE_TIME_DISTRIBUTION_GRAPH_ID).create(workerCreateListener));
+
+        remoteNode.addNext(new ResponseTimeDistributionMinutePersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        remoteNode.addNext(new ResponseTimeDistributionHourTransformNode())
+            .addNext(new ResponseTimeDistributionHourPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        remoteNode.addNext(new ResponseTimeDistributionDayTransformNode())
+            .addNext(new ResponseTimeDistributionDayPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        remoteNode.addNext(new ResponseTimeDistributionMonthTransformNode())
+            .addNext(new ResponseTimeDistributionMonthPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionHourPersistenceWorker.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorkerProvider;
+import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionHourPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionHourPersistenceWorker extends PersistenceWorker<ResponseTimeDistribution> {
+
+    private ResponseTimeDistributionHourPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_HOUR_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, ResponseTimeDistribution> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(IResponseTimeDistributionHourPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<ResponseTimeDistribution, ResponseTimeDistributionHourPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public ResponseTimeDistributionHourPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new ResponseTimeDistributionHourPersistenceWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+
+    @GraphComputingMetric(name = "/persistence/onWork/" + ResponseTimeDistributionTable.TABLE + "/hour")
+    @Override protected void onWork(ResponseTimeDistribution input) {
+        super.onWork(input);
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionHourTransformNode.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.core.graph.Next;
+import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionHourTransformNode implements NodeProcessor<ResponseTimeDistribution, ResponseTimeDistribution> {
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_HOUR_TRANSFORM_NODE_ID;
+    }
+
+    @Override
+    public void process(ResponseTimeDistribution responseTimeDistribution, Next<ResponseTimeDistribution> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.minuteToHour(responseTimeDistribution.getTimeBucket());
+
+        ResponseTimeDistribution newResponseTimeDistribution = ResponseTimeDistributionCopy.copy(responseTimeDistribution);
+        newResponseTimeDistribution.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + responseTimeDistribution.getMetricId());
+        newResponseTimeDistribution.setTimeBucket(timeBucket);
+        next.execute(newResponseTimeDistribution);
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionMinuteAggregationWorker.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
+import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMinuteAggregationWorker extends AggregationWorker<ResponseTimeDistribution, ResponseTimeDistribution> {
+
+    private ResponseTimeDistributionMinuteAggregationWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_MINUTE_AGGREGATION_WORKER_ID;
+    }
+
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<ResponseTimeDistribution, ResponseTimeDistribution, ResponseTimeDistributionMinuteAggregationWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public ResponseTimeDistributionMinuteAggregationWorker workerInstance(ModuleManager moduleManager) {
+            return new ResponseTimeDistributionMinuteAggregationWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+
+    @GraphComputingMetric(name = "/aggregation/onWork/" + ResponseTimeDistributionTable.TABLE)
+    @Override protected void onWork(ResponseTimeDistribution message) throws WorkerException {
+        super.onWork(message);
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionMinutePersistenceWorker.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorkerProvider;
+import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionMinutePersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMinutePersistenceWorker extends PersistenceWorker<ResponseTimeDistribution> {
+
+    private ResponseTimeDistributionMinutePersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_MINUTE_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, ResponseTimeDistribution> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(IResponseTimeDistributionMinutePersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<ResponseTimeDistribution, ResponseTimeDistributionMinutePersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public ResponseTimeDistributionMinutePersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new ResponseTimeDistributionMinutePersistenceWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+
+    @GraphComputingMetric(name = "/persistence/onWork/" + ResponseTimeDistributionTable.TABLE + "/minute")
+    @Override protected void onWork(ResponseTimeDistribution input) {
+        super.onWork(input);
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionMinuteRemoteWorker.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
+import org.apache.skywalking.apm.collector.remote.service.Selector;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMinuteRemoteWorker extends AbstractRemoteWorker<ResponseTimeDistribution, ResponseTimeDistribution> {
+
+    private ResponseTimeDistributionMinuteRemoteWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_MINUTE_REMOTE_WORKER_ID;
+    }
+
+    @Override protected void onWork(ResponseTimeDistribution responseTimeDistribution) {
+        onNext(responseTimeDistribution);
+    }
+
+    @Override public Selector selector() {
+        return Selector.HashCode;
+    }
+
+    public static class Factory extends AbstractRemoteWorkerProvider<ResponseTimeDistribution, ResponseTimeDistribution, ResponseTimeDistributionMinuteRemoteWorker> {
+        public Factory(ModuleManager moduleManager, RemoteSenderService remoteSenderService, int graphId) {
+            super(moduleManager, remoteSenderService, graphId);
+        }
+
+        @Override public ResponseTimeDistributionMinuteRemoteWorker workerInstance(ModuleManager moduleManager) {
+            return new ResponseTimeDistributionMinuteRemoteWorker(moduleManager);
+        }
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionMonthPersistenceWorker.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorkerProvider;
+import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionMonthPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMonthPersistenceWorker extends PersistenceWorker<ResponseTimeDistribution> {
+
+    private ResponseTimeDistributionMonthPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_MONTH_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, ResponseTimeDistribution> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(IResponseTimeDistributionMonthPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<ResponseTimeDistribution, ResponseTimeDistributionMonthPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public ResponseTimeDistributionMonthPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new ResponseTimeDistributionMonthPersistenceWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+
+    @GraphComputingMetric(name = "/persistence/onWork/" + ResponseTimeDistributionTable.TABLE + "/month")
+    @Override protected void onWork(ResponseTimeDistribution input) {
+        super.onWork(input);
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionMonthTransformNode.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
+import org.apache.skywalking.apm.collector.core.graph.Next;
+import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMonthTransformNode implements NodeProcessor<ResponseTimeDistribution, ResponseTimeDistribution> {
+
+    @Override public int id() {
+        return MetricWorkerIdDefine.RESPONSE_TIME_DISTRIBUTION_MONTH_TRANSFORM_NODE_ID;
+    }
+
+    @Override
+    public void process(ResponseTimeDistribution responseTimeDistribution, Next<ResponseTimeDistribution> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.minuteToMonth(responseTimeDistribution.getTimeBucket());
+
+        ResponseTimeDistribution newResponseTimeDistribution = ResponseTimeDistributionCopy.copy(responseTimeDistribution);
+        newResponseTimeDistribution.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + responseTimeDistribution.getMetricId());
+        newResponseTimeDistribution.setTimeBucket(timeBucket);
+        next.execute(newResponseTimeDistribution);
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionSpanListener.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.global.std;
+
+import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricGraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator;
+import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.EntrySpanListener;
+import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.ExitSpanListener;
+import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener;
+import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.LocalSpanListener;
+import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListener;
+import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
+import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric;
+import org.apache.skywalking.apm.collector.core.graph.Graph;
+import org.apache.skywalking.apm.collector.core.graph.GraphManager;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionSpanListener implements FirstSpanListener, EntrySpanListener, ExitSpanListener, LocalSpanListener {
+
+    private static final Logger logger = LoggerFactory.getLogger(ResponseTimeDistributionSpanListener.class);
+
+    private long timeBucket;
+    private boolean isError = false;
+    private int entrySpanDuration = 0;
+    private int firstSpanDuration = 0;
+
+    @Override public boolean containsPoint(Point point) {
+        return Point.First.equals(point) || Point.Entry.equals(point) || Point.Exit.equals(point) || Point.Local.equals(point);
+    }
+
+    @Override public void parseEntry(SpanDecorator spanDecorator, int applicationId, int instanceId, String segmentId) {
+        isError = isError || spanDecorator.getIsError();
+        entrySpanDuration = (int)(spanDecorator.getEndTime() - spanDecorator.getStartTime());
+    }
+
+    @Override
+    public void parseFirst(SpanDecorator spanDecorator, int applicationId, int instanceId,
+        String segmentId) {
+        isError = isError || spanDecorator.getIsError();
+
+        if (spanDecorator.getStartTimeMinuteTimeBucket() == 0) {
+            long startTimeMinuteTimeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanDecorator.getStartTime());
+            spanDecorator.setStartTimeMinuteTimeBucket(startTimeMinuteTimeBucket);
+        }
+        timeBucket = spanDecorator.getStartTimeMinuteTimeBucket();
+
+        firstSpanDuration = (int)(spanDecorator.getEndTime() - spanDecorator.getStartTime());
+    }
+
+    @Override public void parseExit(SpanDecorator spanDecorator, int applicationId, int instanceId, String segmentId) {
+        isError = isError || spanDecorator.getIsError();
+    }
+
+    @Override public void parseLocal(SpanDecorator spanDecorator, int applicationId, int instanceId, String segmentId) {
+        isError = isError || spanDecorator.getIsError();
+    }
+
+    @Override public void build() {
+        int step = getStep();
+
+        ResponseTimeDistribution distribution = new ResponseTimeDistribution();
+        distribution.setMetricId(String.valueOf(step));
+        distribution.setId(timeBucket + Const.ID_SPLIT + distribution.getMetricId());
+        distribution.setStep(step);
+        distribution.setCalls(1);
+        distribution.setTimeBucket(timeBucket);
+
+        if (isError) {
+            distribution.setErrorCalls(1);
+        } else {
+            distribution.setSuccessCalls(1);
+        }
+
+        Graph<ResponseTimeDistribution> graph = GraphManager.INSTANCE.findGraph(MetricGraphIdDefine.RESPONSE_TIME_DISTRIBUTION_GRAPH_ID, ResponseTimeDistribution.class);
+        graph.start(distribution);
+        logger.debug("push to response time distribution aggregation worker, id: {}", distribution.getId());
+    }
+
+    int getStep() {
+        int abovePoint = 3000;
+        int interval = 50;
+
+        int duration;
+        if (entrySpanDuration == 0) {
+            duration = firstSpanDuration;
+        } else {
+            duration = entrySpanDuration;
+        }
+
+        if (duration > abovePoint) {
+            return abovePoint / interval;
+        } else if (duration <= interval) {
+            return 0;
+        } else {
+            return (int)Math.ceil((double)duration / (double)interval) - 1;
+        }
+    }
+
+    public static class Factory implements SpanListenerFactory {
+
+        @GraphComputingMetric(name = "/segment/parse/createSpanListeners/responseTimeDistributionSpanListener")
+        @Override public SpanListener create(ModuleManager moduleManager) {
+            return new ResponseTimeDistributionSpanListener();
+        }
+    }
+}

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/test/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/std/ResponseTimeDistributionSpanListenerTestCase.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.global.std;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.powermock.reflect.Whitebox;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionSpanListenerTestCase {
+
+    @Test
+    public void testStep() {
+        ResponseTimeDistributionSpanListener listener = new ResponseTimeDistributionSpanListener();
+
+        Whitebox.setInternalState(listener, "entrySpanDuration", 0);
+        Whitebox.setInternalState(listener, "firstSpanDuration", 200);
+        Assert.assertEquals(3, listener.getStep());
+
+        Whitebox.setInternalState(listener, "entrySpanDuration", 10);
+        Assert.assertEquals(0, listener.getStep());
+
+        Whitebox.setInternalState(listener, "entrySpanDuration", 60);
+        Assert.assertEquals(1, listener.getStep());
+
+        Whitebox.setInternalState(listener, "entrySpanDuration", 3100);
+        Assert.assertEquals(60, listener.getStep());
+    }
+}

apm-collector/apm-collector-component/client-component/src/main/java/org/apache/skywalking/apm/collector/client/NameSpace.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.client;
+
+import org.apache.skywalking.apm.collector.core.util.Const;
+
+/**
+ * @author peng-yongsheng
+ */
+public class NameSpace {
+
+    private String nameSpace = Const.EMPTY_STRING;
+
+    public String getNameSpace() {
+        return nameSpace;
+    }
+
+    public void setNameSpace(String nameSpace) {
+        this.nameSpace = nameSpace;
+    }
+}

apm-collector/apm-collector-component/client-component/src/main/java/org/apache/skywalking/apm/collector/client/elasticsearch/ElasticSearchClient.java

@@ -25,6 +25,7 @@ import java.util.List;
 import java.util.function.Consumer;
 import org.apache.skywalking.apm.collector.client.Client;
 import org.apache.skywalking.apm.collector.client.ClientException;
+import org.apache.skywalking.apm.collector.client.NameSpace;
 import org.apache.skywalking.apm.collector.core.data.CommonTable;
 import org.apache.skywalking.apm.collector.core.util.Const;
 import org.apache.skywalking.apm.collector.core.util.StringUtils;
@@ -65,18 +66,18 @@ public class ElasticSearchClient implements Client {
 
     private final String clusterNodes;
 
-    private final String namespace;
+    private final NameSpace namespace;
 
     public ElasticSearchClient(String clusterName, boolean clusterTransportSniffer,
         String clusterNodes) {
         this.clusterName = clusterName;
         this.clusterTransportSniffer = clusterTransportSniffer;
         this.clusterNodes = clusterNodes;
-        this.namespace = Const.EMPTY_STRING;
+        this.namespace = new NameSpace();
     }
 
     public ElasticSearchClient(String clusterName, boolean clusterTransportSniffer,
-        String clusterNodes, String namespace) {
+        String clusterNodes, NameSpace namespace) {
         this.clusterName = clusterName;
         this.clusterTransportSniffer = clusterTransportSniffer;
         this.clusterNodes = clusterNodes;
@@ -187,7 +188,7 @@ public class ElasticSearchClient implements Client {
     public MultiGetRequestBuilder prepareMultiGet(List<?> rows, MultiGetRowHandler rowHandler) {
         MultiGetRequestBuilder prepareMultiGet = client.prepareMultiGet();
         rowHandler.setPrepareMultiGet(prepareMultiGet);
-        rowHandler.setNamespace(namespace);
+        rowHandler.setNamespace(namespace.getNameSpace());
 
         rows.forEach(rowHandler::accept);
 
@@ -221,7 +222,7 @@ public class ElasticSearchClient implements Client {
     }
 
     private String formatIndexName(String indexName) {
-        return formatIndexName(this.namespace, indexName);
+        return formatIndexName(this.namespace.getNameSpace(), indexName);
     }
 
     private static String formatIndexName(String namespace, String indexName) {

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

@@ -97,6 +97,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.rtd.IResponseTimeDistributionDayPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionHourPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionMinutePersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionMonthPersistenceDAO;
 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;
@@ -209,6 +213,11 @@ public class StorageModule extends Module {
         classes.add(ISegmentPersistenceDAO.class);
         classes.add(IInstanceHeartBeatPersistenceDAO.class);
 
+        classes.add(IResponseTimeDistributionMinutePersistenceDAO.class);
+        classes.add(IResponseTimeDistributionHourPersistenceDAO.class);
+        classes.add(IResponseTimeDistributionDayPersistenceDAO.class);
+        classes.add(IResponseTimeDistributionMonthPersistenceDAO.class);
+
         classes.add(IApplicationMinuteMetricPersistenceDAO.class);
         classes.add(IApplicationHourMetricPersistenceDAO.class);
         classes.add(IApplicationDayMetricPersistenceDAO.class);

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/rtd/IResponseTimeDistributionDayPersistenceDAO.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.dao.rtd;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IResponseTimeDistributionDayPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends ResponseTimeDistribution> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/rtd/IResponseTimeDistributionHourPersistenceDAO.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.dao.rtd;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IResponseTimeDistributionHourPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends ResponseTimeDistribution> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/rtd/IResponseTimeDistributionMinutePersistenceDAO.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.dao.rtd;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IResponseTimeDistributionMinutePersistenceDAO<INSERT, UPDATE, STREAM_DATA extends ResponseTimeDistribution> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/rtd/IResponseTimeDistributionMonthPersistenceDAO.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.dao.rtd;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IResponseTimeDistributionMonthPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends ResponseTimeDistribution> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/global/ResponseTimeDistribution.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.table.global;
+
+import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
+import org.apache.skywalking.apm.collector.core.data.StreamData;
+import org.apache.skywalking.apm.collector.core.data.operator.AddMergeOperation;
+import org.apache.skywalking.apm.collector.core.data.operator.CoverMergeOperation;
+import org.apache.skywalking.apm.collector.core.data.operator.NonMergeOperation;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistribution extends StreamData {
+
+    private static final Column[] STRING_COLUMNS = {
+        new Column(ResponseTimeDistributionTable.ID, new NonMergeOperation()),
+        new Column(ResponseTimeDistributionTable.METRIC_ID, new NonMergeOperation()),
+    };
+
+    private static final Column[] LONG_COLUMNS = {
+        new Column(ResponseTimeDistributionTable.TIME_BUCKET, new CoverMergeOperation()),
+        new Column(ResponseTimeDistributionTable.CALLS, new AddMergeOperation()),
+        new Column(ResponseTimeDistributionTable.ERROR_CALLS, new AddMergeOperation()),
+        new Column(ResponseTimeDistributionTable.SUCCESS_CALLS, new AddMergeOperation()),
+    };
+
+    private static final Column[] DOUBLE_COLUMNS = {};
+
+    private static final Column[] INTEGER_COLUMNS = {
+        new Column(ResponseTimeDistributionTable.STEP, new NonMergeOperation()),
+    };
+
+    private static final Column[] BYTE_COLUMNS = {};
+
+    public ResponseTimeDistribution() {
+        super(STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_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);
+    }
+
+    public int getStep() {
+        return getDataInteger(0);
+    }
+
+    public void setStep(int step) {
+        setDataInteger(0, step);
+    }
+
+    public long getTimeBucket() {
+        return getDataLong(0);
+    }
+
+    public void setTimeBucket(long timeBucket) {
+        setDataLong(0, timeBucket);
+    }
+
+    public long getCalls() {
+        return getDataLong(1);
+    }
+
+    public void setCalls(long calls) {
+        setDataLong(1, calls);
+    }
+
+    public long getErrorCalls() {
+        return getDataLong(2);
+    }
+
+    public void setErrorCalls(long errorCalls) {
+        setDataLong(2, errorCalls);
+    }
+
+    public long getSuccessCalls() {
+        return getDataLong(3);
+    }
+
+    public void setSuccessCalls(long successCalls) {
+        setDataLong(3, successCalls);
+    }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ResponseTimeDistribution();
+        }
+    }
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/global/ResponseTimeDistributionTable.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.table.global;
+
+import org.apache.skywalking.apm.collector.core.data.ColumnName;
+import org.apache.skywalking.apm.collector.core.data.CommonTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface ResponseTimeDistributionTable extends CommonTable {
+    String TABLE = "response_time_distribution";
+
+    ColumnName STEP = new ColumnName("step", "s");
+
+    ColumnName CALLS = new ColumnName("calls", "ac");
+
+    ColumnName ERROR_CALLS = new ColumnName("error_calls", "ec");
+
+    ColumnName SUCCESS_CALLS = new ColumnName("success_calls", "sc");
+}
\ No newline at end of file

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

@@ -20,6 +20,7 @@ package org.apache.skywalking.apm.collector.storage.es;
 
 import java.util.UUID;
 import org.apache.skywalking.apm.collector.client.ClientException;
+import org.apache.skywalking.apm.collector.client.NameSpace;
 import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
 import org.apache.skywalking.apm.collector.cluster.ClusterModule;
 import org.apache.skywalking.apm.collector.cluster.service.ModuleListenerService;
@@ -110,6 +111,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.rtd.IResponseTimeDistributionDayPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionHourPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionMinutePersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionMonthPersistenceDAO;
 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;
@@ -215,6 +220,10 @@ import org.apache.skywalking.apm.collector.storage.es.dao.register.ApplicationRe
 import org.apache.skywalking.apm.collector.storage.es.dao.register.InstanceRegisterEsDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.register.NetworkAddressRegisterEsDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.register.ServiceNameRegisterEsDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.rtd.ResponseTimeDistributionDayEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.rtd.ResponseTimeDistributionHourEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.rtd.ResponseTimeDistributionMinuteEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.rtd.ResponseTimeDistributionMonthEsPersistenceDAO;
 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;
@@ -251,12 +260,14 @@ public class StorageModuleEsProvider extends ModuleProvider {
 
     static final String NAME = "elasticsearch";
     private final StorageModuleEsConfig config;
+    private final NameSpace nameSpace;
     private ElasticSearchClient elasticSearchClient;
     private DataTTLKeeperTimer deleteTimer;
 
     public StorageModuleEsProvider() {
         super();
         this.config = new StorageModuleEsConfig();
+        this.nameSpace = new NameSpace();
     }
 
     @Override public String name() {
@@ -272,6 +283,8 @@ public class StorageModuleEsProvider extends ModuleProvider {
     }
 
     @Override public void prepare() throws ServiceNotProvidedException {
+        elasticSearchClient = new ElasticSearchClient(config.getClusterName(), config.getClusterTransportSniffer(), config.getClusterNodes(), nameSpace);
+
         this.registerServiceImplementation(IBatchDAO.class, new BatchEsDAO(elasticSearchClient));
         registerCacheDAO();
         registerRegisterDAO();
@@ -284,7 +297,7 @@ public class StorageModuleEsProvider extends ModuleProvider {
     public void start() throws ModuleStartException {
         try {
             String namespace = getManager().find(ConfigurationModule.NAME).getService(ICollectorConfig.class).getNamespace();
-            elasticSearchClient = new ElasticSearchClient(config.getClusterName(), config.getClusterTransportSniffer(), config.getClusterNodes(), namespace);
+            nameSpace.setNameSpace(namespace);
             elasticSearchClient.initialize();
 
             ElasticSearchStorageInstaller installer = new ElasticSearchStorageInstaller(config.getIndexShardsNumber(), config.getIndexReplicasNumber(), config.isHighPerformanceMode());
@@ -371,6 +384,11 @@ public class StorageModuleEsProvider extends ModuleProvider {
 
         this.registerServiceImplementation(IGlobalTracePersistenceDAO.class, new GlobalTraceEsPersistenceDAO(elasticSearchClient));
 
+        this.registerServiceImplementation(IResponseTimeDistributionMinutePersistenceDAO.class, new ResponseTimeDistributionMinuteEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IResponseTimeDistributionHourPersistenceDAO.class, new ResponseTimeDistributionHourEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IResponseTimeDistributionDayPersistenceDAO.class, new ResponseTimeDistributionDayEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IResponseTimeDistributionMonthPersistenceDAO.class, new ResponseTimeDistributionMonthEsPersistenceDAO(elasticSearchClient));
+
         this.registerServiceImplementation(IApplicationMinuteMetricPersistenceDAO.class, new ApplicationMinuteMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IApplicationHourMetricPersistenceDAO.class, new ApplicationHourMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IApplicationDayMetricPersistenceDAO.class, new ApplicationDayMetricEsPersistenceDAO(elasticSearchClient));

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/rtd/AbstractResponseTimeDistributionEsPersistenceDAO.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.dao.rtd;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric;
+import org.apache.skywalking.apm.collector.storage.es.base.dao.AbstractPersistenceEsDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public abstract class AbstractResponseTimeDistributionEsPersistenceDAO extends AbstractPersistenceEsDAO<ResponseTimeDistribution> {
+
+    AbstractResponseTimeDistributionEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected final String timeBucketColumnNameForDelete() {
+        return ResponseTimeDistributionTable.TIME_BUCKET.getName();
+    }
+
+    @Override protected final ResponseTimeDistribution esDataToStreamData(Map<String, Object> source) {
+        ResponseTimeDistribution responseTimeDistribution = new ResponseTimeDistribution();
+        responseTimeDistribution.setMetricId((String)source.get(ResponseTimeDistributionTable.METRIC_ID.getName()));
+
+        responseTimeDistribution.setStep(((Number)source.get(ResponseTimeDistributionTable.STEP.getName())).intValue());
+
+        responseTimeDistribution.setCalls(((Number)source.get(ResponseTimeDistributionTable.CALLS.getName())).longValue());
+        responseTimeDistribution.setErrorCalls(((Number)source.get(ResponseTimeDistributionTable.ERROR_CALLS.getName())).longValue());
+        responseTimeDistribution.setSuccessCalls(((Number)source.get(ResponseTimeDistributionTable.SUCCESS_CALLS.getName())).longValue());
+
+        responseTimeDistribution.setTimeBucket(((Number)source.get(ResponseTimeDistributionTable.TIME_BUCKET.getName())).longValue());
+        return responseTimeDistribution;
+    }
+
+    @Override protected final Map<String, Object> esStreamDataToEsData(ResponseTimeDistribution streamData) {
+        Map<String, Object> target = new HashMap<>();
+        target.put(ResponseTimeDistributionTable.METRIC_ID.getName(), streamData.getMetricId());
+
+        target.put(ResponseTimeDistributionTable.STEP.getName(), streamData.getStep());
+
+        target.put(ResponseTimeDistributionTable.CALLS.getName(), streamData.getCalls());
+        target.put(ResponseTimeDistributionTable.ERROR_CALLS.getName(), streamData.getErrorCalls());
+        target.put(ResponseTimeDistributionTable.SUCCESS_CALLS.getName(), streamData.getSuccessCalls());
+
+        target.put(ResponseTimeDistributionTable.TIME_BUCKET.getName(), streamData.getTimeBucket());
+
+        return target;
+    }
+
+    @GraphComputingMetric(name = "/persistence/get/" + ResponseTimeDistributionTable.TABLE)
+    @Override public final ResponseTimeDistribution get(String id) {
+        return super.get(id);
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/rtd/ResponseTimeDistributionDayEsPersistenceDAO.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.dao.rtd;
+
+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.rtd.IResponseTimeDistributionDayPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionDayEsPersistenceDAO extends AbstractResponseTimeDistributionEsPersistenceDAO implements IResponseTimeDistributionDayPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ResponseTimeDistribution> {
+
+    public ResponseTimeDistributionDayEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Day.getName();
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/rtd/ResponseTimeDistributionHourEsPersistenceDAO.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.dao.rtd;
+
+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.rtd.IResponseTimeDistributionHourPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionHourEsPersistenceDAO extends AbstractResponseTimeDistributionEsPersistenceDAO implements IResponseTimeDistributionHourPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ResponseTimeDistribution> {
+
+    public ResponseTimeDistributionHourEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Hour.getName();
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/rtd/ResponseTimeDistributionMinuteEsPersistenceDAO.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.dao.rtd;
+
+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.rtd.IResponseTimeDistributionMinutePersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMinuteEsPersistenceDAO extends AbstractResponseTimeDistributionEsPersistenceDAO implements IResponseTimeDistributionMinutePersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ResponseTimeDistribution> {
+
+    public ResponseTimeDistributionMinuteEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Minute.getName();
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/rtd/ResponseTimeDistributionMonthEsPersistenceDAO.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.dao.rtd;
+
+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.rtd.IResponseTimeDistributionMonthPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMonthEsPersistenceDAO extends AbstractResponseTimeDistributionEsPersistenceDAO implements IResponseTimeDistributionMonthPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, ResponseTimeDistribution> {
+
+    public ResponseTimeDistributionMonthEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Month.getName();
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/rtd/AbstractResponseTimeDistributionEsTableDefine.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.rtd;
+
+import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchColumnDefine;
+import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchTableDefine;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public abstract class AbstractResponseTimeDistributionEsTableDefine extends ElasticSearchTableDefine {
+
+    AbstractResponseTimeDistributionEsTableDefine(String name) {
+        super(name);
+    }
+
+    @Override public final void initialize() {
+        addColumn(new ElasticSearchColumnDefine(ResponseTimeDistributionTable.METRIC_ID, ElasticSearchColumnDefine.Type.Keyword.name()));
+        addColumn(new ElasticSearchColumnDefine(ResponseTimeDistributionTable.STEP, ElasticSearchColumnDefine.Type.Integer.name()));
+        addColumn(new ElasticSearchColumnDefine(ResponseTimeDistributionTable.CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(ResponseTimeDistributionTable.ERROR_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(ResponseTimeDistributionTable.SUCCESS_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(ResponseTimeDistributionTable.TIME_BUCKET, ElasticSearchColumnDefine.Type.Long.name()));
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/rtd/ResponseTimeDistributionDayEsTableDefine.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.rtd;
+
+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.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionDayEsTableDefine extends AbstractResponseTimeDistributionEsTableDefine {
+
+    public ResponseTimeDistributionDayEsTableDefine() {
+        super(ResponseTimeDistributionTable.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/rtd/ResponseTimeDistributionHourEsTableDefine.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.rtd;
+
+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.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionHourEsTableDefine extends AbstractResponseTimeDistributionEsTableDefine {
+
+    public ResponseTimeDistributionHourEsTableDefine() {
+        super(ResponseTimeDistributionTable.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/rtd/ResponseTimeDistributionMinuteEsTableDefine.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.rtd;
+
+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.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMinuteEsTableDefine extends AbstractResponseTimeDistributionEsTableDefine {
+
+    public ResponseTimeDistributionMinuteEsTableDefine() {
+        super(ResponseTimeDistributionTable.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/rtd/ResponseTimeDistributionMonthEsTableDefine.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.rtd;
+
+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.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMonthEsTableDefine extends AbstractResponseTimeDistributionEsTableDefine {
+
+    public ResponseTimeDistributionMonthEsTableDefine() {
+        super(ResponseTimeDistributionTable.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

@@ -66,6 +66,11 @@ org.apache.skywalking.apm.collector.storage.es.define.amp.ApplicationHourMetricE
 org.apache.skywalking.apm.collector.storage.es.define.amp.ApplicationDayMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.amp.ApplicationMonthMetricEsTableDefine
 
+org.apache.skywalking.apm.collector.storage.es.define.rtd.ResponseTimeDistributionMinuteEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.rtd.ResponseTimeDistributionHourEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.rtd.ResponseTimeDistributionDayEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.rtd.ResponseTimeDistributionMonthEsTableDefine
+
 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.SegmentDurationEsTableDefine

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java

@@ -103,6 +103,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.rtd.IResponseTimeDistributionDayPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionHourPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionMinutePersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.rtd.IResponseTimeDistributionMonthPersistenceDAO;
 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;
@@ -208,6 +212,10 @@ import org.apache.skywalking.apm.collector.storage.h2.dao.register.ApplicationRe
 import org.apache.skywalking.apm.collector.storage.h2.dao.register.InstanceRegisterH2DAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.register.NetworkAddressRegisterH2DAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.register.ServiceNameRegisterH2DAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.rtd.ResponseTimeDistributionDayH2PersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.rtd.ResponseTimeDistributionHourH2PersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.rtd.ResponseTimeDistributionMinuteH2PersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.rtd.ResponseTimeDistributionMonthH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.smp.ServiceDayMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.smp.ServiceHourMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.smp.ServiceMinuteMetricH2PersistenceDAO;
@@ -334,6 +342,12 @@ public class StorageModuleH2Provider extends ModuleProvider {
         this.registerServiceImplementation(IMemoryPoolMonthMetricPersistenceDAO.class, new MemoryPoolMonthMetricH2PersistenceDAO(h2Client));
 
         this.registerServiceImplementation(IGlobalTracePersistenceDAO.class, new GlobalTraceH2PersistenceDAO(h2Client));
+
+        this.registerServiceImplementation(IResponseTimeDistributionMinutePersistenceDAO.class, new ResponseTimeDistributionMinuteH2PersistenceDAO(h2Client));
+        this.registerServiceImplementation(IResponseTimeDistributionHourPersistenceDAO.class, new ResponseTimeDistributionHourH2PersistenceDAO(h2Client));
+        this.registerServiceImplementation(IResponseTimeDistributionDayPersistenceDAO.class, new ResponseTimeDistributionDayH2PersistenceDAO(h2Client));
+        this.registerServiceImplementation(IResponseTimeDistributionMonthPersistenceDAO.class, new ResponseTimeDistributionMonthH2PersistenceDAO(h2Client));
+
         this.registerServiceImplementation(ISegmentDurationPersistenceDAO.class, new SegmentDurationH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(ISegmentPersistenceDAO.class, new SegmentH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IInstanceHeartBeatPersistenceDAO.class, new InstanceHeartBeatH2PersistenceDAO(h2Client));

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/rtd/AbstractResponseTimeDistributionH2PersistenceDAO.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.h2.dao.rtd;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.skywalking.apm.collector.client.h2.H2Client;
+import org.apache.skywalking.apm.collector.storage.h2.base.dao.AbstractPersistenceH2DAO;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public abstract class AbstractResponseTimeDistributionH2PersistenceDAO extends AbstractPersistenceH2DAO<ResponseTimeDistribution> {
+
+    AbstractResponseTimeDistributionH2PersistenceDAO(H2Client client) {
+        super(client);
+    }
+
+    @Override protected final ResponseTimeDistribution h2DataToStreamData(ResultSet resultSet) throws SQLException {
+        ResponseTimeDistribution responseTimeDistribution = new ResponseTimeDistribution();
+        responseTimeDistribution.setMetricId(resultSet.getString(ResponseTimeDistributionTable.METRIC_ID.getName()));
+
+        responseTimeDistribution.setStep(resultSet.getInt(ResponseTimeDistributionTable.STEP.getName()));
+
+        responseTimeDistribution.setCalls(resultSet.getLong(ResponseTimeDistributionTable.CALLS.getName()));
+        responseTimeDistribution.setErrorCalls(resultSet.getLong(ResponseTimeDistributionTable.ERROR_CALLS.getName()));
+        responseTimeDistribution.setSuccessCalls(resultSet.getLong(ResponseTimeDistributionTable.SUCCESS_CALLS.getName()));
+
+        responseTimeDistribution.setTimeBucket(resultSet.getLong(ResponseTimeDistributionTable.TIME_BUCKET.getName()));
+
+        return responseTimeDistribution;
+    }
+
+    @Override protected final Map<String, Object> streamDataToH2Data(ResponseTimeDistribution streamData) {
+        Map<String, Object> target = new HashMap<>();
+        target.put(ResponseTimeDistributionTable.ID.getName(), streamData.getId());
+        target.put(ResponseTimeDistributionTable.METRIC_ID.getName(), streamData.getMetricId());
+
+        target.put(ResponseTimeDistributionTable.STEP.getName(), streamData.getStep());
+
+        target.put(ResponseTimeDistributionTable.CALLS.getName(), streamData.getCalls());
+        target.put(ResponseTimeDistributionTable.ERROR_CALLS.getName(), streamData.getErrorCalls());
+        target.put(ResponseTimeDistributionTable.SUCCESS_CALLS.getName(), streamData.getSuccessCalls());
+
+        target.put(ResponseTimeDistributionTable.TIME_BUCKET.getName(), streamData.getTimeBucket());
+
+        return target;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/rtd/ResponseTimeDistributionDayH2PersistenceDAO.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.h2.dao.rtd;
+
+import org.apache.skywalking.apm.collector.client.h2.H2Client;
+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.rtd.IResponseTimeDistributionDayPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionDayH2PersistenceDAO extends AbstractResponseTimeDistributionH2PersistenceDAO implements IResponseTimeDistributionDayPersistenceDAO<H2SqlEntity, H2SqlEntity, ResponseTimeDistribution> {
+
+    public ResponseTimeDistributionDayH2PersistenceDAO(H2Client client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Day.getName();
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/rtd/ResponseTimeDistributionHourH2PersistenceDAO.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.h2.dao.rtd;
+
+import org.apache.skywalking.apm.collector.client.h2.H2Client;
+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.rtd.IResponseTimeDistributionHourPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionHourH2PersistenceDAO extends AbstractResponseTimeDistributionH2PersistenceDAO implements IResponseTimeDistributionHourPersistenceDAO<H2SqlEntity, H2SqlEntity, ResponseTimeDistribution> {
+
+    public ResponseTimeDistributionHourH2PersistenceDAO(H2Client client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Hour.getName();
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/rtd/ResponseTimeDistributionMinuteH2PersistenceDAO.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.h2.dao.rtd;
+
+import org.apache.skywalking.apm.collector.client.h2.H2Client;
+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.rtd.IResponseTimeDistributionMinutePersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMinuteH2PersistenceDAO extends AbstractResponseTimeDistributionH2PersistenceDAO implements IResponseTimeDistributionMinutePersistenceDAO<H2SqlEntity, H2SqlEntity, ResponseTimeDistribution> {
+
+    public ResponseTimeDistributionMinuteH2PersistenceDAO(H2Client client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Minute.getName();
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/rtd/ResponseTimeDistributionMonthH2PersistenceDAO.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.h2.dao.rtd;
+
+import org.apache.skywalking.apm.collector.client.h2.H2Client;
+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.rtd.IResponseTimeDistributionMonthPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistribution;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMonthH2PersistenceDAO extends AbstractResponseTimeDistributionH2PersistenceDAO implements IResponseTimeDistributionMonthPersistenceDAO<H2SqlEntity, H2SqlEntity, ResponseTimeDistribution> {
+
+    public ResponseTimeDistributionMonthH2PersistenceDAO(H2Client client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Month.getName();
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/rtd/AbstractResponseTimeDistributionH2TableDefine.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.h2.define.rtd;
+
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2ColumnDefine;
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2TableDefine;
+import org.apache.skywalking.apm.collector.storage.table.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public abstract class AbstractResponseTimeDistributionH2TableDefine extends H2TableDefine {
+
+    AbstractResponseTimeDistributionH2TableDefine(String name) {
+        super(name);
+    }
+
+    @Override public final void initialize() {
+        addColumn(new H2ColumnDefine(ResponseTimeDistributionTable.ID, H2ColumnDefine.Type.Varchar.name()));
+        addColumn(new H2ColumnDefine(ResponseTimeDistributionTable.METRIC_ID, H2ColumnDefine.Type.Varchar.name()));
+        addColumn(new H2ColumnDefine(ResponseTimeDistributionTable.STEP, H2ColumnDefine.Type.Int.name()));
+        addColumn(new H2ColumnDefine(ResponseTimeDistributionTable.CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(ResponseTimeDistributionTable.ERROR_CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(ResponseTimeDistributionTable.SUCCESS_CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(ResponseTimeDistributionTable.TIME_BUCKET, H2ColumnDefine.Type.Bigint.name()));
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/rtd/ResponseTimeDistributionDayH2TableDefine.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.h2.define.rtd;
+
+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.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionDayH2TableDefine extends AbstractResponseTimeDistributionH2TableDefine {
+
+    public ResponseTimeDistributionDayH2TableDefine() {
+        super(ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Day.getName());
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/rtd/ResponseTimeDistributionHourH2TableDefine.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.h2.define.rtd;
+
+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.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionHourH2TableDefine extends AbstractResponseTimeDistributionH2TableDefine {
+
+    public ResponseTimeDistributionHourH2TableDefine() {
+        super(ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Hour.getName());
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/rtd/ResponseTimeDistributionMinuteH2TableDefine.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.h2.define.rtd;
+
+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.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMinuteH2TableDefine extends AbstractResponseTimeDistributionH2TableDefine {
+
+    public ResponseTimeDistributionMinuteH2TableDefine() {
+        super(ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Minute.getName());
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/rtd/ResponseTimeDistributionMonthH2TableDefine.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.h2.define.rtd;
+
+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.global.ResponseTimeDistributionTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ResponseTimeDistributionMonthH2TableDefine extends AbstractResponseTimeDistributionH2TableDefine {
+
+    public ResponseTimeDistributionMonthH2TableDefine() {
+        super(ResponseTimeDistributionTable.TABLE + Const.ID_SPLIT + TimePyramid.Month.getName());
+    }
+}

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

@@ -61,6 +61,12 @@ org.apache.skywalking.apm.collector.storage.h2.define.impp.InstanceMappingDayH2T
 org.apache.skywalking.apm.collector.storage.h2.define.impp.InstanceMappingMonthH2TableDefine
 
 org.apache.skywalking.apm.collector.storage.h2.define.GlobalTraceH2TableDefine
+
+org.apache.skywalking.apm.collector.storage.h2.define.rtd.ResponseTimeDistributionMinuteH2TableDefine
+org.apache.skywalking.apm.collector.storage.h2.define.rtd.ResponseTimeDistributionHourH2TableDefine
+org.apache.skywalking.apm.collector.storage.h2.define.rtd.ResponseTimeDistributionDayH2TableDefine
+org.apache.skywalking.apm.collector.storage.h2.define.rtd.ResponseTimeDistributionMonthH2TableDefine
+
 org.apache.skywalking.apm.collector.storage.h2.define.SegmentDurationH2TableDefine
 org.apache.skywalking.apm.collector.storage.h2.define.SegmentH2TableDefine