HADOOP-3193. Include the address of the client that found the corrupted block

in the log. Also include a CorruptedBlocks metric to track the size of the
corrupted block map.



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@664155 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -81,6 +81,10 @@ Trunk (unreleased changes)
     HADOOP-3452. Change fsck to return non-zero status for a corrupt
     FileSystem. (lohit vijayarenu via cdouglas)
 
+    HADOOP-3193. Include the address of the client that found the corrupted
+    block in the log. Also include a CorruptedBlocks metric to track the size
+    of the corrupted block map. (cdouglas)
+
   NEW FEATURES
 
     HADOOP-3074. Provides a UrlStreamHandler for DFS and other FS,

src/java/org/apache/hadoop/dfs/CorruptReplicasMap.java

@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.dfs;
 
+import org.apache.hadoop.ipc.Server;
+
 import java.util.*;
 import java.io.IOException;
 
@@ -50,13 +52,17 @@ class CorruptReplicasMap{
       nodes.add(dn);
       NameNode.stateChangeLog.info("BLOCK NameSystem.addToCorruptReplicasMap: "+
                                    blk.getBlockName() +
-                                   " added as corrupt on " + dn.getName());
+                                   " added as corrupt on " + dn.getName() +
+                                   " by " + Server.getRemoteIp());
     } else {
       NameNode.stateChangeLog.info("BLOCK NameSystem.addToCorruptReplicasMap: "+
                                    "duplicate requested for " + 
                                    blk.getBlockName() + " to add as corrupt " +
-                                   "on " + dn.getName());
+                                   "on " + dn.getName() +
+                                   " by " + Server.getRemoteIp());
     }
+    NameNode.getNameNodeMetrics().numBlocksCorrupted.set(
+        corruptReplicasMap.size());
   }
 
   /**
@@ -68,8 +74,11 @@ class CorruptReplicasMap{
     FSNamesystem fsNamesystem = FSNamesystem.getFSNamesystem();
     if (fsNamesystem.blocksMap.contains(blk))
       return;
-    if (corruptReplicasMap != null)
+    if (corruptReplicasMap != null) {
       corruptReplicasMap.remove(blk);
+      NameNode.getNameNodeMetrics().numBlocksCorrupted.set(
+          corruptReplicasMap.size());
+    }
   }
 
   /**

src/java/org/apache/hadoop/dfs/NameNodeMetrics.java

@@ -57,6 +57,7 @@ public class NameNodeMetrics implements Updater {
     public MetricsIntValue safeModeTime = new MetricsIntValue("SafemodeTime");
     public MetricsIntValue fsImageLoadTime = 
                                         new MetricsIntValue("fsImageLoadTime");
+    public MetricsIntValue numBlocksCorrupted = new MetricsIntValue("BlocksCorrupted");
 
       
     NameNodeMetrics(Configuration conf, NameNode nameNode) {
@@ -101,6 +102,7 @@ public class NameNodeMetrics implements Updater {
         blockReport.pushMetric(metricsRecord);
         safeModeTime.pushMetric(metricsRecord);
         fsImageLoadTime.pushMetric(metricsRecord);
+        numBlocksCorrupted.pushMetric(metricsRecord);
       }
       metricsRecord.update();
     }