feat: upgrade the credential refresh mechanism and add fallbacks for …

pom.xml

@@ -56,8 +56,7 @@
     <properties>
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
-        <powermock.version>2.0.2</powermock.version>
-        <mockito.version>3.0.0</mockito.version>
+        <mockito.version>4.11.0</mockito.version>
     </properties>
 
     <dependencies>
@@ -69,7 +68,7 @@
         <dependency>
             <groupId>com.aliyun</groupId>
             <artifactId>tea</artifactId>
-            <version>[1.1.14, 2.0.0)</version>
+            <version>[1.2.0, 2.0.0)</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>
@@ -94,38 +93,16 @@
             <version>0.8.8</version>
             <classifier>runtime</classifier>
         </dependency>
-        <dependency>
-            <groupId>org.powermock</groupId>
-            <artifactId>powermock-api-mockito2</artifactId>
-            <version>${powermock.version}</version>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
             <version>${mockito.version}</version>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.powermock</groupId>
-            <artifactId>powermock-module-junit4</artifactId>
-            <version>2.0.2</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>javax.xml.bind</groupId>
-            <artifactId>jaxb-api</artifactId>
-            <version>2.3.0</version>
-        </dependency>
-        <dependency>
-            <groupId>com.sun.xml.bind</groupId>
-            <artifactId>jaxb-core</artifactId>
-            <version>2.3.0</version>
-        </dependency>
-        <dependency>
-            <groupId>com.sun.xml.bind</groupId>
-            <artifactId>jaxb-impl</artifactId>
-            <version>2.3.0</version>
+            <groupId>commons-codec</groupId>
+            <artifactId>commons-codec</artifactId>
+            <version>1.17.1</version>
         </dependency>
     </dependencies>
 
@@ -229,7 +206,7 @@
                         <format>html</format>
                         <format>xml</format>
                     </formats>
-                    <check />
+                    <check/>
                 </configuration>
             </plugin>
             <plugin>

src/main/java/com/aliyun/credentials/models/CredentialModel.java

@@ -87,6 +87,11 @@ public String getProviderName() {
         return providerName;
     }
 
