[b/356461225] Add progress during HDFS dumping (#673)

* [b/356461225] Add progress during HDFS dumping

* Fix HDFS extraction stats, simplify ScanContext & SingleDirScanJob

* Separate PROGRESS logging vs DEBUG logging (see ScanContext.PROGRESS_DEBUG_STATS)

* Address review comments

Author: sten1ee

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/ConnectorArguments.java

@@ -141,11 +141,14 @@ public class ConnectorArguments extends DefaultArguments {
   public static final String OPT_HADOOP_RPC_PROTECTION = "hadoop-rpc-protection";
 
   public static final String OPT_HDFS_PRINCIPAL_PREFIX = "hdfs-principal-prefix";
+
+  public static final String OPT_HDFS_SCAN_ROOT_PATH = "hdfs-scan-root-path";
   public static final String OPT_HADOOP_CORE_SITE_XML_DEFAULT =
       "/etc/hadoop/conf.cloudera.hdfs/core-site.xml";
   public static final String OPT_HADOOP_HDFS_SITE_XML_DEFAULT =
       "/etc/hadoop/conf.cloudera.hdfs/hdfs-site.xml";
   public static final String OPT_HDFS_PRINCIPAL_PREFIX_DEFAULT = "hdfs/_HOST@";
+  public static final String OPT_HDFS_SCAN_ROOT_PATH_DEFAULT = "/";
   // Ranger.
   public static final String OPT_RANGER_PORT_DEFAULT = "6080";
   public static final String OPT_RANGER_PAGE_SIZE = "ranger-page-size";
@@ -484,6 +487,13 @@ public class ConnectorArguments extends DefaultArguments {
           .ofType(String.class)
           .defaultsTo(OPT_HDFS_PRINCIPAL_PREFIX_DEFAULT);
 
+  private final OptionSpec<String> optionHdfsScanRootPath =
+      parser
+          .accepts(OPT_HDFS_SCAN_ROOT_PATH, "HDFS root path to be scanned recursively.")
+          .withRequiredArg()
+          .ofType(String.class)
+          .defaultsTo(OPT_HDFS_SCAN_ROOT_PATH_DEFAULT);
+
   public final OptionSpec<Void> optionKerberosAuthForHadoop =
       parser
           .accepts(OPT_KERBEROS_AUTH_FOR_HADOOP, "Use Kerberos auth for Hadoop.")
@@ -1004,6 +1014,10 @@ public String getHdfsPrincipalPrefix() {
     return getOptions().valueOf(optionHdfsPrincipalPrefix);
   }
 
+  public String getHdfsScanRootPath() {
+    return getOptions().valueOf(optionHdfsScanRootPath);
+  }
+
   @CheckForNull
   public String getGenericQuery() {
     return getOptions().valueOf(optionGenericQuery);

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/TasksRunner.java

@@ -45,7 +45,7 @@
 public class TasksRunner implements TaskRunContextOps {
 
   private static final Logger LOG = LoggerFactory.getLogger(TasksRunner.class);
-  private static final Logger PROGRESS_LOG = LoggerFactory.getLogger("progress-logger");
+  public static final Logger PROGRESS_LOG = LoggerFactory.getLogger("progress-logger");
 
   private AtomicInteger numberOfCompletedTasks;
   private final int totalNumberOfTasks;

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/hdfs/HdfsExtractionTask.java

@@ -16,6 +16,9 @@
  */
 package com.google.edwmigration.dumper.application.dumper.connector.hdfs;
 
+import static com.google.edwmigration.dumper.application.dumper.ConnectorArguments.OPT_HDFS_SCAN_ROOT_PATH;
+import static com.google.edwmigration.dumper.application.dumper.ConnectorArguments.OPT_THREAD_POOL_SIZE;
+import static com.google.edwmigration.dumper.application.dumper.connector.hdfs.SingleDirScanJob.trimExceptionMessage;
 import static java.lang.String.format;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
@@ -35,6 +38,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nonnull;
+import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DistributedFileSystem;
@@ -44,12 +48,17 @@
 public class HdfsExtractionTask extends AbstractTask<Void> implements HdfsExtractionDumpFormat {
   private static final Logger LOG = LoggerFactory.getLogger(HdfsExtractionTask.class);
 
-  private final int poolSize;
+  private final int threadPoolSize;
+  private final String hdfsScanRootPath;
 
   HdfsExtractionTask(@Nonnull ConnectorArguments args) {
     super(HdfsFormat.ZIP_ENTRY_NAME);
-    Preconditions.checkNotNull(args, "Arguments was null.");
-    poolSize = args.getThreadPoolSize();
+    threadPoolSize = args.getThreadPoolSize();
+    Preconditions.checkArgument(
+        threadPoolSize > 0, "Argument %s should be positive number", OPT_THREAD_POOL_SIZE);
+    hdfsScanRootPath = args.getHdfsScanRootPath();
+    Preconditions.checkArgument(
+        !hdfsScanRootPath.isEmpty(), "Argument %s should be non-empty", OPT_HDFS_SCAN_ROOT_PATH);
   }
 
   @Override
@@ -74,21 +83,38 @@ protected Void doRun(TaskRunContext context, @Nonnull ByteSink sink, @Nonnull Ha
     DistributedFileSystem fs = ((HdfsHandle) handle).getDfs();
     // Create a dedicated ExecutorService to use:
     ExecutorService execService =
-        ExecutorManager.newExecutorServiceWithBackpressure("hdfs-extraction", poolSize);
+        ExecutorManager.newExecutorServiceWithBackpressure("hdfs-extraction", threadPoolSize);
     try (Writer output = sink.asCharSink(UTF_8).openBufferedStream();
-        ScanContext scanCtx = new ScanContext(fs, output);
-        ExecutorManager execManager = new ExecutorManager(execService)) {
+        ExecutorManager execManager = new ExecutorManager(execService);
+        ScanContext scanCtx = new ScanContext(execManager, fs, output)) {
 
-      String hdfsPath = "/";
-      FileStatus rootDir = fs.getFileStatus(new Path(hdfsPath));
-      SingleDirScanJob rootJob = new SingleDirScanJob(scanCtx, execManager, rootDir);
-      execManager.execute(rootJob); // The root job executes immediately
+      LOG.info(
+          "Running HDFS extraction\n\t{}: {}\n\t{}: {}",
+          OPT_HDFS_SCAN_ROOT_PATH,
+          hdfsScanRootPath,
+          OPT_THREAD_POOL_SIZE,
+          threadPoolSize);
+      FileStatus rootDir = fs.getFileStatus(new Path(hdfsScanRootPath));
+      scanCtx.submitRootDirScanJob(rootDir, getContentSummaryFor(fs, rootDir));
       execManager.await(); // Wait until all (recursive) tasks are done executing
-      LOG.info(scanCtx.getFormattedStats());
+      LOG.info("Final stats:\n{}", scanCtx.getDetailedStats());
     } finally {
       // Shutdown the dedicated ExecutorService:
       MoreExecutors.shutdownAndAwaitTermination(execService, 100, TimeUnit.MILLISECONDS);
     }
     return null;
   }
+
+  private ContentSummary getContentSummaryFor(DistributedFileSystem dfs, FileStatus file) {
+    try {
+      return dfs.getContentSummary(file.getPath());
+    } catch (org.apache.hadoop.security.AccessControlException exn) {
+      LOG.info(
+          "Progress for HDFS extraction won't be displayed due to AccessControlException: {}",
+          trimExceptionMessage(exn.getMessage()));
+    } catch (IOException exn) {
+      LOG.error("Progress for HDFS extraction won't be displayed due to IOException: ", exn);
+    }
+    return null;
+  }
 }