Persist snapshot validation markers to S3.

Author: mattl-netflix

priam/src/main/java/com/netflix/priam/aws/RemoteBackupPath.java

@@ -50,7 +50,7 @@ private ImmutableList.Builder<String> getPrefix() {
     /* This will ensure that there is some randomness in the path at the start so that remote file systems
     can hash the contents better when we have lot of clusters backing up at the same remote location.
     */
-    private String prependHash(String appName) {
+    public static String prependHash(String appName) {
         return String.format("%d_%s", appName.hashCode() % 10000, appName);
     }
 

priam/src/main/java/com/netflix/priam/aws/S3FileSystem.java

@@ -104,6 +104,11 @@ protected void downloadFileImpl(AbstractBackupPath path, String suffix)
         }
     }
 
+    @Override
+    public void putObject(String bucket, String key, String value) {
+        s3Client.putObject(bucket, key, value);
+    }
+
     private ObjectMetadata getObjectMetadata(File file) {
         ObjectMetadata ret = new ObjectMetadata();
         long lastModified = file.lastModified();

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

@@ -175,4 +175,7 @@ default boolean checkObjectExists(Path remotePath) {
 
     /** Add remote backup path to the object cache*/
     void addObjectCache(Path remotePath);
+    
+    /** Synchronously put the contents of an in-memory String to the blob store. **/
+    void putObject(String bucket, String key, String value);
 }

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

@@ -44,11 +44,12 @@ public class BackupVerificationTask extends Task {
     private static final Logger logger = LoggerFactory.getLogger(BackupVerificationTask.class);
     public static final String JOBNAME = "BackupVerificationService";
 
-    private IBackupRestoreConfig backupRestoreConfig;
-    private BackupVerification backupVerification;
-    private BackupMetrics backupMetrics;
-    private InstanceState instanceState;
-    private BackupNotificationMgr backupNotificationMgr;
+    private final IBackupRestoreConfig backupRestoreConfig;
+    private final BackupVerification backupVerification;
+    private final BackupMetrics backupMetrics;
+    private final InstanceState instanceState;
+    private final BackupNotificationMgr backupNotificationMgr;
+    private final SnapshotValidationMarkerWriter snapshotValidationMarkerWriter;
 
     @Inject
     public BackupVerificationTask(
@@ -57,13 +58,15 @@ public BackupVerificationTask(
             BackupVerification backupVerification,
             BackupMetrics backupMetrics,
             InstanceState instanceState,
-            BackupNotificationMgr backupNotificationMgr) {
+            BackupNotificationMgr backupNotificationMgr,
+            SnapshotValidationMarkerWriter snapshotValidationMarkerWriter) {
         super(configuration);
         this.backupRestoreConfig = backupRestoreConfig;
         this.backupVerification = backupVerification;
         this.backupMetrics = backupMetrics;
         this.instanceState = instanceState;
         this.backupNotificationMgr = backupNotificationMgr;
+        this.snapshotValidationMarkerWriter = snapshotValidationMarkerWriter;
     }
 
     @Override
@@ -104,6 +107,7 @@ public void execute() throws Exception {
                                     snapshotKey);
                             backupNotificationMgr.notify(
                                     snapshotKey, result.getStart().toInstant());
+                            snapshotValidationMarkerWriter.write(snapshotKey);
                         });
 
         if (verifiedBackups.isEmpty()) {

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

@@ -0,0 +1,53 @@
+package com.netflix.priam.backupv2;
+
+import com.netflix.priam.aws.RemoteBackupPath;
+import com.netflix.priam.backup.AbstractBackupPath;
+import com.netflix.priam.backup.IBackupFileSystem;
+import com.netflix.priam.config.IConfiguration;
+import com.netflix.priam.identity.config.InstanceInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import javax.inject.Provider;
+
+public class SnapshotValidationMarkerWriter {
+    private static final Logger logger = LoggerFactory.getLogger(SnapshotValidationMarkerWriter.class);
+    private static final String METADATA_PREFIX = "metadata";
+    private final IConfiguration config;
+    private final InstanceInfo instanceInfo;
+    private final Provider<AbstractBackupPath> pathProvider;
+    private final IBackupFileSystem fs;
+
+    @Inject
+    public SnapshotValidationMarkerWriter(
+            IConfiguration config,
+            InstanceInfo instanceInfo,
+            Provider<AbstractBackupPath> pathProvider,
+            IBackupFileSystem fs) {
+        this.config = config;
+        this.instanceInfo = instanceInfo;
+        this.pathProvider = pathProvider;
+        this.fs = fs;
+    }
+
+    public void write(String remotePath) {
+        AbstractBackupPath path = pathProvider.get();
+        path.parseRemote(remotePath);
+        String key = AbstractBackupPath.PATH_JOINER.join(
+                METADATA_PREFIX,
+                RemoteBackupPath.prependHash(path.getClusterName()),
+                path.getLastModified().toEpochMilli(),
+                instanceInfo.getRac(),
+                path.getToken());
+        try {
+            fs.putObject(config.getBackupPrefix(), key, remotePath);
+        } catch (Exception e) {
+            logger.error("Failed to put snapshot verification marker. bucket: {}, key: {}, value: {}, message: {}",
+                    config.getBackupPrefix(),
+                    key,
+                    remotePath,
+                    e.getLocalizedMessage());
+        }
+    }
+}

priam/src/main/java/com/netflix/priam/resources/BackupServletV2.java

@@ -24,6 +24,7 @@
 import com.netflix.priam.backupv2.BackupV2Service;
 import com.netflix.priam.backupv2.IMetaProxy;
 import com.netflix.priam.backupv2.SnapshotMetaTask;
+import com.netflix.priam.backupv2.*;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.utils.DateUtil;
@@ -63,6 +64,7 @@ public class BackupServletV2 {
     private final BackupNotificationMgr backupNotificationMgr;
     private final IConfiguration config;
     private final DirectorySize directorySize;
+    private final SnapshotValidationMarkerWriter snapshotValidationMarkerWriter;
 
     private static final String REST_SUCCESS = "[\"ok\"]";
 
@@ -78,7 +80,8 @@ public BackupServletV2(
             BackupV2Service backupService,
             BackupNotificationMgr backupNotificationMgr,
             IConfiguration config,
-            DirectorySize directorySize) {
+            DirectorySize directorySize,
+            SnapshotValidationMarkerWriter snapshotValidationMarkerWriter) {
         this.backupStatusMgr = backupStatusMgr;
         this.backupVerification = backupVerification;
         this.snapshotMetaService = snapshotMetaService;
@@ -90,6 +93,7 @@ public BackupServletV2(
         this.backupNotificationMgr = backupNotificationMgr;
         this.config = config;
         this.directorySize = directorySize;
+        this.snapshotValidationMarkerWriter = snapshotValidationMarkerWriter;
     }
 
     @GET
@@ -164,6 +168,7 @@ public Response validateV2SnapshotByDate(
                 result.get().remotePath);
 
         backupNotificationMgr.notify(result.get().remotePath, result.get().snapshotInstant);
+        snapshotValidationMarkerWriter.write(result.get().remotePath);
 
         return Response.ok(result.get().toString()).build();
     }

priam/src/test/java/com/netflix/priam/backup/FakeBackupFileSystem.java

@@ -141,4 +141,9 @@ protected long uploadFileImpl(AbstractBackupPath path, Instant target)
         addFile(path.getRemotePath());
         return path.getBackupFile().length();
     }
+    
+    @Override
+    public void putObject(String bucket, String key, String value) {
+        throw new UnsupportedOperationException();
+    }
 }

priam/src/test/java/com/netflix/priam/backup/NullBackupFileSystem.java

@@ -72,4 +72,7 @@ protected long uploadFileImpl(AbstractBackupPath path, Instant target)
             throws BackupRestoreException {
         return 0;
     }
+
+    @Override
+    public void putObject(String bucket, String key, String value) {}
 }