Major Cloudup rework

* incremental -update operation to skip files which exist.
* improved output
* options to flush/hflush
* standalone document
* replaced "upload" with "copy"

Author: steveloughran

README.md

@@ -171,52 +171,8 @@ com.amazonaws.services.s3.model.AmazonS3Exception: The specified method is not a
 
 ## Command `cloudup` -upload and download files; optimised for cloud storage 
 
-Bulk download of everything from s3a://bucket/qelogs/ to the local dir localquelogs (assuming the default fs is file://)
+See [cloudup](src/main/site/cloudup.md)
 
-Usage
-
-```
-cloudup -s source -d dest [-o] [-i] [-l <largest>] [-t threads] 
-
--s <uri> : source
--d <uri> : dest
--o: overwrite
--i: ignore failures
--t <n> : number of threads
--l <n> : number of "largest" files to start uploading before just randomly picking files.
-
-```
-
-Algorithm
-
-1. source files are listed.
-1. A pool of worker threads is created
-2. the largest N files are queued for upload first, where N is a default or the value set by `-l`.
-1. The remainder of the files are randomized to avoid throttling and then queued
-1. the program waits for everything to complete.
-1. Source and dest FS stats are printed.
-
-This is not `distcp` run across a cluster; it's a single process with some threads. 
-Works best for reading lots of small files from an object store or when you have a 
-mix of large and small files to download or uplaod.
-
-
-
-```
-bin/hadoop jar cloudstore-1.0.jar cloudup \
- -s s3a://bucket/qelogs/ \
- -d localqelogs \
- -t 32 -o
-```
-
-and the other way
-
-```
-bin/hadoop jar cloudstore-1.0.jar cloudup \
- -d localqelogs \
- -s s3a://bucket/qelogs/ \
- -t 32 -o  -l 4
-```
 
 ## Command `committerinfo`
 

pom.xml

@@ -365,6 +365,18 @@
       </properties>
     </profile>
 
+    <profile>
+      <id>cdh7.1</id>
+      <activation>
+        <property>
+          <name>cdh7.1</name>
+        </property>
+      </activation>
+      <properties>
+        <hadoop.version>3.1.1.7.1.7.2000-1</hadoop.version>
+      </properties>
+    </profile>
+
     <profile>
       <id>extra</id>
       <activation>

src/main/java/org/apache/hadoop/fs/store/CommonParameters.java

@@ -54,6 +54,22 @@ public final class CommonParameters {
 
   public static final String BFS = "bfs";
 
+  public static final String BLOCK = "block";
+
+  public static final String CSVFILE = "csv";
+
+  public static final String FLUSH = "flush";
+
+  public static final String HFLUSH = "hflush";
+
+  public static final String IGNORE = "ignore";
+
+  public static final String LARGEST = "largest";
+
+  public static final String OVERWRITE = "overwrite";
+
+  public static final String UPDATE = "update";
+
   private CommonParameters() {
   }
 }

src/main/java/org/apache/hadoop/fs/store/StoreDurationInfo.java

@@ -20,6 +20,7 @@
 
 import java.io.PrintStream;
 import java.time.Duration;
+import java.util.function.Supplier;
 
 import org.slf4j.Logger;
 
@@ -40,10 +41,20 @@ public class StoreDurationInfo
 
   private boolean isFinished;
 
-  private final String text;
+
+  private final Supplier<String> text;
+
+  public static final Supplier<String> EMPTY_TEXT = () -> "";
+
+  private String textStr;
 
   private final Logger log;
 
+  /**
+   * Should the log be at INFO rather than DEBUG.
+   */
+  private final boolean logAtInfo;
+
   private final PrintStream out;
 
   /**
@@ -53,16 +64,40 @@ public class StoreDurationInfo
    * @param args list of arguments
    */
   public StoreDurationInfo(Logger log, String format, Object... args) {
-    started = time();
-    finished = started;
-    this.text = String.format(format, args);
+
+    this(log, true, format, args);
+  }
+
+  /**
+   * Create the duration text from a {@code String.format()} code call
+   * and log either at info or debug.
+   * @param log log to write to
+   * @param logAtInfo should the log be at info, rather than debug
+   * @param format format string
+   * @param args list of arguments
+   */
+  public StoreDurationInfo(Logger log,
+      boolean logAtInfo,
+      String format,
+      Object... args) {
+    this.started = time();
+    this.finished = started;
+    this.text = () -> String.format(format, args);
     this.log = log;
-    out = null;
+    this.logAtInfo = logAtInfo;
+    this.out = null;
     if (log != null) {
-      log.info("Starting: {}", text);
+      if (logAtInfo) {
+        log.info("Starting: {}", getFormattedText());
+      } else {
+        if (log.isDebugEnabled()) {
+          log.debug("Starting: {}", getFormattedText());
+        }
+      }
     }
   }
 
