Pipe: Unbinded the default meter for remaining insertion event count for hybrid degrading & Changed some default values of related properties (apache#15615)

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeOperator.java

@@ -19,7 +19,7 @@
 
 package org.apache.iotdb.confignode.manager.pipe.metric.overview;
 
-import org.apache.iotdb.commons.enums.PipeRemainingTimeRateAverageTime;
+import org.apache.iotdb.commons.enums.PipeRateAverage;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.metric.PipeRemainingOperator;
 import org.apache.iotdb.confignode.manager.pipe.agent.task.PipeConfigNodeSubtask;
@@ -56,7 +56,7 @@ class PipeConfigNodeRemainingTimeOperator extends PipeRemainingOperator {
    * @return The estimated remaining time
    */
   double getRemainingTime() {
-    final PipeRemainingTimeRateAverageTime pipeRemainingTimeCommitRateAverageTime =
+    final PipeRateAverage pipeRemainingTimeCommitRateAverageTime =
         PipeConfig.getInstance().getPipeRemainingTimeCommitRateAverageTime();
 
     // Do not calculate heartbeat event

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

@@ -19,7 +19,7 @@
 
 package org.apache.iotdb.db.pipe.metric.overview;
 
-import org.apache.iotdb.commons.enums.PipeRemainingTimeRateAverageTime;
+import org.apache.iotdb.commons.enums.PipeRateAverage;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.metric.PipeRemainingOperator;
 import org.apache.iotdb.db.pipe.extractor.schemaregion.IoTDBSchemaRegionExtractor;
@@ -106,7 +106,7 @@ void decreaseHeartbeatEventCount() {
       lastInsertNodeEventCountSmoothingTime = System.currentTimeMillis();
     }
     return PipeConfig.getInstance()
-        .getPipeRemainingTimeCommitRateAverageTime()
+        .getPipeRemainingInsertNodeCountAverage()
         .getMeterRate(insertNodeEventCountMeter);
   }
 
@@ -135,7 +135,7 @@ long getRemainingEvents() {
    * @return The estimated remaining time
    */
   double getRemainingTime() {
-    final PipeRemainingTimeRateAverageTime pipeRemainingTimeCommitRateAverageTime =
+    final PipeRateAverage pipeRemainingTimeCommitRateAverageTime =
         PipeConfig.getInstance().getPipeRemainingTimeCommitRateAverageTime();
 
     final double invocationValue = collectInvocationHistogram.getMean();

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java

@@ -23,7 +23,7 @@
 import org.apache.iotdb.commons.client.property.ClientPoolProperty.DefaultProperty;
 import org.apache.iotdb.commons.cluster.NodeStatus;
 import org.apache.iotdb.commons.enums.HandleSystemErrorStrategy;
-import org.apache.iotdb.commons.enums.PipeRemainingTimeRateAverageTime;
+import org.apache.iotdb.commons.enums.PipeRateAverage;
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
 import org.apache.iotdb.rpc.RpcUtils;
@@ -280,7 +280,7 @@ public class CommonConfig {
   private long pipeStorageEngineFlushTimeIntervalMs = Long.MAX_VALUE;
   private int pipeMaxAllowedRemainingInsertEventCountPerPipe = 10000;
   private int pipeMaxAllowedTotalRemainingInsertEventCount = 50000;
-  private int pipeRemainingEventCountSmoothingIntervalSeconds = 15;
+  private int pipeRemainingEventCountSmoothingIntervalSeconds = 10;
 
   private int pipeMetaReportMaxLogNumPerRound = 10;
   private int pipeMetaReportMaxLogIntervalRounds = 36;
@@ -301,8 +301,8 @@ public class CommonConfig {
   private long pipeListeningQueueTransferSnapshotThreshold = 1000;
   private int pipeSnapshotExecutionMaxBatchSize = 1000;
   private long pipeRemainingTimeCommitRateAutoSwitchSeconds = 30;
-  private PipeRemainingTimeRateAverageTime pipeRemainingTimeCommitRateAverageTime =
-      PipeRemainingTimeRateAverageTime.MEAN;
+  private PipeRateAverage pipeRemainingTimeCommitRateAverageTime = PipeRateAverage.FIVE_MINUTES;
+  private PipeRateAverage pipeRemainingInsertNodeCountAverage = PipeRateAverage.ONE_MINUTE;
   private double pipeTsFileScanParsingThreshold = 0.05;
   private double pipeDynamicMemoryHistoryWeight = 0.5;
   private double pipeDynamicMemoryAdjustmentThreshold = 0.05;
@@ -1831,12 +1831,12 @@ public void setPipeRemainingTimeCommitRateAutoSwitchSeconds(
         pipeRemainingTimeCommitRateAutoSwitchSeconds);
   }
 
-  public PipeRemainingTimeRateAverageTime getPipeRemainingTimeCommitRateAverageTime() {
+  public PipeRateAverage getPipeRemainingTimeCommitRateAverageTime() {
     return pipeRemainingTimeCommitRateAverageTime;
   }
 
   public void setPipeRemainingTimeCommitRateAverageTime(
-      PipeRemainingTimeRateAverageTime pipeRemainingTimeCommitRateAverageTime) {
+      PipeRateAverage pipeRemainingTimeCommitRateAverageTime) {
     if (Objects.equals(
         this.pipeRemainingTimeCommitRateAverageTime, pipeRemainingTimeCommitRateAverageTime)) {
       return;
@@ -1847,6 +1847,21 @@ public void setPipeRemainingTimeCommitRateAverageTime(
         pipeRemainingTimeCommitRateAverageTime);
   }
 
+  public PipeRateAverage getPipeRemainingInsertNodeCountAverage() {
+    return pipeRemainingInsertNodeCountAverage;
+  }
+
+  public void setPipeRemainingInsertNodeCountAverage(
+      PipeRateAverage pipeRemainingInsertNodeCountAverage) {
+    if (Objects.equals(
+        this.pipeRemainingInsertNodeCountAverage, pipeRemainingInsertNodeCountAverage)) {
+      return;
+    }
+    this.pipeRemainingInsertNodeCountAverage = pipeRemainingInsertNodeCountAverage;
+    logger.info(
+        "pipeRemainingInsertEventCountAverage is set to {}", pipeRemainingInsertNodeCountAverage);
+  }
+
   public double getPipeTsFileScanParsingThreshold() {
     return pipeTsFileScanParsingThreshold;
   }

…ms/PipeRemainingTimeRateAverageTime.java …iotdb/commons/enums/PipeRateAverage.javaiotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/enums/PipeRemainingTimeRateAverageTime.java renamed to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/enums/PipeRateAverage.java iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/enums/PipeRemainingTimeRateAverageTime.java renamed to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/enums/PipeRateAverage.java

@@ -19,11 +19,9 @@
 
 package org.apache.iotdb.commons.enums;
 
-import org.apache.iotdb.commons.pipe.config.PipeConfig;
-
 import com.codahale.metrics.Meter;
 
-public enum PipeRemainingTimeRateAverageTime {
+public enum PipeRateAverage {
   ONE_MINUTE,
   FIVE_MINUTES,
   FIFTEEN_MINUTES,
@@ -41,9 +39,7 @@ public double getMeterRate(final Meter meter) {
         return meter.getMeanRate();
       default:
         throw new UnsupportedOperationException(
-            String.format(
-                "The type %s is not supported in average time of pipe remaining time.",
-                PipeConfig.getInstance().getPipeRemainingTimeCommitRateAverageTime()));
+            String.format("The type %s is not supported in pipe rate average.", this));
     }
   }
 }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java

@@ -21,7 +21,7 @@
 
 import org.apache.iotdb.commons.conf.CommonConfig;
 import org.apache.iotdb.commons.conf.CommonDescriptor;
-import org.apache.iotdb.commons.enums.PipeRemainingTimeRateAverageTime;
+import org.apache.iotdb.commons.enums.PipeRateAverage;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -223,10 +223,14 @@ public long getPipeRemainingTimeCommitAutoSwitchSeconds() {
     return COMMON_CONFIG.getPipeRemainingTimeCommitRateAutoSwitchSeconds();
   }
 
-  public PipeRemainingTimeRateAverageTime getPipeRemainingTimeCommitRateAverageTime() {
+  public PipeRateAverage getPipeRemainingTimeCommitRateAverageTime() {
     return COMMON_CONFIG.getPipeRemainingTimeCommitRateAverageTime();
   }
 
+  public PipeRateAverage getPipeRemainingInsertNodeCountAverage() {
+    return COMMON_CONFIG.getPipeRemainingInsertNodeCountAverage();
+  }
+
   public double getPipeTsFileScanParsingThreshold() {
     return COMMON_CONFIG.getPipeTsFileScanParsingThreshold();
   }
@@ -513,6 +517,8 @@ public void printAllConfigs() {
         getPipeRemainingTimeCommitAutoSwitchSeconds());
     LOGGER.info(
         "PipeRemainingTimeCommitRateAverageTime: {}", getPipeRemainingTimeCommitRateAverageTime());
+    LOGGER.info(
+        "PipePipeRemainingInsertEventCountAverage: {}", getPipeRemainingInsertNodeCountAverage());
     LOGGER.info("PipeTsFileScanParsingThreshold(): {}", getPipeTsFileScanParsingThreshold());
 
     LOGGER.info("PipeDynamicMemoryHistoryWeight: {}", getPipeDynamicMemoryHistoryWeight());

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java

@@ -22,7 +22,7 @@
 import org.apache.iotdb.commons.conf.CommonConfig;
 import org.apache.iotdb.commons.conf.ConfigurationFileUtils;
 import org.apache.iotdb.commons.conf.TrimProperties;
-import org.apache.iotdb.commons.enums.PipeRemainingTimeRateAverageTime;
+import org.apache.iotdb.commons.enums.PipeRateAverage;
 
 import java.io.IOException;
 import java.util.Optional;
@@ -184,12 +184,19 @@ public static void loadPipeStaticConfig(CommonConfig config, TrimProperties prop
                 "pipe_snapshot_execution_max_batch_size",
                 String.valueOf(config.getPipeSnapshotExecutionMaxBatchSize()))));
     config.setPipeRemainingTimeCommitRateAverageTime(
-        PipeRemainingTimeRateAverageTime.valueOf(
+        PipeRateAverage.valueOf(
             properties
                 .getProperty(
                     "pipe_remaining_time_commit_rate_average_time",
                     String.valueOf(config.getPipeRemainingTimeCommitRateAverageTime()))
                 .trim()));
+    config.setPipeRemainingInsertNodeCountAverage(
+        PipeRateAverage.valueOf(
+            properties
+                .getProperty(
+                    "pipe_remaining_insert_node_count_average",
+                    String.valueOf(config.getPipeRemainingInsertNodeCountAverage()))
+                .trim()));
   }
 
   public static void loadPipeInternalConfig(CommonConfig config, TrimProperties properties)