Pipe: Modify Sink batch event length related metrics (#16066)

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java

@@ -364,6 +364,12 @@ public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeInterva
     }
   }
 
+  public void setEventSizeHistogram(Histogram eventSizeHistogram) {
+    if (outputPipeConnector instanceof IoTDBSink) {
+      ((IoTDBSink) outputPipeConnector).setBatchEventSizeHistogram(eventSizeHistogram);
+    }
+  }
+
   //////////////////////////// Error report ////////////////////////////
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java

@@ -135,17 +135,6 @@ private void createAutoGauge(final String taskID) {
         Tag.CREATION_TIME.toString(),
         String.valueOf(connector.getCreationTime()));
     // Metrics related to IoTDB connector
-    metricService.createAutoGauge(
-        Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
-        MetricLevel.IMPORTANT,
-        connector,
-        PipeSinkSubtask::getBatchSize,
-        Tag.NAME.toString(),
-        connector.getAttributeSortedString(),
-        Tag.INDEX.toString(),
-        String.valueOf(connector.getConnectorIndex()),
-        Tag.CREATION_TIME.toString(),
-        String.valueOf(connector.getCreationTime()));
     metricService.createAutoGauge(
         Metric.PIPE_TOTAL_UNCOMPRESSED_SIZE.toString(),
         MetricLevel.IMPORTANT,
@@ -263,6 +252,14 @@ private void createHistogram(final String taskID) {
             Tag.CREATION_TIME.toString(),
             String.valueOf(connector.getCreationTime()));
     connector.setTsFileBatchTimeIntervalHistogram(tsFileBatchTimeIntervalHistogram);
+
+    Histogram eventSizeHistogram =
+        metricService.getOrCreateHistogram(
+            Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
+            MetricLevel.IMPORTANT,
+            Tag.NAME.toString(),
+            connector.getAttributeSortedString());
+    connector.setEventSizeHistogram(eventSizeHistogram);
   }
 
   @Override
@@ -334,15 +331,6 @@ private void removeAutoGauge(final String taskID) {
         Tag.CREATION_TIME.toString(),
         String.valueOf(connector.getCreationTime()));
     // Metrics related to IoTDB connector
-    metricService.remove(
-        MetricType.AUTO_GAUGE,
-        Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
-        Tag.NAME.toString(),
-        connector.getAttributeSortedString(),
-        Tag.INDEX.toString(),
-        String.valueOf(connector.getConnectorIndex()),
-        Tag.CREATION_TIME.toString(),
-        String.valueOf(connector.getCreationTime()));
     metricService.remove(
         MetricType.AUTO_GAUGE,
         Metric.PIPE_TOTAL_UNCOMPRESSED_SIZE.toString(),
@@ -440,6 +428,12 @@ private void removeHistogram(final String taskID) {
         connector.getAttributeSortedString(),
         Tag.CREATION_TIME.toString(),
         String.valueOf(connector.getCreationTime()));
+
+    metricService.remove(
+        MetricType.HISTOGRAM,
+        Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
+        Tag.NAME.toString(),
+        connector.getAttributeSortedString());
   }
 
   //////////////////////////// register & deregister (pipe integration) ////////////////////////////

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java

@@ -36,15 +36,14 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
-import java.util.function.BiConsumer;
 
 public abstract class PipeTabletEventBatch implements AutoCloseable {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PipeTabletEventBatch.class);
   private static PipeModelFixedMemoryBlock pipeModelFixedMemoryBlock = null;
 
   protected final List<EnrichedEvent> events = new ArrayList<>();
-  protected final BiConsumer<Long, Long> recordMetric;
+  protected final TriLongConsumer recordMetric;
 
   private final int maxDelayInMs;
   private long firstEventProcessingTime = Long.MIN_VALUE;
