(iotdb) branch master updated: Pipe: accurate recording of heartbeat event duration to improve observability (#11820)

[rong]

This is an automated email from the ASF dual-hosted git repository.

rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 5a19759e11d Pipe: accurate recording of heartbeat event duration to 
improve observability (#11820)
5a19759e11d is described below

commit 5a19759e11d106999d4af6b531a9eefdf79e1c77
Author: V_Galaxy <dyc1904821...@gmail.com>
AuthorDate: Tue Jan 2 15:37:36 2024 +0800

    Pipe: accurate recording of heartbeat event duration to improve 
observability (#11820)
    
    Considering that the various stages of handling the pipe heartbeat event 
are asynchronous, such as the possibility of the assign timestamp being 
recorded while the publish timestamp has not been recorded, calculating 
publishedToAssigned may result in issues. This commit addresses these problems.
---
 .../event/common/heartbeat/PipeHeartbeatEvent.java | 40 ++++++++++++++++------
 1 file changed, 29 insertions(+), 11 deletions(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
index 3e9f3b29d4b..1deedd1120f 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
@@ -146,24 +146,30 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
   public void onAssigned() {
     if (shouldPrintMessage) {
       timeAssigned = System.currentTimeMillis();
-      PipeHeartbeatEventMetrics.getInstance()
-          .recordPublishedToAssignedTime(timeAssigned - timePublished);
+      if (timePublished != 0) {
+        PipeHeartbeatEventMetrics.getInstance()
+            .recordPublishedToAssignedTime(timeAssigned - timePublished);
+      }
     }
   }
 
   public void onProcessed() {
     if (shouldPrintMessage) {
       timeProcessed = System.currentTimeMillis();
-      PipeHeartbeatEventMetrics.getInstance()
-          .recordAssignedToProcessedTime(timeProcessed - timeAssigned);
+      if (timeAssigned != 0) {
+        PipeHeartbeatEventMetrics.getInstance()
+            .recordAssignedToProcessedTime(timeProcessed - timeAssigned);
+      }
     }
   }
 
   public void onTransferred() {
     if (shouldPrintMessage) {
       timeTransferred = System.currentTimeMillis();
-      PipeHeartbeatEventMetrics.getInstance()
-          .recordProcessedToTransferredTime(timeTransferred - timeProcessed);
+      if (timeProcessed != 0) {
+        PipeHeartbeatEventMetrics.getInstance()
+            .recordProcessedToTransferredTime(timeTransferred - timeProcessed);
+      }
     }
   }
 
@@ -221,14 +227,26 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
   public String toString() {
     final String unknownMessage = "Unknown";
 
+    final String startTimeMessage =
+        (timePublished != 0)
+            ? DateTimeUtils.convertLongToDate(timePublished, "ms")
+            : unknownMessage;
     final String publishedToAssignedMessage =
-        timeAssigned != 0 ? (timeAssigned - timePublished) + "ms" : 
unknownMessage;
+        (timeAssigned != 0 && timePublished != 0)
+            ? (timeAssigned - timePublished) + "ms"
+            : unknownMessage;
     final String assignedToProcessedMessage =
-        timeProcessed != 0 ? (timeProcessed - timeAssigned) + "ms" : 
unknownMessage;
+        (timeProcessed != 0 && timeAssigned != 0)
+            ? (timeProcessed - timeAssigned) + "ms"
+            : unknownMessage;
     final String processedToTransferredMessage =
-        timeTransferred != 0 ? (timeTransferred - timeProcessed) + "ms" : 
unknownMessage;
+        (timeTransferred != 0 && timeProcessed != 0)
+            ? (timeTransferred - timeProcessed) + "ms"
+            : unknownMessage;
     final String totalTimeMessage =
-        timeTransferred != 0 ? (timeTransferred - timePublished) + "ms" : 
unknownMessage;
+        (timeTransferred != 0 && timePublished != 0)
+            ? (timeTransferred - timePublished) + "ms"
+            : unknownMessage;
 
     final String disruptorSizeMessage = Integer.toString(disruptorSize);
 
@@ -259,7 +277,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
         + "', dataRegionId="
         + dataRegionId
         + ", startTime="
-        + DateTimeUtils.convertLongToDate(timePublished, "ms")
+        + startTimeMessage
         + ", publishedToAssigned="
         + publishedToAssignedMessage
         + ", assignedToProcessed="