Pipe: Fix lost transfer time metric & Support transfer time tracking for RawTablet events converted from InsertNode/TsFile events (apache#16364)

* fix metric

* fix

* fix

Author: XNX02

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java

@@ -162,6 +162,10 @@ public String getDeviceId() {
         : null;
   }
 
+  public long getExtractTime() {
+    return extractTime;
+  }
+
   /////////////////////////// EnrichedEvent ///////////////////////////
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java

@@ -263,6 +263,25 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
     // Actually release the occupied memory.
     tablet = null;
     eventParser = null;
+
+    // Update metrics of the source event
+    if (needToReport && shouldReportOnCommit && Objects.nonNull(pipeName)) {
+      if (sourceEvent instanceof PipeInsertNodeTabletInsertionEvent) {
+        PipeDataNodeSinglePipeMetrics.getInstance()
+            .updateInsertNodeTransferTimer(
+                pipeName,
+                creationTime,
+                System.nanoTime()
+                    - ((PipeInsertNodeTabletInsertionEvent) sourceEvent).getExtractTime());
+      } else if (sourceEvent instanceof PipeTsFileInsertionEvent) {
+        PipeDataNodeSinglePipeMetrics.getInstance()
+            .updateTsFileTransferTimer(
+                pipeName,
+                creationTime,
+                System.nanoTime() - ((PipeTsFileInsertionEvent) sourceEvent).getExtractTime());
+      }
+    }
+
     return true;
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

@@ -298,6 +298,10 @@ public long getTimePartitionId() {
     return resource.getTimePartition();
   }
 
+  public long getExtractTime() {
+    return extractTime;
+  }
+
   /////////////////////////// EnrichedEvent ///////////////////////////
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java

@@ -39,6 +39,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 
 public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
 
@@ -236,7 +237,24 @@ public void decreaseInsertNodeEventCount(
         remainingEventAndTimeOperatorMap.computeIfAbsent(
             pipeName + "_" + creationTime,
             k -> new PipeDataNodeRemainingEventAndTimeOperator(pipeName, creationTime));
+
     operator.decreaseInsertNodeEventCount();
+
+    if (transferTime > 0) {
+      operator.getInsertNodeTransferTimer().update(transferTime, TimeUnit.NANOSECONDS);
+    }
+  }
+
+  public void updateInsertNodeTransferTimer(
+      final String pipeName, final long creationTime, final long transferTime) {
+    if (transferTime > 0) {
+      remainingEventAndTimeOperatorMap
+          .computeIfAbsent(
+              pipeName + "_" + creationTime,
+              k -> new PipeDataNodeRemainingEventAndTimeOperator(pipeName, creationTime))
+          .getInsertNodeTransferTimer()
+          .update(transferTime, TimeUnit.NANOSECONDS);
+    }
   }
 
   public void increaseRawTabletEventCount(final String pipeName, final long creationTime) {
@@ -271,6 +289,22 @@ public void decreaseTsFileEventCount(
             k -> new PipeDataNodeRemainingEventAndTimeOperator(pipeName, creationTime));
 
     operator.decreaseTsFileEventCount();
+
+    if (transferTime > 0) {
+      operator.getTsFileTransferTimer().update(transferTime, TimeUnit.NANOSECONDS);
+    }
+  }
+
+  public void updateTsFileTransferTimer(
+      final String pipeName, final long creationTime, final long transferTime) {
+    if (transferTime > 0) {
+      remainingEventAndTimeOperatorMap
+          .computeIfAbsent(
+              pipeName + "_" + creationTime,
+              k -> new PipeDataNodeRemainingEventAndTimeOperator(pipeName, creationTime))
+          .getTsFileTransferTimer()
+          .update(transferTime, TimeUnit.NANOSECONDS);
+    }
   }
 
   public void increaseHeartbeatEventCount(final String pipeName, final long creationTime) {