[to dev/1.3] Pipe: add ConfigNode memory control for log reducer (#18069)

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java

@@ -56,6 +56,8 @@ public class ConfigNodeDescriptor {
 
   private final ConfigNodeConfig conf = new ConfigNodeConfig();
 
+  private final ConfigNodeMemoryConfig memoryConfig = new ConfigNodeMemoryConfig();
+
   static {
     URL systemConfigUrl = getPropsUrl(CommonConfig.SYSTEM_CONFIG_NAME);
     URL configNodeUrl = getPropsUrl(CommonConfig.OLD_CONFIG_NODE_CONFIG_NAME);
@@ -79,6 +81,10 @@ public ConfigNodeConfig getConf() {
     return conf;
   }
 
+  public ConfigNodeMemoryConfig getMemoryConfig() {
+    return memoryConfig;
+  }
+
   /**
    * Get props url location.
    *
@@ -143,10 +149,13 @@ private void loadProps() {
       LOGGER.warn(
           "Couldn't load the configuration {} from any of the known sources.",
           CommonConfig.SYSTEM_CONFIG_NAME);
+      memoryConfig.init(trimProperties);
     }
   }
 
   private void loadProperties(TrimProperties properties) throws BadNodeUrlException, IOException {
+    memoryConfig.init(properties);
+
     conf.setClusterName(
         properties.getProperty(IoTDBConstant.CLUSTER_NAME, conf.getClusterName()).trim());
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeMemoryConfig.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.conf;
+
+import org.apache.iotdb.commons.conf.TrimProperties;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ConfigNodeMemoryConfig {
+  public static final String PIPE_MEMORY_MANAGER_NAME = "Pipe";
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ConfigNodeMemoryConfig.class);
+
+  private long maxMemorySizeInBytes;
+
+  private long pipeMemorySizeInBytes;
+
+  private long freeMemorySizeInBytes;
+
+  public void init(final TrimProperties properties) {
+    final String memoryAllocateProportion =
+        properties.getProperty("confignode_memory_proportion", null);
+
+    maxMemorySizeInBytes = Runtime.getRuntime().maxMemory();
+    pipeMemorySizeInBytes = maxMemorySizeInBytes / 10;
+    freeMemorySizeInBytes = maxMemorySizeInBytes - pipeMemorySizeInBytes;
+
+    if (memoryAllocateProportion != null) {
+      final String[] proportions = memoryAllocateProportion.split(":");
+      if (proportions.length >= 2) {
+        int proportionSum = 0;
+        for (final String proportion : proportions) {
+          proportionSum += Integer.parseInt(proportion.trim());
+        }
+
+        if (proportionSum != 0) {
+          pipeMemorySizeInBytes =
+              maxMemorySizeInBytes * Integer.parseInt(proportions[0].trim()) / proportionSum;
+          freeMemorySizeInBytes = maxMemorySizeInBytes - pipeMemorySizeInBytes;
+        }
+      } else {
+        LOGGER.warn(
+            "The parameter confignode_memory_proportion should be in the form of Pipe:Free, "
+                + "but got {}. Use default value 1:9.",
+            memoryAllocateProportion);
+      }
+    }
+
+    LOGGER.info("initial ConfigNode allocateMemoryForPipe = {}", pipeMemorySizeInBytes);
+    LOGGER.info("initial ConfigNode freeMemory = {}", freeMemorySizeInBytes);
+  }
+
+  public long getMaxMemorySizeInBytes() {
+    return maxMemorySizeInBytes;
+  }
+
+  public long getPipeMemorySizeInBytes() {
+    return pipeMemorySizeInBytes;
+  }
+
+  public long getFreeMemorySizeInBytes() {
+    return freeMemorySizeInBytes;
+  }
+}

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java

@@ -26,10 +26,13 @@
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
+import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
 import org.apache.iotdb.commons.service.IService;
 import org.apache.iotdb.commons.service.ServiceType;
 import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
 import org.apache.iotdb.confignode.manager.pipe.resource.PipeConfigNodeCopiedFileDirStartupCleaner;
+import org.apache.iotdb.confignode.manager.pipe.resource.PipeConfigNodeResourceManager;
 import org.apache.iotdb.confignode.manager.pipe.source.ConfigRegionListeningQueue;
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 
@@ -55,6 +58,7 @@ public class PipeConfigNodeRuntimeAgent implements IService {
   @Override
   public synchronized void start() {
     PipeConfig.getInstance().printAllConfigs();
+    initPipePeriodicalLogReducer();
 
     // PipeTasks will not be started here and will be started by "HandleLeaderChange"
     // procedure when the consensus layer notify leader ready
@@ -91,6 +95,12 @@ public synchronized void stop() {
     LOGGER.info("PipeRuntimeConfigNodeAgent stopped");
   }
 
+  private void initPipePeriodicalLogReducer() {
+    PipePeriodicalLogReducer.setMemoryResizeFunction(
+        PipeConfigNodeResourceManager::resizeLogReducerMemory);
+    PipeLogger.setLogger(PipePeriodicalLogReducer::log);
+  }
+
   public boolean isShutdown() {
     return isShutdown.get();
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/resource/PipeConfigNodeResourceManager.java

@@ -22,12 +22,16 @@
 import org.apache.iotdb.commons.pipe.resource.log.PipeLogManager;
 import org.apache.iotdb.commons.pipe.resource.ref.PipePhantomReferenceManager;
 import org.apache.iotdb.commons.pipe.resource.snapshot.PipeSnapshotResourceManager;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.manager.pipe.resource.ref.PipeConfigNodePhantomReferenceManager;
 import org.apache.iotdb.confignode.manager.pipe.resource.snapshot.PipeConfigNodeSnapshotResourceManager;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 public class PipeConfigNodeResourceManager {
 
   private final PipeSnapshotResourceManager pipeSnapshotResourceManager;
+  private final AtomicLong pipeLogReducerMemoryUsageInBytes = new AtomicLong(0);
   private final PipeLogManager pipeLogManager;
   private final PipePhantomReferenceManager pipePhantomReferenceManager;
 
@@ -36,6 +40,10 @@ public static PipeSnapshotResourceManager snapshot() {
         .pipeSnapshotResourceManager;
   }
 
+  public static long resizeLogReducerMemory(final long targetSizeInBytes) {
+    return PipeResourceManagerHolder.INSTANCE.resizePipeLogReducerMemory(targetSizeInBytes);
+  }
+
   public static PipeLogManager log() {
     return PipeConfigNodeResourceManager.PipeResourceManagerHolder.INSTANCE.pipeLogManager;
   }
@@ -46,6 +54,14 @@ public static PipePhantomReferenceManager ref() {
 
   ///////////////////////////// SINGLETON /////////////////////////////
 
+  private long resizePipeLogReducerMemory(final long targetSizeInBytes) {
+    final long pipeMemorySizeInBytes =
+        ConfigNodeDescriptor.getInstance().getMemoryConfig().getPipeMemorySizeInBytes();
+    final long resizedSizeInBytes = Math.min(Math.max(0, targetSizeInBytes), pipeMemorySizeInBytes);
+    pipeLogReducerMemoryUsageInBytes.set(resizedSizeInBytes);
+    return pipeLogReducerMemoryUsageInBytes.get();
+  }
+
   private PipeConfigNodeResourceManager() {
     pipeSnapshotResourceManager = new PipeConfigNodeSnapshotResourceManager();
     pipeLogManager = new PipeLogManager();

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeMemoryConfigTest.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.conf;
+
+import org.apache.iotdb.commons.conf.TrimProperties;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ConfigNodeMemoryConfigTest {
+
+  @Test
+  public void testConfigNodeMemoryProportion() {
+    final TrimProperties properties = new TrimProperties();
+    properties.setProperty("confignode_memory_proportion", "1:3");
+
+    final ConfigNodeMemoryConfig memoryConfig = new ConfigNodeMemoryConfig();
+    memoryConfig.init(properties);
+
+    Assert.assertEquals(
+        Runtime.getRuntime().maxMemory() / 4, memoryConfig.getPipeMemorySizeInBytes());
+    Assert.assertEquals(
+        Runtime.getRuntime().maxMemory() * 3 / 4, memoryConfig.getFreeMemorySizeInBytes());
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -26,6 +26,7 @@
 import org.apache.iotdb.commons.conf.TrimProperties;
 import org.apache.iotdb.commons.exception.BadNodeUrlException;
 import org.apache.iotdb.commons.pipe.config.PipeDescriptor;
+import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
 import org.apache.iotdb.commons.schema.SchemaConstant;
 import org.apache.iotdb.commons.service.metric.MetricService;
 import org.apache.iotdb.commons.utils.NodeUrlUtils;
@@ -34,7 +35,6 @@
 import org.apache.iotdb.confignode.rpc.thrift.TRatisConfig;
 import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
 import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.pipe.resource.log.PipePeriodicalLogReducer;
 import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.LastCacheLoadStrategy;
 import org.apache.iotdb.db.service.metrics.IoTDBInternalLocalReporter;
 import org.apache.iotdb.db.storageengine.StorageEngine;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java

@@ -32,13 +32,15 @@
 import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
+import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
 import org.apache.iotdb.commons.service.IService;
 import org.apache.iotdb.commons.service.ServiceType;
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner;
-import org.apache.iotdb.db.pipe.resource.log.PipePeriodicalLogReducer;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import org.apache.iotdb.db.pipe.source.schemaregion.SchemaRegionListeningQueue;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
 import org.apache.iotdb.db.service.ResourcesInformationHolder;
@@ -70,6 +72,8 @@ public class PipeDataNodeRuntimeAgent implements IService {
   private final PipePeriodicalPhantomReferenceCleaner pipePeriodicalPhantomReferenceCleaner =
       new PipePeriodicalPhantomReferenceCleaner();
 
+  private PipeMemoryBlock pipeLogReducerMemoryBlock;
+
   //////////////////////////// System Service Interface ////////////////////////////
 
   public synchronized void preparePipeResources(
@@ -85,6 +89,22 @@ public synchronized void preparePipeResources(
 
     IoTDBPipePattern.setDevicePathGetter(CompactionPathUtils::getPath);
     IoTDBPipePattern.setMeasurementPathGetter(CompactionPathUtils::getPath);
+    initPipePeriodicalLogReducer();
+  }
+
+  private void initPipePeriodicalLogReducer() {
+    if (pipeLogReducerMemoryBlock == null) {
+      pipeLogReducerMemoryBlock =
+          PipeDataNodeResourceManager.memory()
+              .tryAllocate(PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes());
+    }
+
+    PipePeriodicalLogReducer.setMemoryResizeFunction(
+        targetSizeInBytes -> {
+          PipeDataNodeResourceManager.memory()
+              .resize(pipeLogReducerMemoryBlock, Math.max(0, targetSizeInBytes), false);
+          return pipeLogReducerMemoryBlock.getMemoryUsageInBytes();
+        });
     PipeLogger.setLogger(PipePeriodicalLogReducer::log);
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -31,6 +31,7 @@
 import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
 import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
 import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
+import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
 import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapPseudoTPipeTransferRequest;
 import org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferSliceReqHandler;
 import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
@@ -52,7 +53,6 @@
 import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementTSStatusVisitor;
 import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementToBatchVisitor;
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
-import org.apache.iotdb.db.pipe.resource.log.PipePeriodicalLogReducer;
 import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
 import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req;
 import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;

iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template

@@ -730,6 +730,11 @@ disk_space_warning_threshold=0.05
 # effectiveMode: restart
 datanode_memory_proportion=3:3:1:1:1:1
 
+# ConfigNode Memory Allocation Ratio: Pipe and Free Memory.
+# The parameter form is a:b, where a and b are integers. Currently, only PipePeriodicalLogReducer is connected to Pipe memory on ConfigNode.
+# effectiveMode: restart
+confignode_memory_proportion=1:9
+
 # Schema Memory Allocation Ratio: SchemaRegion, SchemaCache, and PartitionCache.
 # The parameter form is a:b:c, where a, b and c are integers. for example: 1:1:1 , 6:2:1
 # effectiveMode: restart