HADOOP-18257 audit log parser

* tool is invoked through hadoop s3guard command
* which can now also be invoked as "hadoop s3a"!
* tests are improved.
* OperationDuration implements DurationTracker for bit more completeness

TODO
* split out the record parsing into a hadoop MR record read/write
* so support large scale parsing
* cli tool just glues that together either for small parallelised extraction
  or for aggregation to one file.
* but a bulk job would work with a larger dataset

Change-Id: I25e333592d1058b460b0bfda5313a20de13c2e35

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/DurationInfo.java

@@ -93,7 +93,7 @@ public String toString() {
 
   @Override
   public void close() {
-    finished();
+    super.close();
     if (logAtInfo) {
       log.info("{}", this);
     } else {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/OperationDuration.java

@@ -22,13 +22,14 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.statistics.DurationTracker;
 
 /**
  * Little duration counter.
  */
 @InterfaceAudience.Public
 @InterfaceStability.Unstable
-public class OperationDuration {
+public class OperationDuration implements DurationTracker {
 
   /**
    * Time in millis when the operation started.
@@ -65,6 +66,16 @@ public void finished() {
     finished = time();
   }
 
+  @Override
+  public void failed() {
+    finished();
+  }
+
+  @Override
+  public void close() {
+    finished();
+  }
+
   /**
    * Return the duration as {@link #humanTime(long)}.
    * @return a printable duration.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/audit/AuditTool.java

@@ -20,6 +20,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.PrintStream;
 import java.io.PrintWriter;
 import java.net.URI;
 import java.util.Arrays;
@@ -34,6 +35,7 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.audit.mapreduce.S3AAuditLogMergerAndParser;
+import org.apache.hadoop.fs.s3a.s3guard.S3GuardTool;
 import org.apache.hadoop.util.ExitUtil;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
@@ -47,10 +49,15 @@
  * Its functionality is to parse the audit log files
  * and generate avro file.
  */
-public class AuditTool extends Configured implements Tool, Closeable {
+public class AuditTool extends S3GuardTool {
 
   private static final Logger LOG = LoggerFactory.getLogger(AuditTool.class);
 
+  /**
+   * Name of audit tool: {@value}.
+   */
+  public static final String AUDIT = "audit";
+
   private final S3AAuditLogMergerAndParser s3AAuditLogMergerAndParser =
       new S3AAuditLogMergerAndParser();
 
@@ -82,8 +89,8 @@ public class AuditTool extends Configured implements Tool, Closeable {
 
   private PrintWriter out;
 
-  public AuditTool() {
-    super();
+  public AuditTool(final Configuration conf) {
+    super(conf);
   }
 
   /**
@@ -101,15 +108,19 @@ public String getName() {
 
   /**
    * This run method in AuditTool takes source and destination path of bucket,
-   * and check if there are directories and pass these paths to merge and
+   * and checks if there are directories and pass these paths to merge and
    * parse audit log files.
    *
    * @param args argument list
    * @return SUCCESS i.e, '0', which is an exit code
    * @throws Exception on any failure.
    */
   @Override
-  public int run(String[] args) throws Exception {
+  public int run(final String[] args, final PrintStream stream)
+      throws Exception, ExitUtil.ExitException {
+
+    this.out = new PrintWriter(stream);
+
     preConditionArgsSizeCheck(args);
     List<String> paths = Arrays.asList(args);
 
@@ -120,17 +131,7 @@ public int run(String[] args) throws Exception {
     Path destPath = new Path(paths.get(1));
 
     // Setting the file system
-    URI fsURI = new URI(logsPath.toString());
-    FileSystem fileSystem = FileSystem.get(fsURI, new Configuration());
-
-    FileStatus logsFileStatus = fileSystem.getFileStatus(logsPath);
-    if (logsFileStatus.isFile()) {
-      errorln("Expecting a directory, but " + logsPath.getName() + " is a"
-          + " file which was passed as an argument");
-      throw invalidArgs(
-          "Expecting a directory, but " + logsPath.getName() + " is a"
-              + " file which was passed as an argument");
-    }
+    FileSystem fileSystem = logsPath.getFileSystem(getConf());
 
     // Calls S3AAuditLogMergerAndParser for implementing merging, passing of
     // audit log files and converting into avro file
@@ -151,40 +152,9 @@ private void preConditionArgsSizeCheck(String[] args) {
     }
   }
 
-  protected static void errorln(String x) {
-    System.err.println(x);
-  }
-
-  /**
-   * Build the exception to raise on invalid arguments.
-   *
-   * @param format string format
-   * @param args   optional arguments for the string
-   * @return a new exception to throw
-   */
-  protected static ExitUtil.ExitException invalidArgs(
-      String format, Object... args) {
-    return exitException(INVALID_ARGUMENT, format, args);
-  }
-
-  /**
-   * Build a exception to throw with a formatted message.
-   *
-   * @param exitCode exit code to use
-   * @param format   string format
-   * @param args     optional arguments for the string
-   * @return a new exception to throw
-   */
-  protected static ExitUtil.ExitException exitException(
-      final int exitCode,
-      final String format,
-      final Object... args) {
-    return new ExitUtil.ExitException(exitCode,
-        String.format(format, args));
-  }
 
   /**
-   * Flush all active output channels, including {@Code System.err},
+   * Flush all active output channels, including {@code System.err},
    * so as to stay in sync with any JRE log messages.
    */
   private void flush() {
@@ -196,44 +166,12 @@ private void flush() {
     System.err.flush();
   }
 
-  @Override
-  public void close() throws IOException {
+
+  public void closeOutput() throws IOException {
     flush();
     if (out != null) {
       out.close();
     }
   }
 
-  /**
-   * Inner entry point, with no logging or system exits.
-   *
-   * @param conf configuration
-   * @param argv argument list
-   * @return an exception
-   * @throws Exception Exception.
-   */
-  public static int exec(Configuration conf, String... argv) throws Exception {
-    try (AuditTool auditTool = new AuditTool()) {
-      return ToolRunner.run(conf, auditTool, argv);
-    }
-  }
-
-  /**
-   * Main entry point.
-   *
-   * @param argv args list
-   */
-  public static void main(String[] argv) {
-    try {
-      ExitUtil.terminate(exec(new Configuration(), argv));
-    } catch (ExitUtil.ExitException e) {
-      LOG.error("Exception while Terminating the command ran :{}",
-          e.toString());
-      System.exit(e.status);
-    } catch (Exception e) {
-      LOG.error("Exception while Terminating the command ran :{}",
-          e.toString(), e);
-      ExitUtil.halt(-1, e);
-    }
-  }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/audit/mapreduce/S3AAuditLogMergerAndParser.java

@@ -43,7 +43,10 @@
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.LineRecordReader;
+import org.apache.hadoop.util.DurationInfo;
 
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE;
 import static org.apache.hadoop.fs.s3a.audit.S3LogParser.AWS_LOG_REGEXP_GROUPS;
 import static org.apache.hadoop.fs.s3a.audit.S3LogParser.BYTESSENT_GROUP;
 import static org.apache.hadoop.fs.s3a.audit.S3LogParser.LOG_ENTRY_PATTERN;
@@ -65,9 +68,30 @@ public class S3AAuditLogMergerAndParser {
   private static final String REFERRER_HEADER_KEY = "referrer";
 
   // Basic parsing counters.
+
+  /**
+   * Number of audit log files parsed.
+   */
+  private long logFilesParsed = 0;
+
+  /**
+   * Number of log entries parsed.
+   */
   private long auditLogsParsed = 0;
+
+
+
+
+  /**
+   * How many referrer headers were parsed.
+   */
   private long referrerHeaderLogParsed = 0;
 
+  /**
+   * How many records were skipped due to lack of referrer or other reason.
+   */
+  private long recordsSkipped = 0;
+
   /**
    * parseAuditLog method helps in parsing the audit log
    * into key-value pairs using regular expressions.
@@ -78,7 +102,7 @@ public class S3AAuditLogMergerAndParser {
   public HashMap<String, String> parseAuditLog(String singleAuditLog) {
     HashMap<String, String> auditLogMap = new HashMap<>();
     if (singleAuditLog == null || singleAuditLog.isEmpty()) {
-      LOG.info("This is an empty string or null string, expected a valid string to parse");
+      LOG.debug("This is an empty string or null string, expected a valid string to parse");
       return auditLogMap;
     }
     final Matcher matcher = LOG_ENTRY_PATTERN.matcher(singleAuditLog);
@@ -107,11 +131,10 @@ public HashMap<String, String> parseAuditLog(String singleAuditLog) {
    *                       audit log.
    * @return returns a map which contains key-value pairs of referrer headers
    */
-  public HashMap<String, String> parseReferrerHeader(String referrerHeader) {
+  public static HashMap<String, String> parseReferrerHeader(String referrerHeader) {
     HashMap<String, String> referrerHeaderMap = new HashMap<>();
     if (referrerHeader == null || referrerHeader.isEmpty()) {
-      LOG.info(
-          "This is an empty string or null string, expected a valid string to parse");
+      LOG.debug("This is an empty string or null string, expected a valid string to parse");
       return referrerHeaderMap;
     }
 
@@ -163,23 +186,26 @@ public boolean mergeAndParseAuditLogFiles(FileSystem fileSystem,
       Path logsPath,
       Path destPath) throws IOException {
 
-    // Listing file in given path
+    // List source log files
     RemoteIterator<LocatedFileStatus> listOfLogFiles =
         fileSystem.listFiles(logsPath, true);
 
-    Path destFile = destPath;
+    Path destFile = new Path(destPath, "AuditLogFile.avro");
 
     try (FSDataOutputStream fsDataOutputStream = fileSystem.create(destFile)) {
 
       // Iterating over the list of files to merge and parse
       while (listOfLogFiles.hasNext()) {
+        logFilesParsed++;
         FileStatus fileStatus = listOfLogFiles.next();
         int fileLength = (int) fileStatus.getLen();
         byte[] byteBuffer = new byte[fileLength];
 
-        try (FSDataInputStream fsDataInputStream =
+        try (DurationInfo duration = new DurationInfo(LOG, "Processing %s", fileStatus.getPath());
+            FSDataInputStream fsDataInputStream =
             awaitFuture(fileSystem.openFile(fileStatus.getPath())
                 .withFileStatus(fileStatus)
+                .opt(FS_OPTION_OPENFILE_READ_POLICY, FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE)
                 .build())) {
 
           // Instantiating generated AvroDataRecord class
@@ -226,6 +252,7 @@ public boolean mergeAndParseAuditLogFiles(FileSystem fileSystem,
               if (StringUtils.isBlank(referrerHeader) || referrerHeader
                   .equals("-")) {
                 LOG.debug("Invalid referrer header for this audit log...");
+                recordsSkipped++;
                 continue;
               }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardTool.java

@@ -51,6 +51,7 @@
 import org.apache.hadoop.fs.s3a.Constants;
 import org.apache.hadoop.fs.s3a.S3AFileSystem;
 import org.apache.hadoop.fs.s3a.WriteOperationHelper;
+import org.apache.hadoop.fs.s3a.audit.AuditTool;
 import org.apache.hadoop.fs.s3a.auth.RolePolicies;
 import org.apache.hadoop.fs.s3a.auth.delegation.S3ADelegationTokens;
 import org.apache.hadoop.fs.s3a.commit.CommitConstants;
@@ -992,6 +993,9 @@ public static int run(Configuration conf, String... args) throws
     }
     switch (subCommand) {
 
+    case AuditTool.AUDIT:
+      command = new AuditTool(conf);
+      break;
     case BucketInfo.NAME:
       command = new BucketInfo(conf);
       break;

hadoop-tools/hadoop-aws/src/main/shellprofile.d/hadoop-s3a.sh

@@ -0,0 +1,37 @@
+#!/usr/bin/env bash
+
+# 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.
+
+if ! declare -f hadoop_subcommand_s3a >/dev/null 2>/dev/null; then
+
+  if [[ "${HADOOP_SHELL_EXECNAME}" = hadoop ]]; then
+    hadoop_add_subcommand "s3a" client "S3 Commands"
+  fi
+
+  # this can't be indented otherwise shelldocs won't get it
+
+## @description  s3a command for hadoop
+## @audience     public
+## @stability    stable
+## @replaceable  yes
+function hadoop_subcommand_s3a
+{
+  # shellcheck disable=SC2034
+  HADOOP_CLASSNAME=org.apache.hadoop.fs.s3a.s3guard.S3GuardTool
+  hadoop_add_to_classpath_tools hadoop-aws
+}
+
+fi