Migrate fully to AWS SDK V2

build.gradle

@@ -38,15 +38,18 @@ allprojects {
 
       // AWS Services - SDK v1 (for EC2, AutoScaling, STS)
       implementation 'com.amazonaws:aws-java-sdk-core:latest.release'
-      implementation 'com.amazonaws:aws-java-sdk-sns:latest.release'
       implementation 'com.amazonaws:aws-java-sdk-ec2:latest.release'
       implementation 'com.amazonaws:aws-java-sdk-autoscaling:latest.release'
       implementation 'com.amazonaws:aws-java-sdk-sts:latest.release'
 
-      // AWS Services - SDK v2 (for S3, SNS)
-      implementation platform('software.amazon.awssdk:bom:2.20.0')
+      // AWS Services - SDK v2 (for S3)
+      implementation platform('software.amazon.awssdk:bom:2.27.21')
       implementation 'software.amazon.awssdk:s3'
-      implementation 'software.amazon.awssdk:sts' // Needed for S3 role assumption
+        implementation 'software.amazon.awssdk:sts' // Needed for S3 role assumption
+        implementation("software.amazon.awssdk:imds")
+        implementation("software.amazon.awssdk:ec2")
+        implementation("software.amazon.awssdk:autoscaling")
+        implementation("software.amazon.awssdk:url-connection-client")
 
       implementation 'com.google.inject:guice:4.2.2'
       implementation 'com.google.inject.extensions:guice-servlet:4.2.2'

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

@@ -18,7 +18,6 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import com.netflix.priam.aws.auth.IS3Credential;
 import com.netflix.priam.backup.AbstractBackupPath;
 import com.netflix.priam.backup.BackupRestoreException;
 import com.netflix.priam.backup.DynamicRateLimiter;
@@ -27,9 +26,9 @@
 import com.netflix.priam.compress.CompressionType;
 import com.netflix.priam.compress.ICompression;
 import com.netflix.priam.config.IConfiguration;
+import com.netflix.priam.cred.ICredential;
 import com.netflix.priam.identity.config.InstanceInfo;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.utils.BoundedExponentialRetryCallable;
 import com.netflix.priam.utils.ByteBufferInputStream;
 import com.netflix.priam.utils.SystemUtils;
@@ -70,15 +69,14 @@ public class S3FileSystem extends S3FileSystemBase {
 
     @Inject
     public S3FileSystem(
-            @Named("awss3roleassumption") IS3Credential cred,
+            @Named("awss3roleassumption") ICredential cred,
             Provider<AbstractBackupPath> pathProvider,
             ICompression compress,
             final IConfiguration config,
             BackupMetrics backupMetrics,
-            BackupNotificationMgr backupNotificationMgr,
             InstanceInfo instanceInfo,
             DynamicRateLimiter dynamicRateLimiter) {
-        super(pathProvider, compress, config, backupMetrics, backupNotificationMgr);
+        super(pathProvider, compress, config, backupMetrics);
         s3Client =
                 S3Client.builder()
                         .credentialsProvider(cred.getAwsCredentialProvider())

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

@@ -13,29 +13,27 @@
  */
 package com.netflix.priam.aws;
 
-import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.RateLimiter;
 import com.netflix.priam.backup.AbstractBackupPath;
 import com.netflix.priam.backup.AbstractFileSystem;
 import com.netflix.priam.backup.BackupRestoreException;
 import com.netflix.priam.compress.ICompression;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.scheduler.BlockingSubmitThreadPoolExecutor;
-import java.nio.file.Path;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Optional;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.stream.Collectors;
-import javax.inject.Provider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.*;
 
+import javax.inject.Provider;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
+
 public abstract class S3FileSystemBase extends AbstractFileSystem {
     private static final int MAX_CHUNKS = 9995; // 10K is AWS limit, minus a small buffer
     private static final Logger logger = LoggerFactory.getLogger(S3FileSystemBase.class);
@@ -50,9 +48,8 @@ public abstract class S3FileSystemBase extends AbstractFileSystem {
             Provider<AbstractBackupPath> pathProvider,
             ICompression compress,
             final IConfiguration config,
-            BackupMetrics backupMetrics,
-            BackupNotificationMgr backupNotificationMgr) {
-        super(config, backupMetrics, backupNotificationMgr, pathProvider);
+            BackupMetrics backupMetrics) {
+        super(config, backupMetrics, pathProvider);
         this.compress = compress;
         this.config = config;
 

priam/src/main/java/com/netflix/priam/aws/auth/EC2RoleAssumptionCredential.java

@@ -13,19 +13,23 @@
  */
 package com.netflix.priam.aws.auth;
 
-import com.amazonaws.auth.AWSCredentialsProvider;
-import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.cred.ICredential;
 import com.netflix.priam.identity.config.InstanceInfo;
+import org.apache.commons.lang3.Validate;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+
 import javax.inject.Inject;
 
 public class EC2RoleAssumptionCredential implements ICredential {
-    private static final String AWS_ROLE_ASSUMPTION_SESSION_NAME = "AwsRoleAssumptionSession";
+    private static final String SESSION_NAME = "AwsRoleAssumptionSession";
     private final ICredential cred;
     private final IConfiguration config;
     private final InstanceInfo instanceInfo;
-    private AWSCredentialsProvider stsSessionCredentialsProvider;
+    private AwsCredentialsProvider stsSessionCredentialsProvider;
 
     @Inject
     public EC2RoleAssumptionCredential(
@@ -36,43 +40,25 @@ public EC2RoleAssumptionCredential(
     }
 
     @Override
-    public AWSCredentialsProvider getAwsCredentialProvider() {
-        if (this.config.isDualAccount() || this.stsSessionCredentialsProvider == null) {
+    public AwsCredentialsProvider getAwsCredentialProvider() {
+        if (this.stsSessionCredentialsProvider == null) {
             synchronized (this) {
                 if (this.stsSessionCredentialsProvider == null) {
-
-                    String roleArn;
-                    /**
-                     * Create the assumed IAM role based on the environment. For example, if the
-                     * current environment is VPC, then the assumed role is for EC2 classic, and
-                     * vice versa.
-                     */
-                    if (instanceInfo.getInstanceEnvironment()
-                            == InstanceInfo.InstanceEnvironment.CLASSIC) {
-                        roleArn = this.config.getClassicEC2RoleAssumptionArn();
-                        // Env is EC2 classic --> IAM assumed role for VPC created
-                    } else {
-                        roleArn = this.config.getVpcEC2RoleAssumptionArn();
-                        // Env is VPC --> IAM assumed role for EC2 classic created.
-                    }
-
-                    //
-                    if (roleArn == null || roleArn.isEmpty())
-                        throw new NullPointerException(
-                                "Role ARN is null or empty probably due to missing config entry");
-
-                    /**
-                     * Get handle to an implementation that uses AWS Security Token Service (STS) to
-                     * create temporary, short-lived session with explicit refresh for session/token
-                     * expiration.
-                     */
+                    String roleArn =
+                            instanceInfo.getInstanceEnvironment() == InstanceInfo.InstanceEnvironment.CLASSIC
+                                    ? this.config.getClassicEC2RoleAssumptionArn()
+                                    : this.config.getVpcEC2RoleAssumptionArn();
+                    Validate.notEmpty(roleArn, "roleArn is empty");
                     try {
+                        StsClient stsClient =
+                                StsClient.builder().credentialsProvider(cred.getAwsCredentialProvider()).build();
+                        AssumeRoleRequest assumeRoleRequest =
+                                AssumeRoleRequest.builder().roleArn(roleArn).roleSessionName(SESSION_NAME).build();
                         this.stsSessionCredentialsProvider =
-                                new STSAssumeRoleSessionCredentialsProvider(
-                                        this.cred.getAwsCredentialProvider(),
-                                        roleArn,
-                                        AWS_ROLE_ASSUMPTION_SESSION_NAME);
-
+                                StsAssumeRoleCredentialsProvider.builder()
+                                        .stsClient(stsClient)
+                                        .refreshRequest(assumeRoleRequest)
+                                        .build();
                     } catch (Exception ex) {
                         throw new IllegalStateException(
                                 "Exception in getting handle to AWS Security Token Service (STS).  Msg: "
@@ -82,7 +68,6 @@ public AWSCredentialsProvider getAwsCredentialProvider() {
                 }
             }
         }
-
         return this.stsSessionCredentialsProvider;
     }
 }

priam/src/main/java/com/netflix/priam/aws/auth/S3InstanceCredential.java → priam/src/main/java/com/netflix/priam/aws/auth/InstanceCredential.java

@@ -13,18 +13,18 @@
  */
 package com.netflix.priam.aws.auth;
 
-import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import com.netflix.priam.cred.ICredential;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider;
 
 /*
- * Provides credentials from the S3 instance.
+ * Provides credentials that work with SDK V2
  */
-public class S3InstanceCredential implements IS3Credential {
+public class InstanceCredential implements ICredential {
 
     private final InstanceProfileCredentialsProvider credentialsProvider;
 
-    public S3InstanceCredential() {
+    public InstanceCredential() {
         this.credentialsProvider = InstanceProfileCredentialsProvider.create();
     }
 

priam/src/main/java/com/netflix/priam/aws/auth/S3RoleAssumptionCredential.java

@@ -14,67 +14,52 @@
 package com.netflix.priam.aws.auth;
 
 import com.netflix.priam.config.IConfiguration;
-import javax.inject.Inject;
-import javax.inject.Named;
-import javax.inject.Singleton;
+import com.netflix.priam.cred.ICredential;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.auth.credentials.AwsCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.services.sts.StsClient;
 import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
 import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
 
+import javax.inject.Inject;
+import javax.inject.Named;
+import javax.inject.Singleton;
+
 @Singleton
-public class S3RoleAssumptionCredential implements IS3Credential {
-    private static final String AWS_ROLE_ASSUMPTION_SESSION_NAME = "S3RoleAssumptionSession";
+public class S3RoleAssumptionCredential implements ICredential {
+    private static final String SESSION_NAME = "S3RoleAssumptionSession";
     private static final Logger logger = LoggerFactory.getLogger(S3RoleAssumptionCredential.class);
 
-    private final IS3Credential cred;
+    private final ICredential cred;
     private final IConfiguration config;
-    private AwsCredentialsProvider stsSessionCredentialsProvider;
+    private AwsCredentialsProvider credentialsProvider;
 
     @Inject
-    public S3RoleAssumptionCredential(
-            @Named("s3") IS3Credential cred, IConfiguration config) {
+    public S3RoleAssumptionCredential(@Named("s3") ICredential cred, IConfiguration config) {
         this.cred = cred;
         this.config = config;
     }
 
     public AwsCredentialsProvider getAwsCredentialProvider() {
-        if (this.stsSessionCredentialsProvider == null) {
+        if (this.credentialsProvider == null) {
             synchronized (this) {
-                if (this.stsSessionCredentialsProvider == null) {
-
-                    final String roleArn = this.config.getAWSRoleAssumptionArn();
-                    // IAM role created for bucket own by account "awsprodbackup"
+                if (this.credentialsProvider == null) {
+                    final String roleArn = config.getAWSRoleAssumptionArn();
                     if (roleArn == null || roleArn.isEmpty()) {
-                        logger.warn(
-                                "Role ARN is null or empty probably due to missing config entry. Falling back to instance level credentials");
-                        this.stsSessionCredentialsProvider = this.cred.getAwsCredentialProvider();
-                        // throw new NullPointerException("Role ARN is null or empty probably due to
-                        // missing config entry");
+                        logger.warn("Role ARN is empty due to missing config. Using instance level credentials");
+                        credentialsProvider = cred.getAwsCredentialProvider();
                     } else {
-                        // Get handle to an implementation that uses AWS Security Token Service
-                        // (STS) to create temporary, short-lived session with explicit refresh for
-                        // session/token expiration.
                         try {
-
-                            StsClient stsClient = StsClient.builder()
-                                    .credentialsProvider(cred.getAwsCredentialProvider())
-                                    .build();
-
-                            AssumeRoleRequest assumeRoleRequest = AssumeRoleRequest.builder()
-                                    .roleArn(roleArn)
-                                    .roleSessionName(AWS_ROLE_ASSUMPTION_SESSION_NAME)
-                                    .build();
-
-                            this.stsSessionCredentialsProvider =
+                            StsClient stsClient =
+                                    StsClient.builder().credentialsProvider(cred.getAwsCredentialProvider()).build();
+                            AssumeRoleRequest assumeRoleRequest =
+                                    AssumeRoleRequest.builder().roleArn(roleArn).roleSessionName(SESSION_NAME).build();
+                            credentialsProvider =
                                     StsAssumeRoleCredentialsProvider.builder()
                                             .stsClient(stsClient)
                                             .refreshRequest(assumeRoleRequest)
                                             .build();
-
                         } catch (Exception ex) {
                             throw new IllegalStateException(
                                     "Exception in getting handle to AWS Security Token Service (STS).  Msg: "
@@ -85,7 +70,6 @@ public AwsCredentialsProvider getAwsCredentialProvider() {
                 }
             }
         }
-
-        return this.stsSessionCredentialsProvider;
+        return credentialsProvider;
     }
 }