HDDS-6567. Store datanode command queue counts from heartbeat in DatanodeInfo in SCM

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java

@@ -284,7 +284,7 @@ private void addQueuedCommandCounts(
       reportProto.addCommand(entry.getKey())
           .addCount(entry.getValue());
     }
-    requestBuilder.setQueuedCommandReport(reportProto.build());
+    requestBuilder.setCommandQueueReport(reportProto.build());
   }
 
   /**

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java

@@ -256,8 +256,8 @@ public void testheartbeatWithAllReports() throws Exception {
     Assert.assertTrue(heartbeat.hasContainerReport());
     Assert.assertTrue(heartbeat.getCommandStatusReportsCount() != 0);
     Assert.assertTrue(heartbeat.hasContainerActions());
-    Assert.assertTrue(heartbeat.hasQueuedCommandReport());
-    CommandQueueReportProto queueCount = heartbeat.getQueuedCommandReport();
+    Assert.assertTrue(heartbeat.hasCommandQueueReport());
+    CommandQueueReportProto queueCount = heartbeat.getCommandQueueReport();
     Assert.assertEquals(queueCount.getCommandCount(), commands.size());
     Assert.assertEquals(queueCount.getCountCount(), commands.size());
     for (int i = 0; i < commands.size(); i++) {

hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto

@@ -138,7 +138,7 @@ message SCMHeartbeatRequestProto {
   optional PipelineActionsProto pipelineActions = 7;
   optional PipelineReportsProto pipelineReports = 8;
   optional LayoutVersionProto dataNodeLayoutVersion = 9;
-  optional CommandQueueReportProto queuedCommandReport = 10;
+  optional CommandQueueReportProto commandQueueReport = 10;
 }
 
 message CommandQueueReportProto {
@@ -309,6 +309,7 @@ message PipelineAction {
  */
 message SCMCommandProto {
   enum Type {
+    unknownScmCommand = 0;
     reregisterCommand = 1;
     deleteBlocksCommand = 2;
     closeContainerCommand = 3;

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/events/SCMEvents.java

@@ -32,6 +32,7 @@
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.NodeReportFromDatanode;
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.PipelineActionsFromDatanode;
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.PipelineReportFromDatanode;
+import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.CommandQueueReportFromDatanode;
 import org.apache.hadoop.hdds.scm.server.SCMDatanodeProtocolServer.NodeRegistrationContainerReport;
 import org.apache.hadoop.hdds.server.events.Event;
 import org.apache.hadoop.hdds.server.events.TypedEvent;
@@ -49,6 +50,15 @@ public final class SCMEvents {
   public static final TypedEvent<NodeReportFromDatanode> NODE_REPORT =
       new TypedEvent<>(NodeReportFromDatanode.class, "Node_Report");
 
+  /**
+   * Queued Command counts are sent out by Datanodes. This report is received by
+   * SCMDatanodeHeartbeatDispatcher and the COMMAND_QUEUE_REPORT Event is
+   * generated.
+   */
+  public static final TypedEvent<CommandQueueReportFromDatanode>
+      COMMAND_QUEUE_REPORT = new TypedEvent<>(
+          CommandQueueReportFromDatanode.class, "Command_Queue_Report");
+
   /**
    * Event generated on DataNode registration.
    */

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/CommandQueueReportHandler.java

@@ -0,0 +1,49 @@
+/**
+ * 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.hadoop.hdds.scm.node;
+
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher.CommandQueueReportFromDatanode;
+import org.apache.hadoop.hdds.server.events.EventHandler;
+import org.apache.hadoop.hdds.server.events.EventPublisher;
+
+/**
+ * Handles QueuedCommand Reports from datanode.
+ */
+public class CommandQueueReportHandler implements
+    EventHandler<CommandQueueReportFromDatanode> {
+
+  private final NodeManager nodeManager;
+
+  public CommandQueueReportHandler(NodeManager nodeManager) {
+    Preconditions.checkNotNull(nodeManager);
+    this.nodeManager = nodeManager;
+  }
+
+  @Override
+  public void onMessage(CommandQueueReportFromDatanode queueReportFromDatanode,
+                        EventPublisher publisher) {
+    Preconditions.checkNotNull(queueReportFromDatanode);
+    DatanodeDetails dn = queueReportFromDatanode.getDatanodeDetails();
+    Preconditions.checkNotNull(dn, "QueueReport is "
+        + "missing DatanodeDetails.");
+    nodeManager.processNodeCommandQueueReport(dn,
+        queueReportFromDatanode.getReport());
+  }
+}

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java

@@ -22,16 +22,22 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.LayoutVersionProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.StorageReportProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.MetadataStorageReportProto;
 import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
@@ -41,6 +47,8 @@
  */
 public class DatanodeInfo extends DatanodeDetails {
 
+  private static final Logger LOG = LoggerFactory.getLogger(DatanodeInfo.class);
+
   private final ReadWriteLock lock;
 
   private volatile long lastHeartbeatTime;
@@ -49,6 +57,7 @@ public class DatanodeInfo extends DatanodeDetails {
   private List<StorageReportProto> storageReports;
   private List<MetadataStorageReportProto> metadataStorageReports;
   private LayoutVersionProto lastKnownLayoutVersion;
+  private Map<SCMCommandProto.Type, Integer> commandCounts;
 
   private NodeStatus nodeStatus;
 
@@ -69,6 +78,7 @@ public DatanodeInfo(DatanodeDetails datanodeDetails, NodeStatus nodeStatus,
     this.storageReports = Collections.emptyList();
     this.nodeStatus = nodeStatus;
     this.metadataStorageReports = Collections.emptyList();
+    this.commandCounts = new HashMap<>();
   }
 
   /**
@@ -272,6 +282,57 @@ public void setNodeStatus(NodeStatus newNodeStatus) {
     }
   }
 
+  /**
+   * Set the current command counts for this datanode, as reported in the last
+   * heartbeat.
+   * @param cmds Proto message containing a list of command count pairs.
+   */
+  public void setCommandCounts(CommandQueueReportProto cmds) {
+    try {
+      int count = cmds.getCommandCount();
+      lock.writeLock().lock();
+      for (int i = 0; i < count; i++) {
+        SCMCommandProto.Type command = cmds.getCommand(i);
+        if (command == SCMCommandProto.Type.unknownScmCommand) {
+          LOG.warn("Unknown SCM Command received from {} in the "
+              + "heartbeat. SCM and the DN may not be at the same version.",
+              this);
+          continue;
+        }
+        int cmdCount = cmds.getCount(i);
+        if (cmdCount < 0) {
+          LOG.warn("Command count of {} from {} should be greater than zero. " +
+              "Setting it to zero", cmdCount, this);
+          cmdCount = 0;
+        }
+        commandCounts.put(command, cmdCount);
+      }
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+
+  /**
+   * Retrieve the number of queued commands of the given type, as reported by
+   * the datanode at the last heartbeat.
+   * @param cmd The command for which to receive the queued command count
+   * @return -1 if we have no information about the count, or an integer >= 0
+   *         indicating the command count at the last heartbeat.
+   */
+  public int getCommandCount(SCMCommandProto.Type cmd) {
+    try {
+      lock.readLock().lock();
+      Integer count = commandCounts.get(cmd);
+      if (count == null) {
+        return -1;
+      } else {
+        return count.intValue();
+      }
+    } finally {
+      lock.readLock().unlock();
+    }
+  }
+
   @Override
   public int hashCode() {
     return super.hashCode();

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java

@@ -20,9 +20,11 @@
 import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.defaultLayoutVersionProto;
 
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
@@ -296,6 +298,24 @@ void processNodeReport(DatanodeDetails datanodeDetails,
   void processLayoutVersionReport(DatanodeDetails datanodeDetails,
                          LayoutVersionProto layoutReport);
 
+  /**
+   * Process the Command Queue Report sent from datanodes as part of the
+   * heartbeat message.
+   * @param datanodeDetails
+   * @param commandReport
+   */
+  void processNodeCommandQueueReport(DatanodeDetails datanodeDetails,
+      CommandQueueReportProto commandReport);
+
+  /**
+   * Get the number of commands of the given type queued on the datanode at the
+   * last heartbeat. If the Datanode has not reported information for the given
+   * command type, -1 will be returned.
+   * @param cmdType
+   * @return The queued count or -1 if no data has been received from the DN.
+   */
+  int getNodeQueuedCommandCount(DatanodeDetails datanodeDetails,
+      SCMCommandProto.Type cmdType) throws NodeNotFoundException;
 
   /**
    * Get list of SCMCommands in the Command Queue for a particular Datanode.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java

@@ -27,6 +27,8 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
@@ -621,6 +623,48 @@ public void processLayoutVersionReport(DatanodeDetails datanodeDetails,
     }
   }
 
+  /**
+   * Process Command Queue Reports from the Datanode Heartbeat.
+   *
+   * @param datanodeDetails
+   * @param commandQueueReportProto
+   */
+  @Override
+  public void processNodeCommandQueueReport(DatanodeDetails datanodeDetails,
+      CommandQueueReportProto commandQueueReportProto) {
+    LOG.debug("Processing Command Queue Report from [datanode={}]",
+        datanodeDetails.getHostName());
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Command Queue Report is received from [datanode={}]: " +
+          "<json>{}</json>", datanodeDetails.getHostName(),
+          commandQueueReportProto.toString().replaceAll("\n", "\\\\n"));
+    }
+    try {
+      DatanodeInfo datanodeInfo = nodeStateManager.getNode(datanodeDetails);
+      if (commandQueueReportProto != null) {
+        datanodeInfo.setCommandCounts(commandQueueReportProto);
+        metrics.incNumNodeCommandQueueReportProcessed();
+      }
+    } catch (NodeNotFoundException e) {
+      metrics.incNumNodeCommandQueueReportProcessingFailed();
+      LOG.warn("Got Command Queue Report from unregistered datanode {}",
+          datanodeDetails);
+    }
+  }
+
+  /**
+   * Get the number of commands of the given type queued on the datanode at the
+   * last heartbeat. If the Datanode has not reported information for the given
+   * command type, -1 will be returned.
+   * @param cmdType
+   * @return The queued count or -1 if no data has been received from the DN.
+   */
+  public int getNodeQueuedCommandCount(DatanodeDetails datanodeDetails,
+      SCMCommandProto.Type cmdType) throws NodeNotFoundException {
+    DatanodeInfo datanodeInfo = nodeStateManager.getNode(datanodeDetails);
+    return datanodeInfo.getCommandCount(cmdType);
+  }
+
   /**
    * Returns the aggregated node stats.
    *

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeMetrics.java

@@ -50,6 +50,8 @@ public final class SCMNodeMetrics implements MetricsSource {
   private @Metric MutableCounterLong numHBProcessingFailed;
   private @Metric MutableCounterLong numNodeReportProcessed;
   private @Metric MutableCounterLong numNodeReportProcessingFailed;
+  private @Metric MutableCounterLong numNodeCommandQueueReportProcessed;
+  private @Metric MutableCounterLong numNodeCommandQueueReportProcessingFailed;
   private @Metric String textMetric;
 
   private final MetricsRegistry registry;
@@ -111,6 +113,20 @@ void incNumNodeReportProcessingFailed() {
     numNodeReportProcessingFailed.incr();
   }
 
+  /**
+   * Increments number of Command Queue reports processed.
+   */
+  void incNumNodeCommandQueueReportProcessed() {
+    numNodeCommandQueueReportProcessed.incr();
+  }
+
+  /**
+   * Increments number of Command Queue reports where processing failed.
+   */
+  void incNumNodeCommandQueueReportProcessingFailed() {
+    numNodeCommandQueueReportProcessingFailed.incr();
+  }
+
   /**
    * Get aggregated counter and gauge metrics.
    */

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeHeartbeatDispatcher.java

@@ -19,6 +19,7 @@
 
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.CRLStatusReport;
 import org.apache.hadoop.hdds.protocol.proto
@@ -58,6 +59,7 @@
 import static org.apache.hadoop.hdds.scm.events.SCMEvents.CMD_STATUS_REPORT;
 import static org.apache.hadoop.hdds.scm.events.SCMEvents.PIPELINE_ACTIONS;
 import static org.apache.hadoop.hdds.scm.events.SCMEvents.PIPELINE_REPORT;
+import static org.apache.hadoop.hdds.scm.events.SCMEvents.COMMAND_QUEUE_REPORT;
 import static org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature.INITIAL_VERSION;
 import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.toLayoutVersionProto;
 
@@ -132,6 +134,13 @@ public List<SCMCommand> dispatch(SCMHeartbeatRequestProto heartbeat) {
                 heartbeat.getNodeReport()));
       }
 
+      if (heartbeat.hasCommandQueueReport()) {
+        LOG.debug("Dispatching Queued Command Report");
+        eventPublisher.fireEvent(COMMAND_QUEUE_REPORT,
+            new CommandQueueReportFromDatanode(datanodeDetails,
+                heartbeat.getCommandQueueReport()));
+      }
+
       if (heartbeat.hasContainerReport()) {
         LOG.debug("Dispatching Container Report.");
         eventPublisher.fireEvent(
@@ -233,6 +242,17 @@ public NodeReportFromDatanode(DatanodeDetails datanodeDetails,
     }
   }
 
+  /**
+   * Command Queue Report with origin.
+   */
+  public static class CommandQueueReportFromDatanode
+      extends ReportFromDatanode<CommandQueueReportProto> {
+    public CommandQueueReportFromDatanode(DatanodeDetails datanodeDetails,
+                                          CommandQueueReportProto report) {
+      super(datanodeDetails, report);
+    }
+  }
+
   /**
    * Layout report event payload with origin.
    */