MAPREDUCE-7474. Task save failure stats in job stats + more

Statistics Collection and Printing
* New statistic task_stage_save_summary_file to distinguish from
  other saving operations (job success/report file)
* After a failure to save a task attempt, the iostats of the manifest
  are rebuilt so the stats on failures are updated.
  This will get into the final job _SUCCESS statistics so we can see
  if anything happened
* Make the manifest print command something which can be invoked from
  the commandline: mapred successfile
  This is covered in the docs.

The failure stats regeneration is nice; works by passing down a
lambda-expression of the logic to (re)generate the manifest, and invoking
this on every attempt. As this is where the stats are aggregated,
it includes details on the previous failing attempts.

Directory size for deletion
* Optionally pass down directory count under job dir to cleanup stage
* This is determined in job commit from aggregate statistics;
  unknown elsewhere (app abort etc.).
* It is currently only logged; it may be possible to support an option
  of when to skip the initial serial delete, though it will depend on
  abfs login mechanism.

Testing
* More fault injection scenarios.
* Ability to assert that iostats do not contain specific non-zero stats.
  This is used in ITestAbfsTerasort to assert no task save or rename failures.
  The stats before this change imply this did happen in a job commit; no
  other details, hence the new probe.
* Log manifest committer at debug in mapred-core

Note: if there's a retry process which means the operation can take minutes,
the initial operation will block progress() callbacks so mapreduce jobs
will fail. Spark is unaffected

Change-Id: Id423267de89c7f31e4b1283f9c433b729ff0d87b

Author: steveloughran

hadoop-mapreduce-project/bin/mapred

@@ -37,6 +37,7 @@ function hadoop_usage
   hadoop_add_subcommand "frameworkuploader" admin "mapreduce framework upload"
   hadoop_add_subcommand "version" client "print the version"
   hadoop_add_subcommand "minicluster" client "CLI MiniCluster"
+  hadoop_add_subcommand "successfile" client "Print a _SUCCESS manifest from the manifest and S3A committers"
   hadoop_generate_usage "${HADOOP_SHELL_EXECNAME}" true
 }
 
@@ -102,6 +103,9 @@ function mapredcmd_case
     version)
       HADOOP_CLASSNAME=org.apache.hadoop.util.VersionInfo
     ;;
+    successfile)
+      HADOOP_CLASSNAME=org.apache.hadoop.mapreduce.lib.output.committer.manifest.files.ManifestPrinter
+    ;;
     minicluster)
       hadoop_add_classpath "${HADOOP_YARN_HOME}/${YARN_DIR}/timelineservice"'/*'
       hadoop_add_classpath "${HADOOP_YARN_HOME}/${YARN_DIR}/test"'/*'

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/ManifestCommitterStatisticNames.java

