HDDS-2642. Expose decommission / maintenance metrics via JMX

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

@@ -30,5 +30,5 @@ public interface DatanodeAdminMonitor extends Runnable {
   void startMonitoring(DatanodeDetails dn);
   void stopMonitoring(DatanodeDetails dn);
   Set<DatanodeDetails> getTrackedNodes();
-
+  void setMetrics(NodeDecommissionMetrics metrics);
 }

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hdds.scm.node;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
@@ -38,8 +39,10 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -74,6 +77,51 @@ public class DatanodeAdminMonitorImpl implements DatanodeAdminMonitor {
   private Queue<DatanodeDetails> pendingNodes = new ArrayDeque();
   private Queue<DatanodeDetails> cancelledNodes = new ArrayDeque();
   private Set<DatanodeDetails> trackedNodes = new HashSet<>();
+  private NodeDecommissionMetrics metrics;
+  private long pipelinesWaitingToClose = 0;
+  private long sufficientlyReplicatedContainers = 0;
+  private long trackedDecomMaintenance = 0;
+  private long trackedRecommission = 0;
+  private long unhealthyContainers = 0;
+  private long underReplicatedContainers = 0;
+
+  @SuppressFBWarnings(value = "SIC_INNER_SHOULD_BE_STATIC")
+  private final class ContainerStateInWorkflow {
+    private long sufficientlyReplicated = 0;
+    private long unhealthyContainers = 0;
+    private long underReplicatedContainers = 0;
+    private String host = "";
+
+    private ContainerStateInWorkflow(String host,
+                                     long sufficientlyReplicated,
+                                    long underReplicatedContainers,
+                                    long unhealthyContainers) {
+      this.host = host;
+      this.sufficientlyReplicated = sufficientlyReplicated;
+      this.unhealthyContainers = unhealthyContainers;
+      this.underReplicatedContainers = underReplicatedContainers;
+    }
+
+    public void setAll(long sufficiently,
+                    long under,
+                    long unhealthy) {
+      sufficientlyReplicated = sufficiently;
+      underReplicatedContainers = under;
+      unhealthyContainers = unhealthy;
+    }
+    public void reset() {
+      sufficientlyReplicated = 0L;
+      underReplicatedContainers = 0L;
+      unhealthyContainers = 0L;
+    }
+
+    public String getHost() {
+      return host;
+    }
+  }
+
+  private Map<String, ContainerStateInWorkflow> containerStateByHost;
+  private Map<String, Long> pipelinesWaitingToCloseByHost;
 
   private static final Logger LOG =
       LoggerFactory.getLogger(DatanodeAdminMonitorImpl.class);
@@ -90,6 +138,9 @@ public DatanodeAdminMonitorImpl(
     this.eventQueue = eventQueue;
     this.nodeManager = nodeManager;
     this.replicationManager = replicationManager;
+
+    containerStateByHost = new HashMap<>();
+    pipelinesWaitingToCloseByHost = new HashMap<>();
   }
 
   /**
@@ -117,6 +168,10 @@ public synchronized void stopMonitoring(DatanodeDetails dn) {
     cancelledNodes.add(dn);
   }
 
+  public synchronized void setMetrics(NodeDecommissionMetrics metrics) {
+    this.metrics = metrics;
+  }
+
   /**
    * Get the set of nodes which are currently tracked in the decommissioned
    * and maintenance workflow.
@@ -140,15 +195,18 @@ public synchronized Set<DatanodeDetails> getTrackedNodes() {
   public void run() {
     try {
       synchronized (this) {
+        trackedRecommission = getCancelledCount();
         processCancelledNodes();
         processPendingNodes();
+        trackedDecomMaintenance = getTrackedNodeCount();
       }
       processTransitioningNodes();
       if (trackedNodes.size() > 0 || pendingNodes.size() > 0) {
         LOG.info("There are {} nodes tracked for decommission and " +
                 "maintenance. {} pending nodes.",
             trackedNodes.size(), pendingNodes.size());
       }
+      setMetricsToGauge();
     } catch (Exception e) {
       LOG.error("Caught an error in the DatanodeAdminMonitor", e);
       // Intentionally do not re-throw, as if we do the monitor thread
@@ -168,6 +226,43 @@ public int getTrackedNodeCount() {
     return trackedNodes.size();
   }
 
+  synchronized void setMetricsToGauge() {
+    metrics.setTrackedContainersUnhealthyTotal(unhealthyContainers);
+    metrics.setTrackedRecommissionNodesTotal(trackedRecommission);
+    metrics.setTrackedDecommissioningMaintenanceNodesTotal(
+            trackedDecomMaintenance);
+    metrics.setTrackedContainersUnderReplicatedTotal(
+            underReplicatedContainers);
+    metrics.setTrackedContainersSufficientlyReplicatedTotal(
+            sufficientlyReplicatedContainers);
+    metrics.setTrackedPipelinesWaitingToCloseTotal(pipelinesWaitingToClose);
+    for (Map.Entry<String, Long> e :
+            pipelinesWaitingToCloseByHost.entrySet()) {
+      metrics.metricRecordPipelineWaitingToCloseByHost(e.getKey(),
+              e.getValue());
+    }
+    for (Map.Entry<String, ContainerStateInWorkflow> e :
+            containerStateByHost.entrySet()) {
+      metrics.metricRecordOfReplicationByHost(e.getKey(),
+              e.getValue().sufficientlyReplicated,
+              e.getValue().underReplicatedContainers,
+              e.getValue().unhealthyContainers);
+    }
+  }
+
+  void resetContainerMetrics() {
+    pipelinesWaitingToClose = 0;
+    sufficientlyReplicatedContainers = 0;
+    unhealthyContainers = 0;
+    underReplicatedContainers = 0;
+
+    for (Map.Entry<String, ContainerStateInWorkflow> e :
+            containerStateByHost.entrySet()) {
+      e.getValue().reset();
+    }
+    pipelinesWaitingToCloseByHost.replaceAll((k, v) -> 0L);
+  }
+
   private void processCancelledNodes() {
     while (!cancelledNodes.isEmpty()) {
       DatanodeDetails dn = cancelledNodes.poll();
@@ -188,7 +283,9 @@ private void processPendingNodes() {
   }
 
   private void processTransitioningNodes() {
+    resetContainerMetrics();
     Iterator<DatanodeDetails> iterator = trackedNodes.iterator();
+
     while (iterator.hasNext()) {
       DatanodeDetails dn = iterator.next();
       try {
@@ -278,6 +375,9 @@ private boolean checkPipelinesClosedOnNode(DatanodeDetails dn)
     } else {
       LOG.info("Waiting for pipelines to close for {}. There are {} " +
           "pipelines", dn, pipelines.size());
+      pipelinesWaitingToCloseByHost.put(dn.getHostName(),
+              (long) pipelines.size());
+      pipelinesWaitingToClose += pipelines.size();
       return false;
     }
   }
@@ -327,6 +427,17 @@ private boolean checkContainersReplicatedOnNode(DatanodeDetails dn)
     LOG.info("{} has {} sufficientlyReplicated, {} underReplicated and {} " +
         "unhealthy containers",
         dn, sufficientlyReplicated, underReplicated, unhealthy);
+    containerStateByHost.computeIfAbsent(dn.getHostName(),
+            hostID -> new ContainerStateInWorkflow(hostID,
+                    0L,
+                    0L,
+                    0L)
+    ).setAll(sufficientlyReplicated,
+            underReplicated,
+            unhealthy);
+    sufficientlyReplicatedContainers += sufficientlyReplicated;
+    underReplicatedContainers += underReplicated;
+    unhealthyContainers += unhealthy;
     if (LOG.isDebugEnabled() && underReplicatedIDs.size() < 10000 &&
         unhealthyIDs.size() < 10000) {
       LOG.debug("{} has {} underReplicated [{}] and {} unhealthy [{}] " +

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

@@ -60,6 +60,9 @@ public class NodeDecommissionManager {
   private boolean useHostnames;
   private long monitorInterval;
 
+  // Decommissioning and Maintenance mode progress related metrics.
+  private NodeDecommissionMetrics metrics;
+
   private static final Logger LOG =
       LoggerFactory.getLogger(NodeDecommissionManager.class);
 
@@ -181,6 +184,7 @@ public NodeDecommissionManager(OzoneConfiguration config, NodeManager nm,
     this.scmContext = scmContext;
     this.eventQueue = eventQueue;
     this.replicationManager = rm;
+    this.metrics = null;
 
     executor = Executors.newScheduledThreadPool(1,
         new ThreadFactoryBuilder().setNameFormat("DatanodeAdminManager-%d")
@@ -208,7 +212,8 @@ public NodeDecommissionManager(OzoneConfiguration config, NodeManager nm,
 
     monitor = new DatanodeAdminMonitorImpl(conf, eventQueue, nodeManager,
         replicationManager);
-
+    this.metrics = NodeDecommissionMetrics.create();
+    monitor.setMetrics(this.metrics);
     executor.scheduleAtFixedRate(monitor, monitorInterval, monitorInterval,
         TimeUnit.SECONDS);
   }
@@ -373,6 +378,7 @@ public synchronized void startMaintenance(DatanodeDetails dn, int endInHours)
    *  Stops the decommission monitor from running when SCM is shutdown.
    */
   public void stop() {
+    metrics.unRegister();
     if (executor != null) {
       executor.shutdown();
     }