[#3307] aggregate agent's response time to show application's respons…

…e time

collector/src/main/java/com/navercorp/pinpoint/collector/mapper/thrift/stat/TFAgentStatMapper.java

@@ -29,6 +29,7 @@ public class TFAgentStatMapper {
     private static final TFJvmGcMapper tFJvmGcMapper = new TFJvmGcMapper();
     private static final TFTransactionMapper tFTransactionMapper = new TFTransactionMapper();
     private static final TFActiveTraceMapper tFActiveTraceMapper = new TFActiveTraceMapper();
+    private static final TFResponseTimeMapper tFResponseTimeMapper = new TFResponseTimeMapper();
 
     public List<TFAgentStat> map(AgentStatBo agentStatBo) {
         final TreeMap<Long, TFAgentStat> tFAgentStatMap = new TreeMap<>();
@@ -39,9 +40,22 @@ public List<TFAgentStat> map(AgentStatBo agentStatBo) {
         insertTFJvmGc(tFAgentStatMap, agentStatBo.getJvmGcBos(), agentId, startTimestamp);
         insertTFTransaction(tFAgentStatMap, agentStatBo.getTransactionBos(), agentId, startTimestamp);
         insertTFActiveTrace(tFAgentStatMap, agentStatBo.getActiveTraceBos(), agentId, startTimestamp);
+        insertTFResponseTime(tFAgentStatMap, agentStatBo.getResponseTimeBos(), agentId, startTimestamp);
         return new ArrayList<>(tFAgentStatMap.values());
     }
 
+    private void insertTFResponseTime(TreeMap<Long, TFAgentStat> tFAgentStatMap, List<ResponseTimeBo> responseTimeBoList, String agentId, long startTimestamp) {
+        if (responseTimeBoList == null) {
+            return;
+        }
+
+        for (ResponseTimeBo responseTimeBo : responseTimeBoList) {
+            TFAgentStat tFAgentStat = getOrCreateTFAgentStat(tFAgentStatMap, responseTimeBo.getTimestamp(), agentId, startTimestamp);
+            tFAgentStat.setResponseTime(tFResponseTimeMapper.map(responseTimeBo));
+        }
+
+    }
+
     private void insertTFActiveTrace(TreeMap<Long, TFAgentStat> tFAgentStatMap, List<ActiveTraceBo> activeTraceBoList, String agentId, long startTimestamp) {
         if (activeTraceBoList == null) {
             return;

collector/src/main/java/com/navercorp/pinpoint/collector/mapper/thrift/stat/TFResponseTimeMapper.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2017 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.collector.mapper.thrift.stat;
+
+import com.navercorp.pinpoint.common.server.bo.stat.ResponseTimeBo;
+import com.navercorp.pinpoint.thrift.dto.flink.TFResponseTime;
+
+/**
+ * @author minwoo.jung
+ */
+public class TFResponseTimeMapper {
+
+    public TFResponseTime map(ResponseTimeBo responseTimeBo) {
+        TFResponseTime tFResponseTime = new TFResponseTime();
+        tFResponseTime.setAvg(responseTimeBo.getAvg());
+        return tFResponseTime;
+    }
+}

collector/src/test/java/com/navercorp/pinpoint/collector/mapper/thrift/stat/TFCpuLoadMapperTest.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2017 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.collector.mapper.thrift.stat;
+
+import com.navercorp.pinpoint.common.server.bo.stat.CpuLoadBo;
+import com.navercorp.pinpoint.thrift.dto.flink.TFCpuLoad;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+/**
+ * @author minwoo.jung
+ */
+public class TFCpuLoadMapperTest {
+    @Test
+    public void mapTest() throws Exception {
+        TFCpuLoadMapper tFCpuLoadMapper = new TFCpuLoadMapper();
+        CpuLoadBo cpuLoadBo = new CpuLoadBo();
+        cpuLoadBo.setJvmCpuLoad(30);
+        cpuLoadBo.setSystemCpuLoad(50);
+        TFCpuLoad tFCpuLoad = tFCpuLoadMapper.map(cpuLoadBo);
+        assertEquals(tFCpuLoad.getJvmCpuLoad(), 30, 0);
+        assertEquals(tFCpuLoad.getSystemCpuLoad(), 50, 0);
+    }
+
+}

collector/src/test/java/com/navercorp/pinpoint/collector/mapper/thrift/stat/TFJvmGcMapperTest.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2017 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.collector.mapper.thrift.stat;
+
+import com.navercorp.pinpoint.common.server.bo.stat.JvmGcBo;
+import com.navercorp.pinpoint.thrift.dto.flink.TFJvmGc;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+/**
+ * @author minwoo.jung
+ */
+public class TFJvmGcMapperTest {
+
+    @Test
+    public void mapTest() throws Exception {
+        TFJvmGcMapper tFJvmGcMapper = new TFJvmGcMapper();
+        JvmGcBo jvmGcBo = new JvmGcBo();
+        jvmGcBo.setHeapUsed(3000);
+        jvmGcBo.setNonHeapUsed(500);
+        TFJvmGc tFJvmGc = tFJvmGcMapper.map(jvmGcBo);
+
+        assertEquals(tFJvmGc.getJvmMemoryHeapUsed(), 3000);
+        assertEquals(tFJvmGc.getJvmMemoryNonHeapUsed(), 500);
+    }
+
+}

collector/src/test/java/com/navercorp/pinpoint/collector/mapper/thrift/stat/TFResponseTimeMapperTest.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2017 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.collector.mapper.thrift.stat;
+
+import com.navercorp.pinpoint.common.server.bo.stat.ResponseTimeBo;
+import com.navercorp.pinpoint.thrift.dto.flink.TFResponseTime;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+/**
+ * @author minwoo.jung
+ */
+public class TFResponseTimeMapperTest {
+
+    @Test
+    public void mapTest() {
+        TFResponseTimeMapper mapper = new TFResponseTimeMapper();
+        ResponseTimeBo responseTimeBo = new ResponseTimeBo();
+        responseTimeBo.setAvg(50);
+        TFResponseTime tFResponseTime = mapper.map(responseTimeBo);
+        assertEquals(tFResponseTime.getAvg(), 50);
+    }
+}

collector/src/test/java/com/navercorp/pinpoint/collector/mapper/thrift/stat/TFTransactionMapperTest.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2017 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.collector.mapper.thrift.stat;
+
+import com.navercorp.pinpoint.common.server.bo.stat.TransactionBo;
+import com.navercorp.pinpoint.thrift.dto.flink.TFTransaction;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+/**
+ * @author minwoo.jung
+ */
+public class TFTransactionMapperTest {
+
+    @Test
+    public void mapTest() throws Exception {
+        TFTransactionMapper tFTransactionMapper = new TFTransactionMapper();
+        TransactionBo transactionBo = new TransactionBo();
+        transactionBo.setSampledContinuationCount(5);
+        transactionBo.setUnsampledContinuationCount(6);
+        transactionBo.setSampledNewCount(11);
+        transactionBo.setUnsampledNewCount(10);
+        TFTransaction tFtransaction = tFTransactionMapper.map(transactionBo);
+
+        assertEquals(tFtransaction.getSampledContinuationCount(), 5);
+        assertEquals(tFtransaction.getUnsampledContinuationCount(), 6);
+        assertEquals(tFtransaction.getSampledNewCount(), 11);
+        assertEquals(tFtransaction.getUnsampledNewCount(), 10);
+    }
+
+}

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/codec/stat/join/ResponseTimeCodec.java

@@ -0,0 +1,146 @@
+/*
+ * Copyright 2017 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.common.server.bo.codec.stat.join;
+
+import com.navercorp.pinpoint.common.buffer.Buffer;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.AgentStatDataPointCodec;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.ApplicationStatCodec;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.header.AgentStatHeaderDecoder;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.header.AgentStatHeaderEncoder;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.header.BitCountingHeaderDecoder;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.header.BitCountingHeaderEncoder;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.strategy.StrategyAnalyzer;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.strategy.StringEncodingStrategy;
+import com.navercorp.pinpoint.common.server.bo.codec.stat.strategy.UnsignedLongEncodingStrategy;
+import com.navercorp.pinpoint.common.server.bo.codec.strategy.EncodingStrategy;
+import com.navercorp.pinpoint.common.server.bo.serializer.stat.ApplicationStatDecodingContext;
+import com.navercorp.pinpoint.common.server.bo.stat.join.JoinResponseTimeBo;
+import com.navercorp.pinpoint.common.server.bo.stat.join.JoinStatBo;
+import com.navercorp.pinpoint.common.util.CollectionUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.util.Assert;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author minwoo.jung
+ */
+@Component("joinResponseTimeCodec")
+public class ResponseTimeCodec implements ApplicationStatCodec {
+    private static final byte VERSION = 1;
+
+    private final AgentStatDataPointCodec codec;
+
+    @Autowired
+    public ResponseTimeCodec(AgentStatDataPointCodec codec) {
+        Assert.notNull(codec, "agentStatDataPointCodec must not be null");
+        this.codec = codec;
+    }
+
+
+    @Override
+    public byte getVersion() {
+        return VERSION;
+    }
+
+    @Override
+    public void encodeValues(Buffer valueBuffer, List<JoinStatBo> joinResponseTimeBoList) {
+        if (CollectionUtils.isEmpty(joinResponseTimeBoList)) {
+            throw new IllegalArgumentException("joinResponseTimeBoList must not be empty");
+        }
+
+        final int numValues = joinResponseTimeBoList.size();
+        valueBuffer.putVInt(numValues);
+        List<Long> timestamps = new ArrayList<Long>(numValues);
+        UnsignedLongEncodingStrategy.Analyzer.Builder avgAnalyzerBuilder = new UnsignedLongEncodingStrategy.Analyzer.Builder();
+        UnsignedLongEncodingStrategy.Analyzer.Builder minAvgAnalyzerBuilder = new UnsignedLongEncodingStrategy.Analyzer.Builder();
+        StringEncodingStrategy.Analyzer.Builder minAvgAgentIdAnalyzerBuilder = new StringEncodingStrategy.Analyzer.Builder();
+        UnsignedLongEncodingStrategy.Analyzer.Builder maxAvgAnalyzerBuilder = new UnsignedLongEncodingStrategy.Analyzer.Builder();
+        StringEncodingStrategy.Analyzer.Builder maxAvgAgentIdAnalyzerBuilder = new StringEncodingStrategy.Analyzer.Builder();
+
+        for (JoinStatBo joinStatBo : joinResponseTimeBoList) {
+            JoinResponseTimeBo joinResponseTimeBo = (JoinResponseTimeBo) joinStatBo;
+            timestamps.add(joinResponseTimeBo.getTimestamp());
+            avgAnalyzerBuilder.addValue(joinResponseTimeBo.getAvg());
+            minAvgAnalyzerBuilder.addValue(joinResponseTimeBo.getMinAvg());
+            minAvgAgentIdAnalyzerBuilder.addValue(joinResponseTimeBo.getMinAvgAgentId());
+            maxAvgAnalyzerBuilder.addValue(joinResponseTimeBo.getMaxAvg());
+            maxAvgAgentIdAnalyzerBuilder.addValue(joinResponseTimeBo.getMaxAvgAgentId());
+        }
+
+        codec.encodeTimestamps(valueBuffer, timestamps);
+        encodeDataPoints(valueBuffer, avgAnalyzerBuilder.build(), minAvgAnalyzerBuilder.build(), minAvgAgentIdAnalyzerBuilder.build(), maxAvgAnalyzerBuilder.build(), maxAvgAgentIdAnalyzerBuilder.build());
+    }
+
+    private void encodeDataPoints(Buffer valueBuffer, StrategyAnalyzer<Long> avgAnalyzer, StrategyAnalyzer<Long> minAvgAnalyzer, StrategyAnalyzer<String> minAvgAgentIdAnalyzer, StrategyAnalyzer<Long> maxAvgAnalyzer, StrategyAnalyzer<String> maxAvgAgentIdAnalyzer) {
+        AgentStatHeaderEncoder headerEncoder = new BitCountingHeaderEncoder();
+        headerEncoder.addCode(avgAnalyzer.getBestStrategy().getCode());
+        headerEncoder.addCode(minAvgAnalyzer.getBestStrategy().getCode());
+        headerEncoder.addCode(minAvgAgentIdAnalyzer.getBestStrategy().getCode());
+        headerEncoder.addCode(maxAvgAnalyzer.getBestStrategy().getCode());
+        headerEncoder.addCode(maxAvgAgentIdAnalyzer.getBestStrategy().getCode());
+        final byte[] header = headerEncoder.getHeader();
+        valueBuffer.putPrefixedBytes(header);
+
+        this.codec.encodeValues(valueBuffer, avgAnalyzer.getBestStrategy(), avgAnalyzer.getValues());
+        this.codec.encodeValues(valueBuffer, minAvgAnalyzer.getBestStrategy(), minAvgAnalyzer.getValues());
+        this.codec.encodeValues(valueBuffer, minAvgAgentIdAnalyzer.getBestStrategy(), minAvgAgentIdAnalyzer.getValues());
+        this.codec.encodeValues(valueBuffer, maxAvgAnalyzer.getBestStrategy(), maxAvgAnalyzer.getValues());
+        this.codec.encodeValues(valueBuffer, maxAvgAgentIdAnalyzer.getBestStrategy(), maxAvgAgentIdAnalyzer.getValues());
+    }
+
+    @Override
+    public List<JoinStatBo> decodeValues(Buffer valueBuffer, ApplicationStatDecodingContext decodingContext) {
+        final String id = decodingContext.getApplicationId();
+        final long baseTimestamp = decodingContext.getBaseTimestamp();
+        final long timestampDelta = decodingContext.getTimestampDelta();
+        final long initialTimestamp = baseTimestamp + timestampDelta;
+
+        int numValues = valueBuffer.readVInt();
+        List<Long> timestampList = this.codec.decodeTimestamps(initialTimestamp, valueBuffer, numValues);
+
+        final byte[] header = valueBuffer.readPrefixedBytes();
+        AgentStatHeaderDecoder headerDecoder = new BitCountingHeaderDecoder(header);
+        EncodingStrategy<Long> avgEncodingStrategy = UnsignedLongEncodingStrategy.getFromCode(headerDecoder.getCode());
+        EncodingStrategy<Long> minAvgEncodingStrategy = UnsignedLongEncodingStrategy.getFromCode(headerDecoder.getCode());
+        EncodingStrategy<String> minAvgAgentIdEncodingStrategy = StringEncodingStrategy.getFromCode(headerDecoder.getCode());
+        EncodingStrategy<Long> maxAvgEncodingStrategy = UnsignedLongEncodingStrategy.getFromCode(headerDecoder.getCode());
+        EncodingStrategy<String> maxAvgAgentIdEncodingStrategy = StringEncodingStrategy.getFromCode(headerDecoder.getCode());
+
+        List<Long> avgList = this.codec.decodeValues(valueBuffer, avgEncodingStrategy, numValues);
+        List<Long> minAvgList = this.codec.decodeValues(valueBuffer, minAvgEncodingStrategy, numValues);
+        List<String> minAvgAgentIdList = this.codec.decodeValues(valueBuffer, minAvgAgentIdEncodingStrategy, numValues);
+        List<Long> maxAvgList = this.codec.decodeValues(valueBuffer, maxAvgEncodingStrategy, numValues);
+        List<String> maxAvgAgentIdList = this.codec.decodeValues(valueBuffer, maxAvgAgentIdEncodingStrategy, numValues);
+
+        List<JoinStatBo> joinResponseTimeBoList = new ArrayList<JoinStatBo>();
+        for (int i = 0 ; i < numValues ; ++i) {
+            JoinResponseTimeBo joinResponseTimeBo = new JoinResponseTimeBo();
+            joinResponseTimeBo.setId(id);
+            joinResponseTimeBo.setTimestamp(timestampList.get(i));
+            joinResponseTimeBo.setAvg(avgList.get(i));
+            joinResponseTimeBo.setMinAvg(minAvgList.get(i));
+            joinResponseTimeBo.setMinAvgAgentId(minAvgAgentIdList.get(i));
+            joinResponseTimeBo.setMaxAvg(maxAvgList.get(i));
+            joinResponseTimeBo.setMaxAvgAgentId(maxAvgAgentIdList.get(i));
+            joinResponseTimeBoList.add(joinResponseTimeBo);
+        }
+
+        return joinResponseTimeBoList;
+    }
+}