@@ -187,6 +187,12 @@ public final class ManifestCommitterStatisticNames {
   public static final String OP_SAVE_TASK_MANIFEST =
       "task_stage_save_task_manifest";
 
+  /**
+   * Save a summary file: {@value}.
+   */
+  public static final String OP_SAVE_SUMMARY_FILE =
+      "task_stage_save_summary_file";
+
   /**
    * Task abort: {@value}.
    */

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/files/ManifestPrinter.java

@@ -36,7 +36,7 @@
  */
 public class ManifestPrinter extends Configured implements Tool {
 
-  private static final String USAGE = "ManifestPrinter <success-file>";
+  private static final String USAGE = "successfile <success-file>";
 
   /**
    * Output for printing.
@@ -88,7 +88,7 @@ public ManifestSuccessData loadAndPrintManifest(FileSystem fs, Path path)
     return success;
   }
 
-  private void printManifest(ManifestSuccessData success) {
+  public void printManifest(ManifestSuccessData success) {
     field("succeeded", success.getSuccess());
     field("created", success.getDate());
     field("committer", success.getCommitter());

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/impl/InternalConstants.java

@@ -86,6 +86,7 @@ private InternalConstants() {
       OP_MSYNC,
       OP_PREPARE_DIR_ANCESTORS,
       OP_RENAME_FILE,
+      OP_SAVE_SUMMARY_FILE,
       OP_SAVE_TASK_MANIFEST,
 
       OBJECT_LIST_REQUEST,

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/stages/AbstractJobOrTaskStage.java

@@ -23,6 +23,8 @@
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -62,6 +64,7 @@
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_DELETE_DIR;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_LOAD_MANIFEST;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_MSYNC;
+import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_RENAME_DIR;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_RENAME_FILE;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_SAVE_TASK_MANIFEST;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.impl.AuditingIntegration.enterStageWorker;
@@ -371,6 +374,7 @@ public final IOStatisticsStore getIOStatistics() {
    */
   protected final void progress() {
     if (stageConfig.getProgressable() != null) {
+      LOG.trace("{}: Progressing", getName());
       stageConfig.getProgressable().progress();
     }
   }
@@ -429,7 +433,7 @@ protected final boolean isFile(
    * @return status or null
    * @throws IOException IO Failure.
    */
-  protected final boolean delete(
+  public final boolean delete(
       final Path path,
       final boolean recursive)
       throws IOException {
@@ -445,7 +449,7 @@ protected final boolean delete(
    * @return status or null
    * @throws IOException IO Failure.
    */
-  protected Boolean delete(
+  public Boolean delete(
       final Path path,
       final boolean recursive,
       final String statistic)
@@ -467,7 +471,7 @@ protected Boolean delete(
    * @return outcome.
    * @throws IOException IO Failure.
    */
-  protected boolean deleteFile(
+  public boolean deleteFile(
       final Path path,
       final String statistic)
       throws IOException {
@@ -482,7 +486,7 @@ protected boolean deleteFile(
    * @return true if the directory was created/exists.
    * @throws IOException IO Failure.
    */
-  protected final boolean mkdirs(
+  public final boolean mkdirs(
       final Path path,
       final boolean escalateFailure)
       throws IOException {
@@ -519,7 +523,7 @@ protected final RemoteIterator<FileStatus> listStatusIterator(
    * @return the manifest.
    * @throws IOException IO Failure.
    */
-  protected final TaskManifest loadManifest(
+  public final TaskManifest loadManifest(
       final FileStatus status)
       throws IOException {
     LOG.trace("{}: loadManifest('{}')", getName(), status);
@@ -612,45 +616,84 @@ protected final Path directoryMustExist(
    * @param manifestData the manifest/success file
    * @param tempPath temp path for the initial save
    * @param finalPath final path for rename.
+   * @return the manifest saved.
    * @throws IOException failure to rename after retries.
    */
   @SuppressWarnings("unchecked")
-  protected final <T extends AbstractManifestData> void save(T manifestData,
+  protected final <T extends AbstractManifestData> T save(
+      final T manifestData,
       final Path tempPath,
       final Path finalPath) throws IOException {
+    return saveManifest(() -> manifestData, tempPath, finalPath, OP_SAVE_TASK_MANIFEST);
+  }
 
-    int retryCount = 0;
+  /**
+   * Generate and save a task manifest or summary file.
+   * This is be done by writing to a temp path and then renaming.
+   * <p>
+   * If the destination path exists: Delete it before the rename.
+   * <p>
+   * This will retry so that a rename failure from abfs load or IO errors
+   * such as delete or save failure will not fail the task.
+   * <p>
+   * The {@code manifestSource} supplier is invoked to get the manifest data
+   * on every attempt.
+   * This permits statistics to be updated, <i>including those of failures</i>.
+   * @param manifestSource supplier the manifest/success file
+   * @param tempPath temp path for the initial save
+   * @param finalPath final path for rename.
+   * @param statistic statistic to use for timing
+   * @return the manifest saved.
+   * @throws IOException failure to save/delete/rename after retries.
+   */
+  @SuppressWarnings("unchecked")
+  protected final <T extends AbstractManifestData> T saveManifest(
+      final Supplier<T> manifestSource,
+      final Path tempPath,
+      final Path finalPath,
+      String statistic) throws IOException {
+
+    AtomicInteger retryCount = new AtomicInteger(0);
     RetryPolicy retryPolicy = retryUpToMaximumCountWithProportionalSleep(
         getStageConfig().getManifestSaveAttempts(),
         SAVE_SLEEP_INTERVAL,
         TimeUnit.MILLISECONDS);
-    boolean success = false;
-    while (!success) {
-      try {
-        LOG.info("{}: save manifest to {} then rename as {}'); retry count={}",
-            getName(), tempPath, finalPath, retryCount);
-
-        // save the temp file, overwriting any which remains from an earlier attempt
-        trackDurationOfInvocation(getIOStatistics(), OP_SAVE_TASK_MANIFEST, () ->
-            operations.save(manifestData, tempPath, true));
-
-        // delete the destination in case it exists either from a failed previous
-        // attempt or from a concurrent task commit.
-        deleteFile(finalPath, OP_DELETE);
 
-        // rename temp to final
-        renameFile(tempPath, finalPath);
-        // success flag is only set after the rename.
-        success = true;
+    // loop until returning a value or raising an exception
+    while (true) {
+      try {
+        T manifestData = requireNonNull(manifestSource.get());
+        trackDurationOfInvocation(getIOStatistics(), statistic, () -> {
+          LOG.info("{}: save manifest to {} then rename as {}'); retry count={}",
+              getName(), tempPath, finalPath, retryCount);
+
+          // delete temp path.
+          // even though this is written with overwrite=true, this extra recursive
+          // delete also handles a directory being there.
+          deleteRecursive(tempPath, OP_DELETE);
+
+          // save the temp file, overwriting any which remains from an earlier attempt
+          operations.save(manifestData, tempPath, true);
+
+          // delete the destination in case it exists either from a failed previous
+          // attempt or from a concurrent task commit.
+          delete(finalPath, true, OP_DELETE);
+
+          // rename temp to final
+          renameFile(tempPath, finalPath);
+        });
+        // success: exit and return the final manifest data.
+        return manifestData;
       } catch (IOException e) {
         // failure.
         // log then decide whether to sleep and retry or give up.
         LOG.warn("{}: Failed to save and commit file {} renamed to {}; retry count={}",
             getName(), tempPath, finalPath, retryCount, e);
-        retryCount++;
+        // increment that count.
+        retryCount.incrementAndGet();
         RetryPolicy.RetryAction retryAction;
         try {
-          retryAction = retryPolicy.shouldRetry(e, retryCount, 0, true);
+          retryAction = retryPolicy.shouldRetry(e, retryCount.get(), 0, true);
         } catch (Exception ex) {
           // it's not clear why this probe can raise an exception; it is just
           // caught and mapped to a fail.
@@ -716,7 +759,7 @@ protected final void renameDir(final Path source, final Path dest)
 
     maybeDeleteDest(true, dest);
     executeRenamingOperation("renameDir", source, dest,
-        OP_RENAME_FILE, () ->
+        OP_RENAME_DIR, () ->
         operations.renameDir(source, dest)
     );
   }
@@ -1019,8 +1062,7 @@ protected boolean deleteRecursive(
    */
   protected IOException deleteRecursiveSuppressingExceptions(
       final Path dir,
-      final String statistic)
-      throws IOException {
+      final String statistic) {
     try {
       deleteRecursive(dir, statistic);
       return null;

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/stages/CleanupJobStage.java

@@ -160,10 +160,18 @@ protected Result executeStage(
 
     // to delete.
     LOG.info("{}: Deleting job directory {}", getName(), baseDir);
+    final long directoryCount = args.directoryCount;
+    if (directoryCount > 0) {
+      // log the expected directory count, which drives duration in GCS
+      // and may cause timeouts on azure if the count is too high for a
+      // timely permissions tree scan.
+      LOG.info("{}: Expected directory count: {}", getName(), directoryCount);
+    }
 
+    progress();
+    // check and maybe execute parallel delete of task attempt dirs.
     if (args.deleteTaskAttemptDirsInParallel) {
 
-      // parallel delete of task attempt dirs.
 
       if (args.parallelDeleteAttemptBaseDeleteFirst) {
         // attempt to delete the base dir first.
@@ -180,9 +188,9 @@ protected Result executeStage(
             baseDirDeleted = true;
           } else {
             // failure: log and continue
-            LOG.warn("{}: Exception on initial attempt at deleting base dir {}\n"
-                    + "attempting parallel delete",
-                getName(), baseDir, exception);
+            LOG.warn("{}: Exception on initial attempt at deleting base dir {} and directory count {}"
+                    + "\nFalling back to parallel delete",
+                getName(), baseDir, directoryCount, exception);
           }
         }
       }
@@ -339,27 +347,37 @@ public static final class Arguments {
     /** Ignore failures? */
     private final boolean suppressExceptions;
 
+    /**
+     * Non-final count of directories.
+     * Default value, "0", means "unknown".
+     * This can be dynamically updated during job commit.
+     */
+    private long directoryCount;
+
     /**
      * Arguments to the stage.
      * @param statisticName stage name to report
      * @param enabled is the stage enabled?
      * @param deleteTaskAttemptDirsInParallel delete task attempt dirs in
      * parallel?
      * @param parallelDeleteAttemptBaseDeleteFirst Make an initial attempt to
-     *         delete the base directory in a parallel delete?
+     * delete the base directory in a parallel delete?
      * @param suppressExceptions suppress exceptions?
+     * @param directoryCount directories under job dir; 0 means unknown.
      */
     public Arguments(
         final String statisticName,
         final boolean enabled,
         final boolean deleteTaskAttemptDirsInParallel,
         final boolean parallelDeleteAttemptBaseDeleteFirst,
-        final boolean suppressExceptions) {
+        final boolean suppressExceptions,
+        long directoryCount) {
       this.statisticName = statisticName;
       this.enabled = enabled;
       this.deleteTaskAttemptDirsInParallel = deleteTaskAttemptDirsInParallel;
       this.suppressExceptions = suppressExceptions;
       this.parallelDeleteAttemptBaseDeleteFirst = parallelDeleteAttemptBaseDeleteFirst;
+      this.directoryCount = directoryCount;
     }
 
     public String getStatisticName() {
@@ -382,6 +400,14 @@ public boolean isParallelDeleteAttemptBaseDeleteFirst() {
       return parallelDeleteAttemptBaseDeleteFirst;
     }
 
+    public long getDirectoryCount() {
+      return directoryCount;
+    }
+
+    public void setDirectoryCount(final long directoryCount) {
+      this.directoryCount = directoryCount;
+    }
+
     @Override
     public String toString() {
       return "Arguments{" +
@@ -401,8 +427,9 @@ public String toString() {
   public static final Arguments DISABLED = new Arguments(OP_STAGE_JOB_CLEANUP,
       false,
       false,
-      false, false
-  );
+      false,
+      false,
+      0);
 
   /**
    * Build an options argument from a configuration, using the
@@ -430,7 +457,8 @@ public static Arguments cleanupStageOptionsFromConfig(
         enabled,
         deleteTaskAttemptDirsInParallel,
         parallelDeleteAttemptBaseDeleteFirst,
-        suppressExceptions);
+        suppressExceptions,
+        0);
   }
 
   /**

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/stages/CommitJobStage.java

@@ -37,6 +37,7 @@
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.COMMITTER_BYTES_COMMITTED_COUNT;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.COMMITTER_FILES_COMMITTED_COUNT;
+import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.COMMITTER_TASK_DIRECTORY_COUNT_MEAN;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_STAGE_JOB_COMMIT;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_STAGE_JOB_CREATE_TARGET_DIRS;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_STAGE_JOB_LOAD_MANIFESTS;
@@ -161,7 +162,12 @@ protected CommitJobStage.Result executeStage(
       }
 
       // optional cleanup
-      new CleanupJobStage(stageConfig).apply(arguments.getCleanupArguments());
+      final CleanupJobStage.Arguments cleanupArguments = arguments.getCleanupArguments();
+      // determine the directory count
+      cleanupArguments.setDirectoryCount(iostats.counters()
+          .getOrDefault(COMMITTER_TASK_DIRECTORY_COUNT_MEAN, 0L));
+
+      new CleanupJobStage(stageConfig).apply(cleanupArguments);
 
       // and then, after everything else: optionally validate.
       if (arguments.isValidateOutput()) {