+
   /**
    * Create the duration text from a {@code String.format()} code call.
    * @param out log to write to
@@ -72,25 +107,33 @@ public StoreDurationInfo(Logger log, String format, Object... args) {
   public StoreDurationInfo(PrintStream out, String format, Object... args) {
     started = time();
     finished = started;
-    this.text = String.format(format, args);
+    this.text = () -> String.format(format, args);
     this.out = out;
     this.log = null;
+    this.logAtInfo = false;
+
     if (out != null) {
-      out.printf("Starting: %s%n", text);
+      out.printf("Starting: %s%n", getFormattedText());
     }
   }
 
   /**
    * Create the duration with no output printed.
    */
   public StoreDurationInfo() {
-    this.text = "";
+    this.text = EMPTY_TEXT;
     this.log = null;
     this.out = null;
+    this.logAtInfo = false;
     started = time();
     finished = started;
   }
 
+
+  private String getFormattedText() {
+    return (textStr == null) ? (textStr = text.get()) : textStr;
+  }
+
   private long time() {
     return System.currentTimeMillis();
   }
@@ -112,7 +155,8 @@ public String getDurationString() {
   public static String humanTime(long time) {
     long seconds = (time / 1000);
     long minutes = (seconds / 60);
-    return String.format("%d:%02d.%03d", minutes, seconds % 60, time % 1000);
+    long hours = (minutes / 60);
+    return String.format("%d:%02d:%02d.%03d", hours, minutes % 60, seconds % 60, time % 1000);
   }
 
   /**
@@ -145,10 +189,16 @@ public String toString() {
   public void close() {
     finished();
     if (log != null) {
-      log.info("Duration of {}: {}", text, this);
+      if (logAtInfo) {
+        log.info("Duration of {}: {}", getFormattedText(), this);
+        ;
+      } else {
+        log.debug("Duration of {}: {}", getFormattedText(), this);
+      }
+
     }
     if (out != null) {
-      out.printf("Duration of %s: %s%n", text, this);
+      out.printf("Duration of %s: %s%n", getFormattedText(), this);
     }
   }
 }

src/main/java/org/apache/hadoop/fs/store/StoreEntryPoint.java

@@ -66,6 +66,11 @@
 import static org.apache.hadoop.fs.store.CommonParameters.XMLFILE;
 import static org.apache.hadoop.fs.store.StoreDiagConstants.IOSTATISTICS_LOGGING_LEVEL;
 import static org.apache.hadoop.fs.store.StoreUtils.split;
+import static org.apache.hadoop.fs.store.diag.S3ADiagnosticsInfo.DIRECTORY_MARKER_RETENTION;
+import static org.apache.hadoop.fs.store.diag.S3ADiagnosticsInfo.FS_S3A_CONNECTION_MAXIMUM;
+import static org.apache.hadoop.fs.store.diag.S3ADiagnosticsInfo.FS_S3A_THREADS_MAX;
+import static org.apache.hadoop.fs.store.diag.S3ADiagnosticsInfo.INPUT_FADVISE;
+import static org.apache.hadoop.fs.store.diag.S3ADiagnosticsInfo.INPUT_FADV_NORMAL;
 
 /**
  * Entry point for store applications
@@ -523,6 +528,20 @@ protected Configuration createPreconfiguredConfig()
     return conf;
   }
 
+  /**
+   * Patch the configuration for maximum S3A performance.
+   * @param conf config
+   * @return the now updated config
+   */
+  protected Configuration patchForMaxS3APerformance(Configuration conf) {
+    conf.setBoolean(DIRECTORY_MARKER_RETENTION, true);
+    final int workers = 256;
+    conf.setInt(FS_S3A_CONNECTION_MAXIMUM, workers * 2);
+    conf.setInt(FS_S3A_THREADS_MAX, workers);
+    conf.set(INPUT_FADVISE, INPUT_FADV_NORMAL);
+    return conf;
+  }
+
   protected void printFSInfoInVerbose(FileSystem fs) {
     if (isVerbose()) {
       println();

src/main/java/org/apache/hadoop/fs/store/commands/Bandwidth.java

@@ -35,6 +35,7 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.store.CommonParameters;
 import org.apache.hadoop.fs.store.MinMeanMax;
 import org.apache.hadoop.fs.store.StoreDurationInfo;
 import org.apache.hadoop.fs.store.StoreEntryPoint;
@@ -46,7 +47,11 @@
 
 import static java.util.Optional.empty;
 import static java.util.Optional.of;
+import static org.apache.hadoop.fs.store.CommonParameters.BLOCK;
+import static org.apache.hadoop.fs.store.CommonParameters.CSVFILE;
 import static org.apache.hadoop.fs.store.CommonParameters.DEFINE;
+import static org.apache.hadoop.fs.store.CommonParameters.FLUSH;
+import static org.apache.hadoop.fs.store.CommonParameters.HFLUSH;
 import static org.apache.hadoop.fs.store.CommonParameters.TOKENFILE;
 import static org.apache.hadoop.fs.store.CommonParameters.VERBOSE;
 import static org.apache.hadoop.fs.store.CommonParameters.XMLFILE;
@@ -60,19 +65,15 @@ public class Bandwidth extends StoreEntryPoint {
 
   private static final Logger LOG = LoggerFactory.getLogger(Bandwidth.class);
 
-  public static final String BLOCK = "block";
-  public static final String CSVFILE = "csv";
-  public static final String FLUSH = "flush";
-  public static final String HFLUSH = "hflush";
   public static final String KEEP = "keep";
   public static final String RENAME = "rename";
   public static final String POLICY = "policy";
 
 
   public static final String USAGE
       = "Usage: bandwidth [options] size <path>\n"
+      + optusage(BLOCK, "size", "block size in megabytes")
       + optusage(DEFINE, "key=value", "Define a property")
-      + optusage(BLOCK, "block-size", "block size in megabytes")
       + optusage(CSVFILE, "file", "CSV file to log operation details")
       + optusage(FLUSH, "flush the output after writing each block")
       + optusage(HFLUSH, "hflush() the output after writing each block")
@@ -107,14 +108,16 @@ public Bandwidth() {
         HFLUSH,
         KEEP,
         RENAME,
-        VERBOSE);
+        VERBOSE
+    );
     addValueOptions(
-        TOKENFILE,
-        XMLFILE,
         BLOCK,
         CSVFILE,
         DEFINE,
-        POLICY);
+        POLICY,
+        TOKENFILE,
+        XMLFILE
+        );
   }
 
   @Override

src/main/java/org/apache/hadoop/fs/store/diag/S3ADiagnosticsInfo.java

@@ -30,6 +30,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -43,10 +44,6 @@
 
 import static org.apache.hadoop.fs.s3a.Constants.BUFFER_DIR;
 import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_SECURE_CONNECTIONS;
-import static org.apache.hadoop.fs.s3a.Constants.INPUT_FADVISE;
-import static org.apache.hadoop.fs.s3a.Constants.INPUT_FADV_NORMAL;
-import static org.apache.hadoop.fs.s3a.Constants.INPUT_FADV_RANDOM;
-import static org.apache.hadoop.fs.s3a.Constants.INPUT_FADV_SEQUENTIAL;
 import static org.apache.hadoop.fs.s3a.Constants.SECURE_CONNECTIONS;
 import static org.apache.hadoop.fs.s3a.S3AUtils.getAWSAccessKeys;
 import static org.apache.hadoop.fs.store.StoreUtils.cat;
@@ -128,6 +125,34 @@ public class S3ADiagnosticsInfo extends StoreDiagnosticsInfo {
       "fs.s3a.fast.upload.active.blocks";
 
 
+  /**
+   * Which input strategy to use for buffering, seeking and similar when
+   * reading data.
+   * Value: {@value}
+   */
+  public static final String INPUT_FADVISE =
+      "fs.s3a.experimental.input.fadvise";
+
+  /**
+   * General input. Some seeks, some reads.
+   * Value: {@value}
+   */
+  public static final String INPUT_FADV_NORMAL = "normal";
+
+  /**
+   * Optimized for sequential access.
+   * Value: {@value}
+   */
+  public static final String INPUT_FADV_SEQUENTIAL = "sequential";
+
+  /**
+   * Optimized purely for random seek+read/positionedRead operations;
+   * The performance of sequential IO may be reduced in exchange for
+   * more efficient {@code seek()} operations.
+   * Value: {@value}
+   */
+  public static final String INPUT_FADV_RANDOM = "random";
+
   public static final String DELEGATION_TOKEN_BINDING =
       "fs.s3a.delegation.token.binding";
 

src/main/java/org/apache/hadoop/fs/store/shim/APIShim.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.hadoop.fs.shim.api;
+package org.apache.hadoop.fs.store.shim;
 
 /**
  * An API shim of type {@code TYPE}.