Merge pull request #2085 from HubSpot/s3_uploader_reconcile_files

Periodically reconcile the list of metadata files
HubSpot · Mar 24, 2020 · 65fe523 · 65fe523
2 parents 29a128d + b8c81f9
commit 65fe523
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 0 deletions.
diff --git a/...ploader/src/main/java/com/hubspot/singularity/s3uploader/SingularityS3UploaderDriver.java b/...ploader/src/main/java/com/hubspot/singularity/s3uploader/SingularityS3UploaderDriver.java
@@ -82,6 +82,7 @@ public class SingularityS3UploaderDriver extends WatchServiceHelper implements S
   private final Map<S3UploadMetadata, CompletableFuture<Integer>> immediateUploadersFutures;
 
   private ScheduledFuture<?> future;
+  private ScheduledFuture<?> fileSyncFuture;
 
   @Inject
   public SingularityS3UploaderDriver(SingularityRunnerBaseConfiguration baseConfiguration, SingularityS3UploaderConfiguration configuration, SingularityS3Configuration s3Configuration,
@@ -187,6 +188,17 @@ public void startAndWait() {
       }
     }, configuration.getCheckUploadsEverySeconds(), configuration.getCheckUploadsEverySeconds(), TimeUnit.SECONDS);
 
+    fileSyncFuture = scheduler.scheduleAtFixedRate(() -> {
+      runLock.lock();
+      try {
+        readInitialFiles();
+      } catch (Throwable t) {
+        throw new RuntimeException(t);
+      } finally {
+        runLock.unlock();
+      }
+    }, configuration.getRecheckFilesEverySeconds(), configuration.getRecheckFilesEverySeconds(), TimeUnit.SECONDS);
+
     try {
       super.watch();
     } catch (Throwable t) {
@@ -209,6 +221,10 @@ public void shutdown() {
       runLock.unlock();
     }
 
+    if (fileSyncFuture != null) {
+      fileSyncFuture.cancel(true);
+    }
+
     if (future != null) {
       future.cancel(false);
     }

diff --git a/...in/java/com/hubspot/singularity/s3uploader/config/SingularityS3UploaderConfiguration.java b/...in/java/com/hubspot/singularity/s3uploader/config/SingularityS3UploaderConfiguration.java
@@ -32,6 +32,10 @@ public class SingularityS3UploaderConfiguration extends BaseRunnerConfiguration
   @JsonProperty
   private long checkUploadsEverySeconds = 600;
 
+  @Min(60)
+  @JsonProperty
+  private long recheckFilesEverySeconds = 600;
+
   @Min(1)
   @JsonProperty
   private long stopCheckingAfterMillisWithoutNewFile = TimeUnit.HOURS.toMillis(168);
@@ -104,6 +108,14 @@ public void setCheckUploadsEverySeconds(long checkUploadsEverySeconds) {
     this.checkUploadsEverySeconds = checkUploadsEverySeconds;
   }
 
+  public long getRecheckFilesEverySeconds() {
+    return recheckFilesEverySeconds;
+  }
+
+  public void setRecheckFilesEverySeconds(long recheckFilesEverySeconds) {
+    this.recheckFilesEverySeconds = recheckFilesEverySeconds;
+  }
+
   public long getStopCheckingAfterMillisWithoutNewFile() {
     return stopCheckingAfterMillisWithoutNewFile;
   }