HDFS-7390. Provide JMX metrics per storage type. (Benoy Antony)

Author: Benoy Antony

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java

@@ -39,6 +39,8 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicLong;
 
+import javax.management.ObjectName;
+
 import org.apache.hadoop.HadoopIllegalArgumentException;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
@@ -85,6 +87,7 @@
 import org.apache.hadoop.hdfs.server.protocol.ReceivedDeletedBlockInfo;
 import org.apache.hadoop.hdfs.server.protocol.StorageReceivedDeletedBlocks;
 import org.apache.hadoop.hdfs.util.LightWeightLinkedSet;
+import org.apache.hadoop.metrics2.util.MBeans;
 import org.apache.hadoop.net.Node;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.Daemon;
@@ -94,14 +97,15 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Sets;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * Keeps information related to the blocks stored in the Hadoop cluster.
  */
 @InterfaceAudience.Private
-public class BlockManager {
+public class BlockManager implements BlockStatsMXBean {
 
   public static final Logger LOG = LoggerFactory.getLogger(BlockManager.class);
   public static final Logger blockLog = NameNode.blockStateChangeLog;
@@ -129,6 +133,7 @@ public class BlockManager {
   private final AtomicLong postponedMisreplicatedBlocksCount = new AtomicLong(0L);
   private final long startupDelayBlockDeletionInMs;
   private final BlockReportLeaseManager blockReportLeaseManager;
+  private ObjectName mxBeanName;
 
   /** Used by metrics */
   public long getPendingReplicationBlocksCount() {
@@ -468,6 +473,7 @@ public void activate(Configuration conf) {
     pendingReplications.start();
     datanodeManager.activate(conf);
     this.replicationThread.start();
+    mxBeanName = MBeans.register("NameNode", "BlockStats", this);
   }
 
   public void close() {
@@ -3944,6 +3950,8 @@ enum MisReplicationResult {
   public void shutdown() {
     stopReplicationInitializer();
     blocksMap.close();
+    MBeans.unregister(mxBeanName);
+    mxBeanName = null;
   }
 
   public void clear() {
@@ -3954,4 +3962,9 @@ public void clear() {
   public BlockReportLeaseManager getBlockReportLeaseManager() {
     return blockReportLeaseManager;
   }
+
+  @Override // BlockStatsMXBean
+  public Map<StorageType, StorageTypeStats> getStorageTypeStats() {
+    return  datanodeManager.getDatanodeStatistics().getStorageTypeStats();
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockStatsMXBean.java

@@ -0,0 +1,36 @@
+/**
+ * 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.hdfs.server.blockmanagement;
+
+import java.util.Map;
+
+import org.apache.hadoop.fs.StorageType;
+
+/**
+ * This is an interface used to retrieve statistic information related to
+ * block management.
+ */
+public interface BlockStatsMXBean {
+
+  /**
+   * The statistics of storage types.
+   *
+   * @return get storage statistics per storage type
+   */
+  Map<StorageType, StorageTypeStats> getStorageTypeStats();
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeStatistics.java

@@ -17,6 +17,9 @@
  */
 package org.apache.hadoop.hdfs.server.blockmanagement;
 
+import java.util.Map;
+
+import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.hdfs.protocol.ClientProtocol;
 
 /** Datanode statistics */
@@ -71,4 +74,7 @@ public interface DatanodeStatistics {
 
   /** @return the expired heartbeats */
   public int getExpiredHeartbeats();
+
+  /** @return Storage Tier statistics*/
+  Map<StorageType, StorageTypeStats> getStorageTypeStats();
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/HeartbeatManager.java

@@ -18,9 +18,15 @@
 package org.apache.hadoop.hdfs.server.blockmanagement;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.IdentityHashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DFSUtilClient;
 import org.apache.hadoop.hdfs.protocol.DatanodeID;
@@ -189,6 +195,11 @@ public synchronized int getExpiredHeartbeats() {
     return stats.expiredHeartbeats;
   }
 
+  @Override
+  public  Map<StorageType, StorageTypeStats> getStorageTypeStats() {
+    return stats.statsMap.get();
+  }
+
   synchronized void register(final DatanodeDescriptor d) {
     if (!d.isAlive) {
       addDatanode(d);
@@ -393,6 +404,9 @@ public void run() {
    * For decommissioning/decommissioned nodes, only used capacity is counted.
    */
   private static class Stats {
+
+    private final StorageTypeStatsMap statsMap = new StorageTypeStatsMap();
+
     private long capacityTotal = 0L;
     private long capacityUsed = 0L;
     private long capacityRemaining = 0L;
@@ -420,6 +434,14 @@ private void add(final DatanodeDescriptor node) {
       }
       cacheCapacity += node.getCacheCapacity();
       cacheUsed += node.getCacheUsed();
+      Set<StorageType> storageTypes = new HashSet<>();
+      for (DatanodeStorageInfo storageInfo : node.getStorageInfos()) {
+        statsMap.addStorage(storageInfo, node);
+        storageTypes.add(storageInfo.getStorageType());
+      }
+      for (StorageType storageType : storageTypes) {
+        statsMap.addNode(storageType, node);
+      }
     }
 
     private void subtract(final DatanodeDescriptor node) {
@@ -436,12 +458,84 @@ private void subtract(final DatanodeDescriptor node) {
       }
       cacheCapacity -= node.getCacheCapacity();
       cacheUsed -= node.getCacheUsed();
+      Set<StorageType> storageTypes = new HashSet<>();
+      for (DatanodeStorageInfo storageInfo : node.getStorageInfos()) {
+        statsMap.subtractStorage(storageInfo, node);
+        storageTypes.add(storageInfo.getStorageType());
+      }
+      for (StorageType storageType : storageTypes) {
+        statsMap.subtractNode(storageType, node);
+      }
     }
 
     /** Increment expired heartbeat counter. */
     private void incrExpiredHeartbeats() {
       expiredHeartbeats++;
     }
   }
-}
 
+  /** StorageType specific statistics.
+   * For decommissioning/decommissioned nodes, only used capacity is counted.
+   */
+
+  static final class StorageTypeStatsMap {
+
+    private Map<StorageType, StorageTypeStats> storageTypeStatsMap =
+          new IdentityHashMap<>();
+
+    private StorageTypeStatsMap() {}
+
+    private StorageTypeStatsMap(StorageTypeStatsMap other) {
+      storageTypeStatsMap =
+          new IdentityHashMap<>(other.storageTypeStatsMap);
+      for (Map.Entry<StorageType, StorageTypeStats> entry :
+          storageTypeStatsMap.entrySet()) {
+        entry.setValue(new StorageTypeStats(entry.getValue()));
+      }
+    }
+
+    private Map<StorageType, StorageTypeStats> get() {
+      return Collections.unmodifiableMap(storageTypeStatsMap);
+    }
+
+    private void addNode(StorageType storageType,
+        final DatanodeDescriptor node) {
+      StorageTypeStats storageTypeStats =
+          storageTypeStatsMap.get(storageType);
+      if (storageTypeStats == null) {
+        storageTypeStats = new StorageTypeStats();
+        storageTypeStatsMap.put(storageType, storageTypeStats);
+      }
+      storageTypeStats.addNode(node);
+    }
+
+    private void addStorage(final DatanodeStorageInfo info,
+        final DatanodeDescriptor node) {
+      StorageTypeStats storageTypeStats =
+          storageTypeStatsMap.get(info.getStorageType());
+      if (storageTypeStats == null) {
+        storageTypeStats = new StorageTypeStats();
+        storageTypeStatsMap.put(info.getStorageType(), storageTypeStats);
+      }
+      storageTypeStats.addStorage(info, node);
+    }
+
+    private void subtractStorage(final DatanodeStorageInfo info,
+        final DatanodeDescriptor node) {
+      StorageTypeStats storageTypeStats =
+          storageTypeStatsMap.get(info.getStorageType());
+      if (storageTypeStats != null) {
+        storageTypeStats.subtractStorage(info, node);
+      }
+    }
+
+    private void subtractNode(StorageType storageType,
+        final DatanodeDescriptor node) {
+      StorageTypeStats storageTypeStats =
+          storageTypeStatsMap.get(storageType);
+      if (storageTypeStats != null) {
+        storageTypeStats.subtractNode(node);
+      }
+    }
+  }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/StorageTypeStats.java

@@ -0,0 +1,115 @@
+/**
+ * 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.hdfs.server.blockmanagement;
+
+import java.beans.ConstructorProperties;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Statistics per StorageType.
+ *
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class StorageTypeStats {
+  private long capacityTotal = 0L;
+  private long capacityUsed = 0L;
+  private long capacityRemaining = 0L;
+  private long blockPoolUsed = 0L;
+  private int nodesInService = 0;
+
+  @ConstructorProperties({"capacityTotal",
+      "capacityUsed", "capacityRemaining",  "blockPoolUsed", "nodesInService"})
+  public StorageTypeStats(long capacityTotal, long capacityUsed,
+      long capacityRemaining, long blockPoolUsed, int nodesInService) {
+    this.capacityTotal = capacityTotal;
+    this.capacityUsed = capacityUsed;
+    this.capacityRemaining = capacityRemaining;
+    this.blockPoolUsed = blockPoolUsed;
+    this.nodesInService = nodesInService;
+  }
+
+  public long getCapacityTotal() {
+    return capacityTotal;
+  }
+
+  public long getCapacityUsed() {
+    return capacityUsed;
+  }
+
+  public long getCapacityRemaining() {
+    return capacityRemaining;
+  }
+
+  public long getBlockPoolUsed() {
+    return blockPoolUsed;
+  }
+
+  public int getNodesInService() {
+    return nodesInService;
+  }
+
+  StorageTypeStats() {}
+
+  StorageTypeStats(StorageTypeStats other) {
+    capacityTotal = other.capacityTotal;
+    capacityUsed = other.capacityUsed;
+    capacityRemaining = other.capacityRemaining;
+    blockPoolUsed = other.blockPoolUsed;
+    nodesInService = other.nodesInService;
+  }
+
+  void addStorage(final DatanodeStorageInfo info,
+      final DatanodeDescriptor node) {
+    capacityUsed += info.getDfsUsed();
+    blockPoolUsed += info.getBlockPoolUsed();
+    if (!(node.isDecommissionInProgress() || node.isDecommissioned())) {
+      capacityTotal += info.getCapacity();
+      capacityRemaining += info.getRemaining();
+    } else {
+      capacityTotal += info.getDfsUsed();
+    }
+  }
+
+  void addNode(final DatanodeDescriptor node) {
+    if (!(node.isDecommissionInProgress() || node.isDecommissioned())) {
+      nodesInService++;
+    }
+  }
+
+  void subtractStorage(final DatanodeStorageInfo info,
+      final DatanodeDescriptor node) {
+    capacityUsed -= info.getDfsUsed();
+    blockPoolUsed -= info.getBlockPoolUsed();
+    if (!(node.isDecommissionInProgress() || node.isDecommissioned())) {
+      capacityTotal -= info.getCapacity();
+      capacityRemaining -= info.getRemaining();
+    } else {
+      capacityTotal -= info.getDfsUsed();
+    }
+  }
+
+  void subtractNode(final DatanodeDescriptor node) {
+    if (!(node.isDecommissionInProgress() || node.isDecommissioned())) {
+      nodesInService--;
+    }
+  }
+}