HDDS-17724. Limit command-status report size in datanode heartbeats.

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java

@@ -73,6 +73,9 @@ public final class HddsConfigKeys {
       "hdds.heartbeat.report.max.limit";
   public static final int HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT =
       10000;
+  public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE =
+      "hdds.heartbeat.command.status.report.max.size";
+  public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT = "64MB";
   // Configuration to allow volume choosing policy.
   public static final String HDDS_DATANODE_VOLUME_CHOOSING_POLICY =
       "hdds.datanode.volume.choosing.policy";

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java

@@ -112,7 +112,10 @@ public class ScmConfig extends ReconfigurableConfig {
               + "The block num is counted at the replica level."
               + "If SCM has 100000 blocks which need to be deleted and the "
               + "configuration is 5000 then it would only send 5000 blocks "
-              + "for deletion to the datanodes."
+              + "for deletion to the datanodes. Increasing this value can "
+              + "increase delete command acknowledgement size; each "
+              + "individual acknowledgement must fit within the IPC size "
+              + "limit (ipc.maximum.data.length)."
   )
   private int blockDeletionLimit;
 

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -1805,8 +1805,22 @@
     <description>
       Maximum number of report messages (not containers) a datanode includes in
       a single heartbeat to an SCM or Recon endpoint. Reports beyond the limit
-      stay queued and are sent in an immediate follow-up heartbeat. Values below
-      1 are treated as 1.
+      stay queued and are sent in later scheduled heartbeats. Values below 1
+      are treated as 1.
+    </description>
+  </property>
+  <property>
+    <name>hdds.heartbeat.command.status.report.max.size</name>
+    <value>64MB</value>
+    <tag>DATANODE</tag>
+    <description>
+      Maximum serialized size of command status reports included in one
+      heartbeat. The effective limit can be lower when other heartbeat fields
+      consume the safe RPC payload budget. Reports beyond the limit stay queued
+      for a later heartbeat. A single command status that exceeds this limit is
+      sent as the only command status report when it still fits within the
+      remaining safe RPC payload, ensuring that it does not block all later
+      command statuses.
     </description>
   </property>
   <property>

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java

@@ -23,6 +23,7 @@
 import com.google.common.base.CaseFormat;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
 import org.apache.commons.text.WordUtils;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
 import org.apache.hadoop.hdds.scm.net.HostAndPort;
@@ -60,6 +61,14 @@ public final class DatanodeQueueMetrics implements MetricsSource {
       "ContainerActionQueue";
   public static final String PIPELINE_ACTION_QUEUE_PREFIX =
       "PipelineActionQueue";
+  public static final String COMMAND_STATUS_REPORT_QUEUE_PREFIX =
+      "CommandStatusReportQueue";
+  public static final String COMMAND_STATUS_REPORT_SENT_PREFIX =
+      "CommandStatusReportSent";
+  public static final String COMMAND_STATUS_REPORT_REQUEUED_PREFIX =
+      "CommandStatusReportRequeued";
+  public static final String COMMAND_STATUS_REPORT_OVERSIZED_PREFIX =
+      "CommandStatusReportOversized";
 
   private MetricsRegistry registry;
 
@@ -71,6 +80,13 @@ public final class DatanodeQueueMetrics implements MetricsSource {
   private Map<HostAndPort, MetricsInfo> incrementalReportsQueueMap;
   private Map<HostAndPort, MetricsInfo> containerActionQueueMap;
   private Map<HostAndPort, MetricsInfo> pipelineActionQueueMap;
+  private Map<HostAndPort, MetricsInfo> commandStatusReportQueueBytesMap;
+  private Map<HostAndPort, MetricsInfo> commandStatusReportSentBytesMap;
+  private Map<HostAndPort, MetricsInfo> commandStatusReportRequeuedBytesMap;
+  private Map<HostAndPort, MetricsInfo> oversizedCommandStatusReportBytesMap;
+  private Map<HostAndPort, AtomicLong> commandStatusReportSentBytes;
+  private Map<HostAndPort, AtomicLong> commandStatusReportRequeuedBytes;
+  private Map<HostAndPort, AtomicLong> oversizedCommandStatusReportBytes;
 
   public DatanodeQueueMetrics(DatanodeStateMachine datanodeStateMachine) {
     this.registry = new MetricsRegistry(METRICS_SOURCE_NAME);
@@ -110,6 +126,13 @@ private void initializeQueues() {
     incrementalReportsQueueMap = new HashMap<>();
     containerActionQueueMap = new HashMap<>();
     pipelineActionQueueMap = new HashMap<>();
+    commandStatusReportQueueBytesMap = new HashMap<>();
+    commandStatusReportSentBytesMap = new HashMap<>();
+    commandStatusReportRequeuedBytesMap = new HashMap<>();
+    oversizedCommandStatusReportBytesMap = new HashMap<>();
+    commandStatusReportSentBytes = new HashMap<>();
+    commandStatusReportRequeuedBytes = new HashMap<>();
+    oversizedCommandStatusReportBytes = new HashMap<>();
   }
 
   @Override
@@ -150,6 +173,28 @@ public void getMetrics(MetricsCollector collector, boolean b) {
           datanodeStateMachine.getContext().getPipelineActionQueueSize()
               .getOrDefault(entry.getKey(), 0));
     }
+    Map<HostAndPort, Long> queuedCommandStatusReportBytes =
+        datanodeStateMachine.getContext().getCommandStatusReportQueueBytes();
+    for (Map.Entry<HostAndPort, MetricsInfo> entry:
+        commandStatusReportQueueBytesMap.entrySet()) {
+      builder.addGauge(entry.getValue(),
+          queuedCommandStatusReportBytes.getOrDefault(entry.getKey(), 0L));
+    }
+    for (Map.Entry<HostAndPort, MetricsInfo> entry:
+        commandStatusReportSentBytesMap.entrySet()) {
+      builder.addCounter(entry.getValue(),
+          commandStatusReportSentBytes.get(entry.getKey()).get());
+    }
+    for (Map.Entry<HostAndPort, MetricsInfo> entry:
+        commandStatusReportRequeuedBytesMap.entrySet()) {
+      builder.addCounter(entry.getValue(),
+          commandStatusReportRequeuedBytes.get(entry.getKey()).get());
+    }
+    for (Map.Entry<HostAndPort, MetricsInfo> entry:
+        oversizedCommandStatusReportBytesMap.entrySet()) {
+      builder.addGauge(entry.getValue(),
+          oversizedCommandStatusReportBytes.get(entry.getKey()).get());
+    }
   }
 
   public static synchronized void unRegister() {
@@ -158,24 +203,84 @@ public static synchronized void unRegister() {
   }
 
   public void addEndpoint(HostAndPort endpoint) {
+    String endpointName = CaseFormat.UPPER_UNDERSCORE
+        .to(CaseFormat.UPPER_CAMEL, endpoint.getHostName());
     incrementalReportsQueueMap.computeIfAbsent(endpoint,
         k -> getMetricsInfo(INCREMENTAL_REPORT_QUEUE_PREFIX,
-            CaseFormat.UPPER_UNDERSCORE
-                .to(CaseFormat.UPPER_CAMEL, k.getHostName())));
+            endpointName));
     containerActionQueueMap.computeIfAbsent(endpoint,
         k -> getMetricsInfo(CONTAINER_ACTION_QUEUE_PREFIX,
-            CaseFormat.UPPER_UNDERSCORE
-                .to(CaseFormat.UPPER_CAMEL, k.getHostName())));
+            endpointName));
     pipelineActionQueueMap.computeIfAbsent(endpoint,
         k -> getMetricsInfo(PIPELINE_ACTION_QUEUE_PREFIX,
-            CaseFormat.UPPER_UNDERSCORE
-                .to(CaseFormat.UPPER_CAMEL, k.getHostName())));
+            endpointName));
+    commandStatusReportQueueBytesMap.computeIfAbsent(endpoint,
+        k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_QUEUE_PREFIX,
+            endpointName, "Queued command status report bytes for "));
+    commandStatusReportSentBytesMap.computeIfAbsent(endpoint,
+        k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_SENT_PREFIX,
+            endpointName, "Command status report bytes sent to "));
+    commandStatusReportRequeuedBytesMap.computeIfAbsent(endpoint,
+        k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_REQUEUED_PREFIX,
+            endpointName, "Command status report bytes requeued for "));
+    oversizedCommandStatusReportBytesMap.computeIfAbsent(endpoint,
+        k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_OVERSIZED_PREFIX,
+            endpointName, "Oversized command status report bytes for "));
+    commandStatusReportSentBytes.computeIfAbsent(endpoint,
+        k -> new AtomicLong());
+    commandStatusReportRequeuedBytes.computeIfAbsent(endpoint,
+        k -> new AtomicLong());
+    oversizedCommandStatusReportBytes.computeIfAbsent(endpoint,
+        k -> new AtomicLong());
   }
 
   public void removeEndpoint(HostAndPort endpoint) {
     incrementalReportsQueueMap.remove(endpoint);
     containerActionQueueMap.remove(endpoint);
     pipelineActionQueueMap.remove(endpoint);
+    commandStatusReportQueueBytesMap.remove(endpoint);
+    commandStatusReportSentBytesMap.remove(endpoint);
+    commandStatusReportRequeuedBytesMap.remove(endpoint);
+    oversizedCommandStatusReportBytesMap.remove(endpoint);
+    commandStatusReportSentBytes.remove(endpoint);
+    commandStatusReportRequeuedBytes.remove(endpoint);
+    oversizedCommandStatusReportBytes.remove(endpoint);
+  }
+
+  public void incrCommandStatusReportBytesSent(
+      HostAndPort endpoint, long bytes) {
+    commandStatusReportSentBytes.computeIfAbsent(endpoint,
+        k -> new AtomicLong()).addAndGet(bytes);
+  }
+
+  public void incrCommandStatusReportBytesRequeued(
+      HostAndPort endpoint, long bytes) {
+    commandStatusReportRequeuedBytes.computeIfAbsent(endpoint,
+        k -> new AtomicLong()).addAndGet(bytes);
+  }
+
+  public void setOversizedCommandStatusReportBytes(
+      HostAndPort endpoint, long bytes) {
+    oversizedCommandStatusReportBytes.computeIfAbsent(endpoint,
+        k -> new AtomicLong()).set(bytes);
+  }
+
+  @VisibleForTesting
+  public long getCommandStatusReportBytesSent(HostAndPort endpoint) {
+    AtomicLong value = commandStatusReportSentBytes.get(endpoint);
+    return value == null ? 0 : value.get();
+  }
+
+  @VisibleForTesting
+  public long getCommandStatusReportBytesRequeued(HostAndPort endpoint) {
+    AtomicLong value = commandStatusReportRequeuedBytes.get(endpoint);
+    return value == null ? 0 : value.get();
+  }
+
+  @VisibleForTesting
+  public long getOversizedCommandStatusReportBytes(HostAndPort endpoint) {
+    AtomicLong value = oversizedCommandStatusReportBytes.get(endpoint);
+    return value == null ? 0 : value.get();
   }
 
   @VisibleForTesting
@@ -198,4 +303,10 @@ private MetricsInfo getMetricsInfo(String prefix, String metricName) {
     String description = "Queue size of " + metricName + " from " + prefix;
     return info(metric, description);
   }
+
+  private MetricsInfo getByteMetricsInfo(String prefix, String metricName,
+      String description) {
+    String metric = prefix + WordUtils.capitalize(metricName) + "Bytes";
+    return info(metric, description + metricName);
+  }
 }