HDFS-15717. Improve fsck logging. Contributed by Kihwal Lee and Ahmed Hussein (#2620)

(cherry picked from commit be35fa186cc0c5c1f90028a9383851d47791bf04)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -5461,13 +5461,19 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
   private static UserGroupInformation getRemoteUser() throws IOException {
     return NameNode.getRemoteUser();
   }
-  
+
   /**
-   * Log fsck event in the audit log 
+   * Log fsck event in the audit log.
+   *
+   * @param succeeded Whether authorization succeeded.
+   * @param src Path of affected source file.
+   * @param remoteAddress Remote address of the request.
+   * @throws IOException if {@link #getRemoteUser()} fails.
    */
-  void logFsckEvent(String src, InetAddress remoteAddress) throws IOException {
+  void logFsckEvent(boolean succeeded, String src, InetAddress remoteAddress)
+      throws IOException {
     if (isAuditEnabled()) {
-      logAuditEvent(true, getRemoteUser(),
+      logAuditEvent(succeeded, getRemoteUser(),
                     remoteAddress,
                     "fsck", src, null, null);
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FsckServlet.java

@@ -59,15 +59,22 @@ public class FsckServlet extends DfsServlet {
         @Override
         public Object run() throws Exception {
           NameNode nn = NameNodeHttpServer.getNameNodeFromContext(context);
-          
+
           final FSNamesystem namesystem = nn.getNamesystem();
           final BlockManager bm = namesystem.getBlockManager();
-          final int totalDatanodes = 
-              namesystem.getNumberOfDatanodes(DatanodeReportType.LIVE); 
-          new NamenodeFsck(conf, nn,
+          final int totalDatanodes =
+              namesystem.getNumberOfDatanodes(DatanodeReportType.LIVE);
+          NamenodeFsck fsck = new NamenodeFsck(conf, nn,
               bm.getDatanodeManager().getNetworkTopology(), pmap, out,
-              totalDatanodes, remoteAddress).fsck();
-          
+              totalDatanodes, remoteAddress);
+          String auditSource = fsck.getAuditSource();
+          boolean success = false;
+          try {
+            fsck.fsck();
+            success = true;
+          } finally {
+            namesystem.logFsckEvent(success, auditSource, remoteAddress);
+          }
           return null;
         }
       });

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NamenodeFsck.java

@@ -144,6 +144,7 @@ public class NamenodeFsck implements DataEncryptionKeyFactory {
   private boolean showMaintenanceState = false;
   private long staleInterval;
   private Tracer tracer;
+  private String auditSource;
 
   /**
    * True if we encountered an internal error during FSCK, such as not being
@@ -167,7 +168,7 @@ public class NamenodeFsck implements DataEncryptionKeyFactory {
 
   String path = "/";
 
-  private String blockIds = null;
+  private String[] blockIds = null;
 
   // We return back N files that are corrupt; the list of files returned is
   // ordered by block id; to allow continuation support, pass in the last block
@@ -237,9 +238,15 @@ public class NamenodeFsck implements DataEncryptionKeyFactory {
       } else if (key.equals("includeSnapshots")) {
         this.snapshottableDirs = new ArrayList<String>();
       } else if (key.equals("blockId")) {
-        this.blockIds = pmap.get("blockId")[0];
+        this.blockIds = pmap.get("blockId")[0].split(" ");
       }
     }
+    this.auditSource = (blockIds != null)
+        ? "blocksIds=" + Arrays.asList(blockIds) : path;
+  }
+
+  public String getAuditSource() {
+    return auditSource;
   }
 
   /**
@@ -324,18 +331,18 @@ public class NamenodeFsck implements DataEncryptionKeyFactory {
   /**
    * Check files on DFS, starting from the indicated path.
    */
-  public void fsck() {
+  public void fsck() throws AccessControlException {
     final long startTime = Time.monotonicNow();
     try {
       if(blockIds != null) {
-        String[] blocks = blockIds.split(" ");
+        namenode.getNamesystem().checkSuperuserPrivilege();
         StringBuilder sb = new StringBuilder();
         sb.append("FSCK started by " +
             UserGroupInformation.getCurrentUser() + " from " +
             remoteAddress + " at " + new Date());
         out.println(sb);
         sb.append(" for blockIds: \n");
-        for (String blk: blocks) {
+        for (String blk: blockIds) {
           if(blk == null || !blk.contains(Block.BLOCK_FILE_PREFIX)) {
             out.println("Incorrect blockId format: " + blk);
             continue;
@@ -345,7 +352,6 @@ public class NamenodeFsck implements DataEncryptionKeyFactory {
           sb.append(blk + "\n");
         }
         LOG.info(sb);
-        namenode.getNamesystem().logFsckEvent("/", remoteAddress);
         out.flush();
         return;
       }
@@ -354,7 +360,6 @@ public class NamenodeFsck implements DataEncryptionKeyFactory {
           + " from " + remoteAddress + " for path " + path + " at " + new Date();
       LOG.info(msg);
       out.println(msg);
-      namenode.getNamesystem().logFsckEvent(path, remoteAddress);
 
       if (snapshottableDirs != null) {
         SnapshottableDirectoryStatus[] snapshotDirs =

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFsck.java

@@ -237,6 +237,7 @@ public class TestFsck {
       file.delete();
     }
     Logger logger = ((Log4JLogger) FSNamesystem.auditLog).getLogger();
+    logger.removeAllAppenders();
     logger.setLevel(Level.INFO);
     PatternLayout layout = new PatternLayout("%m%n");
     RollingFileAppender appender =