Warmup objectCache with the latest validated snapshot files to reduce HEAD requests to AWS (#1118)

* add interface for enableSynchronizingDroppedColumns

* Revert "add interface for enableSynchronizingDroppedColumns"

This reverts commit 6d54ae4.

* Warmup objectCache with the latest validated snapshot files to reduce HEAD requests to AWS

* PR comments

Author: chengw-netflix

priam/src/main/java/com/netflix/priam/backup/AbstractFileSystem.java

@@ -26,6 +26,7 @@
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.netflix.priam.backup.AbstractBackupPath.BackupFileType;
+import com.netflix.priam.backupv2.IMetaProxy;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.merics.BackupMetrics;
 import com.netflix.priam.notification.BackupNotificationMgr;
@@ -37,10 +38,7 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Instant;
-import java.util.Date;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.*;
 import javax.inject.Inject;
 import javax.inject.Provider;
@@ -236,10 +234,12 @@ public Long retriableCall() throws Exception {
         return path;
     }
 
-    private void addObjectCache(Path remotePath) {
+    @Override
+    public void addObjectCache(Path remotePath) {
         objectCache.put(remotePath, Boolean.TRUE);
     }
 
+
     @Override
     public boolean checkObjectExists(Path remotePath) {
         // Check in cache, if remote file exists.

priam/src/main/java/com/netflix/priam/backup/BackupHelper.java

@@ -4,6 +4,9 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.inject.ImplementedBy;
+import com.netflix.priam.backupv2.IMetaProxy;
+import com.netflix.priam.backup.IBackupStatusMgr;
+
 import java.io.File;
 import java.io.IOException;
 import java.time.Instant;
@@ -26,4 +29,6 @@ ImmutableList<ListenableFuture<AbstractBackupPath>> uploadAndDeleteAllFiles(
 
     ImmutableSet<AbstractBackupPath> getBackupPaths(
             File dir, AbstractBackupPath.BackupFileType type) throws IOException;
-}
+
+    void warmupCache(IMetaProxy metaProxy, IBackupStatusMgr snapshotStatusMgr) throws Exception;
+}

priam/src/main/java/com/netflix/priam/backup/BackupHelperImpl.java

@@ -5,14 +5,21 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.netflix.priam.backupv2.IMetaProxy;
 import com.netflix.priam.compress.CompressionType;
 import com.netflix.priam.config.BackupsToCompress;
 import com.netflix.priam.config.IConfiguration;
+import com.netflix.priam.utils.DateUtil;
+
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Comparator;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Stream;
 import javax.inject.Inject;
@@ -84,6 +91,25 @@ public ImmutableSet<AbstractBackupPath> getBackupPaths(
         return bps.build();
     }
 
+    @Override
+    public void warmupCache(IMetaProxy metaProxy, IBackupStatusMgr snapshotStatusMgr) throws Exception {
+        Instant now = Instant.now();
+        DateUtil.DateRange dateRange = new DateUtil.DateRange(now.minus(1, ChronoUnit.DAYS), now);
+        Optional<String> snapshotLocation = snapshotStatusMgr.getLatestBackupMetadata(dateRange)
+                .stream()
+                .filter(backupMetadata -> backupMetadata.getLastValidated() != null)
+                .max(Comparator.comparing(BackupMetadata::getStart))
+                .map(BackupMetadata::getSnapshotLocation);
+        if (snapshotLocation.isPresent()) {
+            AbstractBackupPath backupPath = pathFactory.get();
+            backupPath.parseRemote(snapshotLocation.get());
+            BackupRestoreUtil.getMostRecentSnapshotPaths(backupPath, metaProxy, pathFactory)
+                    .stream()
+                    .map(abp -> Paths.get(abp.getRemotePath()))
+                    .forEach(fs::addObjectCache);
+        }
+    }
+
     private CompressionType getCorrectCompressionAlgorithm(
             AbstractBackupPath path, Set<String> compressedFiles) {
         if (path.getLastModified().toEpochMilli() < config.getCompressionTransitionEpochMillis()) {

priam/src/main/java/com/netflix/priam/backup/IBackupFileSystem.java

@@ -172,4 +172,7 @@ default boolean checkObjectExists(Path remotePath) {
 
     /** Clear the cache for the backup file system, if any. */
     void clearCache();
+
+    /** Add remote backup path to the object cache*/
+    void addObjectCache(Path remotePath);
 }

priam/src/main/java/com/netflix/priam/backupv2/BackupV2Service.java

@@ -64,6 +64,10 @@ public void scheduleService() throws Exception {
         if (snapshotMetaTimer == null) {
             SnapshotMetaTask.cleanOldBackups(configuration);
         }
+
+        // Warm up object cache
+        snapshotMetaTask.warmCache();
+
         scheduleTask(scheduler, SnapshotMetaTask.class, snapshotMetaTimer);
 
         if (snapshotMetaTimer != null) {

priam/src/main/java/com/netflix/priam/backupv2/SnapshotMetaTask.java

@@ -255,6 +255,10 @@ public String getName() {
         return JOBNAME;
     }
 
+    public void warmCache() throws Exception {
+        backupHelper.warmupCache(metaProxy, snapshotStatusMgr);
+    }
+
     private void uploadAllFiles(final File backupDir) throws Exception {
         // Process all the snapshots with SNAPSHOT_PREFIX. This will ensure that we "resume" the
         // uploads of previous snapshot leftover as Priam restarted or any failure for any reason

priam/src/test/groovy/com.netflix.priam/backup/TestBackupRestoreUtil.groovy

@@ -59,7 +59,7 @@ class TestBackupRestoreUtil extends Specification {
     }
 
 
-    def "Expected exception KS #keyspace and CF #columnfamily with configuration include #configIncludeFilter and exclude #configExcludeFilter"() {
+    def "Expected exception for shouldOmitFromBackup KS #keyspace and CF #columnfamily with configuration include #configIncludeFilter and exclude #configExcludeFilter"() {
         when:
         new BackupRestoreUtil(configIncludeFilter, configExcludeFilter).shouldOmitFromBackup(keyspace, columnfamily)
 
@@ -106,7 +106,7 @@ class TestBackupRestoreUtil extends Specification {
         null                | null                | "system"   | "peers_v2"                 || true
     }
 
-    def "Expected exception KS #keyspace and CF #columnfamily with configuration include #configIncludeFilter and exclude #configExcludeFilter"() {
+    def "Expected exception for shouldOmitFromRestore KS #keyspace and CF #columnfamily with configuration include #configIncludeFilter and exclude #configExcludeFilter"() {
         when:
         new BackupRestoreUtil(configIncludeFilter, configExcludeFilter).shouldOmitFromRestore(keyspace, columnfamily)