+    @Override
+    public String toString() {
+        return String.format("Credential(accessKeyId=%s, accessKeySecret=%s, securityToken=%s, providerName=%s, expiration=%s)", accessKeyId, accessKeySecret, securityToken, providerName, expiration);
+    }
+
     public static final class Builder {
         private String accessKeyId;
         private String accessKeySecret;

src/main/java/com/aliyun/credentials/policy/NonBlocking.java

@@ -0,0 +1,58 @@
+package com.aliyun.credentials.policy;
+
+import com.aliyun.tea.logging.ClientLogger;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+public class NonBlocking implements PrefetchStrategy {
+    private static final ClientLogger logger = new ClientLogger(NonBlocking.class);
+    private static final int MAX_CONCURRENT_REFRESHES = 100;
+    private static final Semaphore CONCURRENT_REFRESH_LEASES = new Semaphore(MAX_CONCURRENT_REFRESHES);
+    private final AtomicBoolean currentlyRefreshing = new AtomicBoolean(false);
+
+    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(1, Integer.MAX_VALUE, 60, SECONDS,
+            new SynchronousQueue<>(),
+            new ThreadFactory() {
+                public Thread newThread(Runnable r) {
+                    Thread t = Executors.defaultThreadFactory().newThread(r);
+                    t.setName("non-blocking-refresh");
+                    t.setDaemon(true);
+                    return t;
+                }
+            });
+
+    @Override
+    public void prefetch(Runnable valueUpdater) {
+        if (currentlyRefreshing.compareAndSet(false, true)) {
+            if (!CONCURRENT_REFRESH_LEASES.tryAcquire()) {
+                logger.warning("Skipping a background refresh task because there are too many other tasks running.");
+                currentlyRefreshing.set(false);
+                return;
+            }
+            try {
+                executor.submit(() -> {
+                    try {
+                        valueUpdater.run();
+                    } catch (Throwable t) {
+                        logger.logThrowableAsWarning(t);
+                    } finally {
+                        CONCURRENT_REFRESH_LEASES.release();
+                        currentlyRefreshing.set(false);
+                    }
+                });
+            } catch (Throwable t) {
+                logger.logThrowableAsWarning(t);
+                CONCURRENT_REFRESH_LEASES.release();
+                currentlyRefreshing.set(false);
+            }
+        }
+    }
+
+    @Override
+    public void close() {
+        executor.shutdown();
+    }
+}

src/main/java/com/aliyun/credentials/policy/OneCallerBlocks.java

@@ -0,0 +1,18 @@
+package com.aliyun.credentials.policy;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class OneCallerBlocks implements PrefetchStrategy {
+    private final AtomicBoolean currentlyRefreshing = new AtomicBoolean(false);
+
+    @Override
+    public void prefetch(Runnable valueUpdater) {
+        if (currentlyRefreshing.compareAndSet(false, true)) {
+            try {
+                valueUpdater.run();
+            } finally {
+                currentlyRefreshing.set(false);
+            }
+        }
+    }
+}

src/main/java/com/aliyun/credentials/policy/PrefetchStrategy.java

@@ -0,0 +1,9 @@
+package com.aliyun.credentials.policy;
+
+@FunctionalInterface
+public interface PrefetchStrategy extends AutoCloseable {
+    void prefetch(Runnable valueUpdater);
+
+    default void close() {
+    }
+}

src/main/java/com/aliyun/credentials/provider/CLIProfileCredentialsProvider.java

@@ -26,7 +26,7 @@ private CLIProfileCredentialsProvider(Builder builder) {
         this.currentProfileName = builder.profileName == null ? System.getenv("ALIBABA_CLOUD_PROFILE") : builder.profileName;
     }
 
-    static Builder builder() {
+    public static Builder builder() {
         return new Builder();
     }
 

src/main/java/com/aliyun/credentials/provider/DefaultCredentialsProvider.java

@@ -30,11 +30,7 @@ private void createDefaultChain() {
         defaultProviders.add(new SystemPropertiesCredentialsProvider());
         defaultProviders.add(new EnvironmentVariableCredentialsProvider());
         if (AuthUtils.environmentEnableOIDC()) {
-            defaultProviders.add(OIDCRoleArnCredentialProvider.builder()
-                    .roleArn(AuthUtils.getEnvironmentRoleArn())
-                    .oidcProviderArn(AuthUtils.getEnvironmentOIDCProviderArn())
-                    .oidcTokenFilePath(AuthUtils.getEnvironmentOIDCTokenFilePath())
-                    .build());
+            defaultProviders.add(OIDCRoleArnCredentialProvider.builder().build());
         }
         defaultProviders.add(CLIProfileCredentialsProvider.builder().build());
         defaultProviders.add(new ProfileCredentialsProvider());
@@ -99,7 +95,7 @@ public CredentialModel getCredentials() {
                 errorMessages.add(provider.getClass().getSimpleName() + ": " + e.getMessage());
             }
         }
-        throw new CredentialException("Unable to load credentials from any of the providers in the chain: ." + errorMessages);
+        throw new CredentialException("Unable to load credentials from any of the providers in the chain: " + errorMessages);
     }
 
     @Deprecated

src/main/java/com/aliyun/credentials/provider/ECSMetadataServiceCredentialsFetcher.java

@@ -14,12 +14,13 @@
 
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.util.Date;
 import java.util.Map;
 
 public class ECSMetadataServiceCredentialsFetcher {
     private static final String URL_IN_ECS_METADATA = "/latest/meta-data/ram/security-credentials/";
     private static final String URL_IN_METADATA_TOKEN = "/latest/api/token";
-    private static final String ECS_METADAT_FETCH_ERROR_MSG = "Failed to get RAM session credentials from ECS metadata service.";
+    private static final String ECS_METADATA_FETCH_ERROR_MSG = "Failed to get RAM session credentials from ECS metadata service.";
     private URL credentialUrl;
     private final String roleName;
     private final String metadataServiceHost = "100.100.100.200";
@@ -98,7 +99,7 @@ private String getMetadata(CompatibleUrlConnClient client, String url) {
         }
 
         if (response.getResponseCode() != 200) {
-            throw new CredentialException(ECS_METADAT_FETCH_ERROR_MSG + " HttpCode=" + response.getResponseCode());
+            throw new CredentialException(ECS_METADATA_FETCH_ERROR_MSG + " HttpCode=" + response.getResponseCode());
         }
 
         return new String(response.getHttpContent());
@@ -113,7 +114,10 @@ public RefreshResult<CredentialModel> fetch(CompatibleUrlConnClient client) {
         Map<String, String> result = new Gson().fromJson(jsonContent, Map.class);
 
         if (!"Success".equals(result.get("Code"))) {
-            throw new CredentialException(ECS_METADAT_FETCH_ERROR_MSG);
+            throw new CredentialException(ECS_METADATA_FETCH_ERROR_MSG);
+        }
+        if (!result.containsKey("AccessKeyId") || !result.containsKey("AccessKeySecret") || !result.containsKey("SecurityToken")) {
+            throw new CredentialException(String.format("Error retrieving credentials from IMDS result: %s.", jsonContent));
         }
         long expiration = ParameterHelper.getUTCDate(result.get("Expiration")).getTime();
         CredentialModel credential = CredentialModel.builder()
@@ -125,11 +129,24 @@ public RefreshResult<CredentialModel> fetch(CompatibleUrlConnClient client) {
                 .expiration(expiration)
                 .build();
         return RefreshResult.builder(credential)
-                .staleTime(expiration - 3 * 60 * 1000)
+                .staleTime(getStaleTime(expiration))
+                .prefetchTime(getPrefetchTime(expiration))
                 .build();
 
     }
 
+    private long getStaleTime(long expiration) {
+        return expiration <= 0 ?
+                new Date().getTime() + 60 * 60 * 1000
+                : expiration - 15 * 60 * 1000;
+    }
+
+    private long getPrefetchTime(long expiration) {
+        return expiration <= 0 ?
+                new Date().getTime() + 5 * 60 * 1000
+                : new Date().getTime() + 60 * 60 * 1000;
+    }
+
     public URL getCredentialUrl() {
         return credentialUrl;
     }

src/main/java/com/aliyun/credentials/provider/EcsRamRoleCredentialProvider.java

@@ -8,10 +8,21 @@
 import com.aliyun.credentials.utils.AuthUtils;
 import com.aliyun.credentials.utils.ProviderName;
 import com.aliyun.credentials.utils.StringUtils;
+import com.aliyun.tea.logging.ClientLogger;
 
-public class EcsRamRoleCredentialProvider extends SessionCredentialsProvider {
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import static com.aliyun.credentials.provider.RefreshCachedSupplier.StaleValueBehavior.ALLOW;
 
+public class EcsRamRoleCredentialProvider extends SessionCredentialsProvider {
+    private static final ClientLogger logger = new ClientLogger(EcsRamRoleCredentialProvider.class);
+    private static final int ASYNC_REFRESH_INTERVAL_TIME_MINUTES = 1;
     private ECSMetadataServiceCredentialsFetcher fetcher;
+    private volatile ScheduledExecutorService executor;
+    private volatile boolean shouldRefresh = false;
 
     @Deprecated
     public EcsRamRoleCredentialProvider(String roleName) {
@@ -22,6 +33,7 @@ public EcsRamRoleCredentialProvider(String roleName) {
             }
         }
         this.fetcher = new ECSMetadataServiceCredentialsFetcher(roleName);
+        checkCredentialsUpdateAsynchronously();
     }
 
     @Deprecated
@@ -34,6 +46,7 @@ public EcsRamRoleCredentialProvider(Configuration config) {
             }
         }
         this.fetcher = new ECSMetadataServiceCredentialsFetcher(config.getRoleName(), config.getConnectTimeout(), config.getReadTimeout());
+        checkCredentialsUpdateAsynchronously();
     }
 
     @Deprecated
@@ -56,6 +69,7 @@ public EcsRamRoleCredentialProvider(Config config) {
                 config.disableIMDSv1,
                 config.connectTimeout,
                 config.timeout);
+        checkCredentialsUpdateAsynchronously();
     }
 
     private EcsRamRoleCredentialProvider(BuilderImpl builder) {
@@ -70,6 +84,37 @@ private EcsRamRoleCredentialProvider(BuilderImpl builder) {
                 disableIMDSv1,
                 builder.connectionTimeout,
                 builder.readTimeout);
+        checkCredentialsUpdateAsynchronously();
+    }
+
+    private void checkCredentialsUpdateAsynchronously() {
+        if (isAsyncCredentialUpdateEnabled()) {
+            executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
+                public Thread newThread(Runnable r) {
+                    Thread t = Executors.defaultThreadFactory().newThread(r);
+                    t.setName("imds-credentials-check-and-refresh");
+                    t.setDaemon(true);
+                    return t;
+                }
+            });
+            executor.scheduleWithFixedDelay(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        if (shouldRefresh) {
+                            logger.info("Begin checking or refreshing credentials asynchronously");
+                            getCredentials();
+                        }
+                    } catch (Exception re) {
+                        handleAsyncRefreshError(re);
+                    }
+                }
+
+                private void handleAsyncRefreshError(Exception e) {
+                    logger.warning("Failed when checking or refreshing credentials asynchronously, error: {}.", e.getMessage());
+                }
+            }, 0, ASYNC_REFRESH_INTERVAL_TIME_MINUTES, TimeUnit.MINUTES);
+        }
     }
 
     public static Builder builder() {
@@ -79,7 +124,9 @@ public static Builder builder() {
     @Override
     public RefreshResult<CredentialModel> refreshCredentials() {
         try (CompatibleUrlConnClient client = new CompatibleUrlConnClient()) {
-            return fetcher.fetch(client);
+            RefreshResult<CredentialModel> result = fetcher.fetch(client);
+            shouldRefresh = true;
+            return result;
         }
     }
 
@@ -98,6 +145,11 @@ public String getProviderName() {
 
     @Override
     public void close() {
+        super.close();
+        if (executor != null) {
+            executor.shutdownNow();
+            executor = null;
+        }
     }
 
     public interface Builder extends SessionCredentialsProvider.Builder<EcsRamRoleCredentialProvider, Builder> {
@@ -129,6 +181,12 @@ private static final class BuilderImpl
         private Integer connectionTimeout;
         private Integer readTimeout;
 
+        private BuilderImpl() {
+            this.asyncCredentialUpdateEnabled = true;
+            this.jitterEnabled = true;
+            this.staleValueBehavior = ALLOW;
+        }
+
         public Builder roleName(String roleName) {
             this.roleName = roleName;
             return this;