@@ -57,7 +56,7 @@ public abstract class PipeTabletEventBatch implements AutoCloseable {
   protected PipeTabletEventBatch(
       final int maxDelayInMs,
       final long requestMaxBatchSizeInBytes,
-      final BiConsumer<Long, Long> recordMetric) {
+      final TriLongConsumer recordMetric) {
     if (pipeModelFixedMemoryBlock == null) {
       init();
     }
@@ -67,7 +66,7 @@ protected PipeTabletEventBatch(
       this.recordMetric = recordMetric;
     } else {
       this.recordMetric =
-          (timeInterval, bufferSize) -> {
+          (timeInterval, bufferSize, events) -> {
             // do nothing
           };
     }
@@ -142,7 +141,7 @@ public boolean shouldEmit() {
     final long diff = System.currentTimeMillis() - firstEventProcessingTime;
     if (totalBufferSize >= getMaxBatchSizeInBytes() || diff >= maxDelayInMs) {
       allocatedMemoryBlock.updateCurrentMemoryEfficiencyAdjustMem((double) diff / maxDelayInMs);
-      recordMetric.accept(diff, totalBufferSize);
+      recordMetric.accept(diff, totalBufferSize, events.size());
       return true;
     }
     return false;
@@ -235,4 +234,9 @@ public static void init() {
               .forceAllocateForModelFixedMemoryBlock(0, PipeMemoryBlockType.BATCH);
     }
   }
+
+  @FunctionalInterface
+  public interface TriLongConsumer {
+    void accept(long l1, long l2, long l3);
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java

@@ -39,7 +39,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.function.BiConsumer;
 
 public class PipeTabletEventPlainBatch extends PipeTabletEventBatch {
 
@@ -62,7 +61,7 @@ public class PipeTabletEventPlainBatch extends PipeTabletEventBatch {
   PipeTabletEventPlainBatch(
       final int maxDelayInMs,
       final long requestMaxBatchSizeInBytes,
-      final BiConsumer<Long, Long> recordMetric) {
+      final TriLongConsumer recordMetric) {
     super(maxDelayInMs, requestMaxBatchSizeInBytes, recordMetric);
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java

@@ -44,7 +44,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.BiConsumer;
 
 public class PipeTabletEventTsFileBatch extends PipeTabletEventBatch {
 
@@ -69,7 +68,7 @@ public PipeTabletEventTsFileBatch(final int maxDelayInMs, final long requestMaxB
   public PipeTabletEventTsFileBatch(
       final int maxDelayInMs,
       final long requestMaxBatchSizeInBytes,
-      final BiConsumer<Long, Long> recordMetric) {
+      final TriLongConsumer recordMetric) {
     super(maxDelayInMs, requestMaxBatchSizeInBytes, recordMetric);
 
     final AtomicLong tsFileIdGenerator = new AtomicLong(0);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTransferBatchReqBuilder.java

@@ -75,6 +75,8 @@ public class PipeTransferBatchReqBuilder implements AutoCloseable {
   private Histogram tabletBatchTimeIntervalHistogram = new DoNothingHistogram();
   private Histogram tsFileBatchTimeIntervalHistogram = new DoNothingHistogram();
 
+  private Histogram eventSizeHistogram = new DoNothingHistogram();
+
   // If the leader cache is disabled (or unable to find the endpoint of event in the leader cache),
   // the event will be stored in the default batch.
   private final PipeTabletEventBatch defaultBatch;
@@ -220,14 +222,16 @@ public synchronized void close() {
     endPointToBatch.values().forEach(PipeTabletEventPlainBatch::close);
   }
 
-  public void recordTabletMetric(long timeInterval, long bufferSize) {
+  public void recordTabletMetric(long timeInterval, long bufferSize, long eventSize) {
     this.tabletBatchTimeIntervalHistogram.update(timeInterval);
     this.tabletBatchSizeHistogram.update(bufferSize);
+    this.eventSizeHistogram.update(eventSize);
   }
 
-  public void recordTsFileMetric(long timeInterval, long bufferSize) {
+  public void recordTsFileMetric(long timeInterval, long bufferSize, long eventSize) {
     this.tsFileBatchTimeIntervalHistogram.update(timeInterval);
     this.tsFileBatchSizeHistogram.update(bufferSize);
+    this.eventSizeHistogram.update(eventSize);
   }
 
   public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) {
@@ -253,4 +257,10 @@ public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeInterva
       this.tsFileBatchTimeIntervalHistogram = tsFileBatchTimeIntervalHistogram;
     }
   }
+
+  public void setEventSizeHistogram(Histogram eventSizeHistogram) {
+    if (eventSizeHistogram != null) {
+      this.eventSizeHistogram = eventSizeHistogram;
+    }
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java

@@ -861,4 +861,11 @@ public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeInterva
       tabletBatchBuilder.setTsFileBatchTimeIntervalHistogram(tsFileBatchTimeIntervalHistogram);
     }
   }
+
+  @Override
+  public void setBatchEventSizeHistogram(Histogram eventSizeHistogram) {
+    if (tabletBatchBuilder != null) {
+      tabletBatchBuilder.setEventSizeHistogram(eventSizeHistogram);
+    }
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java

@@ -655,4 +655,11 @@ public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeInterva
       tabletBatchBuilder.setTsFileBatchTimeIntervalHistogram(tsFileBatchTimeIntervalHistogram);
     }
   }
+
+  @Override
+  public void setBatchEventSizeHistogram(Histogram eventSizeHistogram) {
+    if (tabletBatchBuilder != null) {
+      tabletBatchBuilder.setEventSizeHistogram(eventSizeHistogram);
+    }
+  }
 }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java

@@ -648,4 +648,8 @@ public void setTabletBatchTimeIntervalHistogram(Histogram tabletBatchTimeInterva
   public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeIntervalHistogram) {
     // do nothing by default
   }
+
+  public void setBatchEventSizeHistogram(Histogram tsFileBatchTimeIntervalHistogram) {
+    // do nothing by default
+  }
 }