[pinpoint-apm#9495] Resolving application tps data consistency issues

ga-ram · Dec 9, 2022 · 2e9f752 · 2e9f752
1 parent 44e6d72
commit 2e9f752
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 7 deletions.
diff --git a/...rc/main/java/com/navercorp/pinpoint/flink/mapper/thrift/stat/JoinTransactionBoMapper.java b/...rc/main/java/com/navercorp/pinpoint/flink/mapper/thrift/stat/JoinTransactionBoMapper.java
@@ -20,6 +20,7 @@
 import com.navercorp.pinpoint.common.server.bo.stat.join.JoinTransactionBo;
 import com.navercorp.pinpoint.thrift.dto.flink.TFAgentStat;
 import com.navercorp.pinpoint.thrift.dto.flink.TFTransaction;
+import org.apache.commons.math3.util.Precision;
 
 /**
  * @author minwoo.jung
@@ -33,7 +34,7 @@ public JoinTransactionBo map(TFAgentStat tFAgentStat) {
         }
 
         TFTransaction tFtransaction = tFAgentStat.getTransaction();
-        final long totalCount = calculateTotalCount(tFtransaction);
+        final long totalCount = calculateTotalCount(tFtransaction, tFAgentStat.getCollectInterval());
         final String agentId = tFAgentStat.getAgentId();
 
         JoinTransactionBo joinTransactionBo = new JoinTransactionBo();
@@ -45,17 +46,27 @@ public JoinTransactionBo map(TFAgentStat tFAgentStat) {
         return joinTransactionBo;
     }
 
-    private long calculateTotalCount(TFTransaction tFtransaction) {
+    private long calculateTotalCount(TFTransaction tFtransaction, long collectInterval) {
         long totalCount = 0;
         totalCount += tFtransaction.getSampledNewCount();
         totalCount += tFtransaction.getSampledContinuationCount();
         totalCount += tFtransaction.getUnsampledNewCount();
         totalCount += tFtransaction.getUnsampledContinuationCount();
         totalCount += tFtransaction.getSkippedNewCount();
         totalCount += tFtransaction.getSkippedContinuationCount();
+
+        totalCount = calculateAvgTotalCount(totalCount, collectInterval);
         return totalCount;
     }
 
+    private static long calculateAvgTotalCount(long totalCount, long timeMs) {
+        if (totalCount <= 0) {
+            return totalCount;
+        }
+
+        return (long) Precision.round(totalCount / (timeMs / 1000D), 1);
+    }
+
     @Override
     public void build(TFAgentStat tFAgentStat, JoinAgentStatBo.Builder builder) {
         JoinTransactionBo joinTransactionBo = this.map(tFAgentStat);

diff --git a/.../test/java/com/navercorp/pinpoint/flink/mapper/thrift/stat/JoinAgentStatBoMapperTest.java b/.../test/java/com/navercorp/pinpoint/flink/mapper/thrift/stat/JoinAgentStatBoMapperTest.java
@@ -220,13 +220,13 @@ public void map3Test() {
         assertEquals(joinTransactionBo.getId(), agentId);
         assertEquals(joinTransactionBo.getTimestamp(), 1491274148454L);
         assertEquals(joinTransactionBo.getCollectInterval(), 5000);
-        assertEquals(joinTransactionBo.getTotalCountJoinValue(), new JoinLongFieldBo(120L, 120L, agentId, 120L, agentId));
+        assertEquals(joinTransactionBo.getTotalCountJoinValue(), new JoinLongFieldBo(24L, 24L, agentId, 24L, agentId));
 
         JoinTransactionBo joinTransactionBo2 = joinTransactionBoList.get(1);
         assertEquals(joinTransactionBo2.getId(), agentId);
         assertEquals(joinTransactionBo2.getTimestamp(), 1491275148454L);
         assertEquals(joinTransactionBo2.getCollectInterval(), 5000);
-        assertEquals(joinTransactionBo2.getTotalCountJoinValue(), new JoinLongFieldBo(124L, 124L, agentId, 124L, agentId));
+        assertEquals(joinTransactionBo2.getTotalCountJoinValue(), new JoinLongFieldBo(24L, 24L, agentId, 24L, agentId));
     }
 
     @Test

diff --git a/...est/java/com/navercorp/pinpoint/flink/mapper/thrift/stat/JoinTransactionBoMapperTest.java b/...est/java/com/navercorp/pinpoint/flink/mapper/thrift/stat/JoinTransactionBoMapperTest.java
@@ -54,7 +54,7 @@ public void mapTest() throws Exception {
         assertEquals(joinTransactionBo.getId(), id);
         assertEquals(joinTransactionBo.getTimestamp(), 1491274138454L);
         assertEquals(joinTransactionBo.getCollectInterval(), 5000);
-        assertEquals(joinTransactionBo.getTotalCountJoinValue(), new JoinLongFieldBo(250L, 250L, id, 250L, id));
+        assertEquals(joinTransactionBo.getTotalCountJoinValue(), new JoinLongFieldBo(50L, 50L, id, 50L, id));
     }
 
     @Test

diff --git a/flink/src/test/java/com/navercorp/pinpoint/flink/process/TBaseFlatMapperTest.java b/flink/src/test/java/com/navercorp/pinpoint/flink/process/TBaseFlatMapperTest.java
@@ -269,13 +269,13 @@ private void assertJoinTransactionBo(List<JoinTransactionBo> joinTransactionBoLi
         assertEquals(joinTransactionBo.getId(), AGENT_ID);
         assertEquals(joinTransactionBo.getTimestamp(), 1491274143454L);
         assertEquals(joinTransactionBo.getCollectInterval(), 5000);
-        assertEquals(joinTransactionBo.getTotalCountJoinValue(), new JoinLongFieldBo(120L, 120L, AGENT_ID, 120L, AGENT_ID));
+        assertEquals(joinTransactionBo.getTotalCountJoinValue(), new JoinLongFieldBo(24L, 24L, AGENT_ID, 24L, AGENT_ID));
 
         JoinTransactionBo joinTransactionBo2 = joinTransactionBoList.get(1);
         assertEquals(joinTransactionBo2.getId(), AGENT_ID);
         assertEquals(joinTransactionBo2.getTimestamp(), 1491274148454L);
         assertEquals(joinTransactionBo2.getCollectInterval(), 5000);
-        assertEquals(joinTransactionBo2.getTotalCountJoinValue(), new JoinLongFieldBo(124L, 124L, AGENT_ID, 124L, AGENT_ID));
+        assertEquals(joinTransactionBo2.getTotalCountJoinValue(), new JoinLongFieldBo(24L, 24L, AGENT_ID, 24L, AGENT_ID));
     }
 
     private TFAgentStatBatch createTFAgentStatBatch3() {