Periodically log transfer progress for RDE SFTP uploads

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=201172975

java/google/registry/rde/LoggingSftpProgressMonitor.java

@@ -0,0 +1,82 @@
+// Copyright 2018 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed 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.
+
+package google.registry.rde;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.flogger.FluentLogger;
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.SftpProgressMonitor;
+import google.registry.util.Clock;
+import javax.inject.Inject;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+
+/** A progress monitor for SFTP operations that writes status to logs periodically. */
+public class LoggingSftpProgressMonitor implements SftpProgressMonitor {
+
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private static final long LOGGING_CHUNK_SIZE_BYTES = 5 * 1024 * 1024;
+
+  private final Clock clock;
+  private long bytesOfLastLog = 0;
+  private int callsSinceLastLog = 0;
+  private DateTime timeOfLastLog;
+
+  @Inject
+  LoggingSftpProgressMonitor(Clock clock) {
+    this.clock = clock;
+  }
+
+  /** Nice display values for SFTP operation modes. */
+  private static final ImmutableMap<Integer, String> OPERATION_MODES =
+      new ImmutableMap.Builder<Integer, String>()
+          .put(ChannelSftp.OVERWRITE, "OVERWRITE")
+          .put(ChannelSftp.RESUME, "RESUME")
+          .put(ChannelSftp.APPEND, "APPEND")
+          .build();
+
+  @Override
+  public void init(int op, String src, String dest, long max) {
+    timeOfLastLog = clock.nowUtc();
+    logger.atInfo().log(
+        "Initiating SFTP transfer from '%s' to '%s' using mode %s, max size %,d bytes.",
+        src, dest, OPERATION_MODES.getOrDefault(op, "(unknown)"), max);
+  }
+
+  @Override
+  public boolean count(long count) {
+    callsSinceLastLog++;
+    long bytesSinceLastLog = count - bytesOfLastLog;
+    if (bytesSinceLastLog > LOGGING_CHUNK_SIZE_BYTES) {
+      DateTime now = clock.nowUtc();
+      logger.atInfo().log(
+          "%,d more bytes transmitted in %,d ms; %,d bytes in total. [%,d calls to count()]",
+          bytesSinceLastLog,
+          new Duration(timeOfLastLog, now).getMillis(),
+          count,
+          callsSinceLastLog);
+      bytesOfLastLog = count;
+      callsSinceLastLog = 0;
+      timeOfLastLog = now;
+    }
+    // True means that the upload continues.
+    return true;
+  }
+
+  @Override
+  public void end() {
+    logger.atInfo().log("SFTP operation finished.");
+  }
+}

java/google/registry/rde/RdeModule.java

@@ -24,6 +24,8 @@ import static google.registry.request.RequestParameters.extractSetOfParameters;
 
 import com.google.appengine.api.taskqueue.Queue;
 import com.google.common.collect.ImmutableSet;
+import com.jcraft.jsch.SftpProgressMonitor;
+import dagger.Binds;
 import dagger.Module;
 import dagger.Provides;
 import google.registry.request.Parameter;
@@ -38,7 +40,7 @@ import org.joda.time.DateTime;
  * @see "google.registry.module.backend.BackendRequestComponent"
  */
 @Module
-public final class RdeModule {
+public abstract class RdeModule {
 
   public static final String PARAM_WATERMARK = "watermark";
   public static final String PARAM_WATERMARKS = "watermarks";
@@ -101,4 +103,10 @@ public final class RdeModule {
   static Queue provideQueueRdeReport() {
     return getQueue("rde-report");
   }
+
+  @Binds
+  abstract SftpProgressMonitor provideSftpProgressMonitor(
+      LoggingSftpProgressMonitor loggingSftpProgressMonitor);
+
+  private RdeModule() {}
 }

java/google/registry/rde/RdeUploadAction.java

@@ -33,6 +33,7 @@ import com.google.common.flogger.FluentLogger;
 import com.google.common.io.ByteStreams;
 import com.jcraft.jsch.JSch;
 import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.SftpProgressMonitor;
 import dagger.Lazy;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.gcs.GcsUtils;
@@ -108,6 +109,7 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
   @Inject RydePgpFileOutputStreamFactory pgpFileFactory;
   @Inject RydePgpSigningOutputStreamFactory pgpSigningFactory;
   @Inject RydeTarOutputStreamFactory tarFactory;
+  @Inject SftpProgressMonitor sftpProgressMonitor;
   @Inject TaskQueueUtils taskQueueUtils;
   @Inject Retrier retrier;
   @Inject @Parameter(RequestParameters.PARAM_TLD) String tld;
@@ -202,8 +204,8 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
    *   }</pre>
    */
   @VisibleForTesting
-  protected void upload(
+  protected void upload(GcsFilename xmlFile, long xmlLength, DateTime watermark, String name)
-      GcsFilename xmlFile, long xmlLength, DateTime watermark, String name) throws Exception {
+      throws Exception {
     logger.atInfo().log("Uploading XML file '%s' to remote path '%s'.", xmlFile, uploadUrl);
     try (InputStream gcsInput = gcsUtils.openInputStream(xmlFile);
         Ghostryde.Decryptor decryptor = ghostryde.openDecryptor(gcsInput, stagingDecryptionKey);
@@ -214,7 +216,8 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
         byte[] signature;
         String rydeFilename = name + ".ryde";
         GcsFilename rydeGcsFilename = new GcsFilename(bucket, rydeFilename);
-        try (OutputStream ftpOutput = ftpChan.get().put(rydeFilename, OVERWRITE);
+        try (OutputStream ftpOutput =
+                ftpChan.get().put(rydeFilename, sftpProgressMonitor, OVERWRITE);
             OutputStream gcsOutput = gcsUtils.openOutputStream(rydeGcsFilename);
             TeeOutputStream teeOutput = new TeeOutputStream(asList(ftpOutput, gcsOutput));
             RydePgpSigningOutputStream signer = pgpSigningFactory.create(teeOutput, signingKey)) {