[core] RESTCatalog: support DLFECSTokenLoader (#5312)

Author: shyjsarah

docs/content/concepts/rest-catalog.md

@@ -109,8 +109,23 @@ WITH (
 );
 ```
 
+- DLF sts token using aliyun ecs role
+```sql
+CREATE CATALOG `paimon-rest-catalog`
+WITH (
+    'type' = 'paimon',
+    'uri' = '<catalog server url>',
+    'metastore' = 'rest',
+    'warehouse' = 'my_instance_name',
+    'token.provider' = 'dlf',
+    'dlf.token-loader' = 'ecs'
+    'dlf.token-ecs-role-name' = 'my_ecs_role_name'
+);
+```
+
 {{< hint info >}}
 The `'warehouse'` is your catalog instance name on the server, not the path.
+The `'dlf.token-ecs-role-name'` is an optional parameter，dlf token loader can automatically obtain it through ecs metadata service. More information about ecs role can be found [here](https://help.aliyun.com/zh/ecs/user-guide/attach-an-instance-ram-role-to-an-ecs-instance?spm=a2c4g.11186623.0.0.7e774adcRAJ0wK).
 {{< /hint >}}
 
 ## Conclusion

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogOptions.java

@@ -85,4 +85,17 @@ public class RESTCatalogOptions {
                     .stringType()
                     .noDefaultValue()
                     .withDescription("REST Catalog auth DLF token loader.");
+
+    public static final ConfigOption<String> DLF_TOKEN_ECS_METADATA_URL =
+            ConfigOptions.key("dlf.token-ecs-metadata-url")
+                    .stringType()
+                    .defaultValue(
+                            "http://100.100.100.200/latest/meta-data/Ram/security-credentials/")
+                    .withDescription("REST Catalog auth DLF token ecs metadata url.");
+
+    public static final ConfigOption<String> DLF_TOKEN_ECS_ROLE_NAME =
+            ConfigOptions.key("dlf.token-ecs-role-name")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("REST Catalog auth DLF token ecs role name.");
 }

paimon-core/src/main/java/org/apache/paimon/rest/auth/DLFECSTokenLoader.java

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.auth;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.rest.ExponentialHttpRetryInterceptor;
+
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.time.Duration;
+import java.util.Arrays;
+
+import static okhttp3.ConnectionSpec.CLEARTEXT;
+import static okhttp3.ConnectionSpec.COMPATIBLE_TLS;
+import static okhttp3.ConnectionSpec.MODERN_TLS;
+import static org.apache.paimon.rest.RESTObjectMapper.OBJECT_MAPPER;
+
+/** DLF Token Loader for ECS Metadata Service. */
+public class DLFECSTokenLoader implements DLFTokenLoader {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DLFECSTokenLoader.class);
+
+    private static final OkHttpClient HTTP_CLIENT =
+            new OkHttpClient.Builder()
+                    .retryOnConnectionFailure(true)
+                    .connectionSpecs(Arrays.asList(MODERN_TLS, COMPATIBLE_TLS, CLEARTEXT))
+                    .addInterceptor(new ExponentialHttpRetryInterceptor(3))
+                    .connectTimeout(Duration.ofMinutes(3))
+                    .readTimeout(Duration.ofMinutes(3))
+                    .build();
+
+    private String ecsMetadataURL;
+
+    private String roleName;
+
+    public DLFECSTokenLoader(String ecsMetaDataURL, @Nullable String roleName) {
+        this.ecsMetadataURL = ecsMetaDataURL;
+        this.roleName = roleName;
+    }
+
+    @Override
+    public DLFToken loadToken() {
+        if (roleName == null) {
+            roleName = getRole(ecsMetadataURL);
+        }
+        return getToken(ecsMetadataURL + roleName);
+    }
+
+    private static String getRole(String url) {
+        try {
+            return getResponseBody(url);
+        } catch (Exception e) {
+            throw new RuntimeException("get role failed, error : " + e.getMessage(), e);
+        }
+    }
+
+    private static DLFToken getToken(String url) {
+        try {
+            String token = getResponseBody(url);
+            return OBJECT_MAPPER.readValue(token, DLFToken.class);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        } catch (Exception e) {
+            throw new RuntimeException("get token failed, error : " + e.getMessage(), e);
+        }
+    }
+
+    @VisibleForTesting
+    protected static String getResponseBody(String url) {
+        Request request = new Request.Builder().url(url).get().build();
+        long startTime = System.currentTimeMillis();
+        try (Response response = HTTP_CLIENT.newCall(request).execute()) {
+            if (response == null) {
+                throw new RuntimeException("get response failed, response is null");
+            }
+            if (!response.isSuccessful()) {
+                throw new RuntimeException("get response failed, response : " + response);
+            }
+            String responseBodyStr = response.body() != null ? response.body().string() : null;
+            if (responseBodyStr == null) {
+                throw new RuntimeException("get response failed, response body is null");
+            }
+            if (LOG.isDebugEnabled()) {
+                LOG.debug(
+                        "get response success, url : {}, cost : {} ms",
+                        url,
+                        System.currentTimeMillis() - startTime);
+            }
+            return responseBodyStr;
+        } catch (RuntimeException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new RuntimeException("get response failed, error : " + e.getMessage(), e);
+        }
+    }
+}

paimon-core/src/main/java/org/apache/paimon/rest/auth/DLFECSTokenLoaderFactory.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.auth;
+
+import org.apache.paimon.options.Options;
+
+import static org.apache.paimon.rest.RESTCatalogOptions.DLF_TOKEN_ECS_METADATA_URL;
+import static org.apache.paimon.rest.RESTCatalogOptions.DLF_TOKEN_ECS_ROLE_NAME;
+
+/** Factory for {@link DLFECSTokenLoader}. */
+public class DLFECSTokenLoaderFactory implements DLFTokenLoaderFactory {
+
+    @Override
+    public String identifier() {
+        return "ecs";
+    }
+
+    @Override
+    public DLFTokenLoader create(Options options) {
+        String ecsMetadataURL = options.get(DLF_TOKEN_ECS_METADATA_URL);
+        String roleName = options.get(DLF_TOKEN_ECS_ROLE_NAME);
+        return new DLFECSTokenLoader(ecsMetadataURL, roleName);
+    }
+}

paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory

@@ -41,3 +41,4 @@ org.apache.paimon.iceberg.migrate.IcebergMigrateHadoopMetadataFactory
 org.apache.paimon.rest.auth.BearTokenAuthProviderFactory
 org.apache.paimon.rest.auth.DLFAuthProviderFactory
 org.apache.paimon.rest.auth.DLFLocalFileTokenLoaderFactory
+org.apache.paimon.rest.auth.DLFECSTokenLoaderFactory

paimon-core/src/test/java/org/apache/paimon/rest/auth/AuthSessionTest.java

@@ -46,6 +46,8 @@
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_ACCESS_KEY_SECRET;
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_REGION;
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_SECURITY_TOKEN;
+import static org.apache.paimon.rest.RESTCatalogOptions.DLF_TOKEN_ECS_METADATA_URL;
+import static org.apache.paimon.rest.RESTCatalogOptions.DLF_TOKEN_ECS_ROLE_NAME;
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_TOKEN_LOADER;
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_TOKEN_PATH;
 import static org.apache.paimon.rest.RESTCatalogOptions.TOKEN;
@@ -291,6 +293,124 @@ public void testCreateDlfAuthProviderByCustomDLFTokenLoader()
         Assert.assertEquals(fetchToken.getSecurityToken(), customToken.getSecurityToken());
     }
 
+    @Test
+    public void testCreateDlfAuthProviderByECSTokenProvider()
+            throws IOException, InterruptedException {
+        MockECSMetadataService mockECSMetadataService = new MockECSMetadataService("EcsTestRole");
+        mockECSMetadataService.start();
+        try {
+            DLFToken theFirstMockToken = generateToken();
+            mockECSMetadataService.setMockToken(theFirstMockToken);
+            String theFirstMockTokenStr =
+                    OBJECT_MAPPER_INSTANCE.writeValueAsString(theFirstMockToken);
+            long tokenRefreshInMills = 1000;
+            // create options with token loader
+            Options options = new Options();
+            options.set(DLF_TOKEN_LOADER.key(), "ecs");
+            options.set(
+                    DLF_TOKEN_ECS_METADATA_URL.key(),
+                    mockECSMetadataService.getUrl() + "latest/meta-data/Ram/security-credentials/");
+            options.set(RESTCatalogOptions.URI.key(), "serverUrl");
+            options.set(DLF_REGION.key(), "cn-hangzhou");
+            options.set(TOKEN_REFRESH_TIME.key(), tokenRefreshInMills + "ms");
+            AuthProvider authProvider =
+                    AuthProviderFactory.createAuthProvider(
+                            AuthProviderEnum.DLF.identifier(), options);
+            ScheduledExecutorService executor =
+                    ThreadPoolUtils.createScheduledThreadPool(1, "refresh-token");
+            AuthSession session = AuthSession.fromRefreshAuthProvider(executor, authProvider);
+            DLFAuthProvider dlfAuthProvider = (DLFAuthProvider) session.getAuthProvider();
+            String theFirstFetchTokenStr =
+                    OBJECT_MAPPER_INSTANCE.writeValueAsString(dlfAuthProvider.token);
+            assertEquals(theFirstFetchTokenStr, theFirstMockTokenStr);
+
+            DLFToken theSecondMockToken = generateToken();
+            String theSecondMockTokenStr =
+                    OBJECT_MAPPER_INSTANCE.writeValueAsString(theSecondMockToken);
+            mockECSMetadataService.setMockToken(theSecondMockToken);
+            Thread.sleep(tokenRefreshInMills * 2);
+            String theSecondFetchTokenStr =
+                    OBJECT_MAPPER_INSTANCE.writeValueAsString(dlfAuthProvider.token);
+            assertEquals(theSecondFetchTokenStr, theSecondMockTokenStr);
+        } finally {
+            mockECSMetadataService.shutdown();
+        }
+    }
+
+    @Test
+    public void testCreateDlfAuthProviderByECSTokenProviderWithDefineRole()
+            throws IOException, InterruptedException {
+        MockECSMetadataService mockECSMetadataService = new MockECSMetadataService("CustomRole");
+        mockECSMetadataService.start();
+        try {
+            DLFToken theFirstMockToken = generateToken();
+            mockECSMetadataService.setMockToken(theFirstMockToken);
+            String theFirstMockTokenStr =
+                    OBJECT_MAPPER_INSTANCE.writeValueAsString(theFirstMockToken);
+            long tokenRefreshInMills = 1000;
+            // create options with token loader
+            Options options = new Options();
+            options.set(DLF_TOKEN_LOADER.key(), "ecs");
+            options.set(
+                    DLF_TOKEN_ECS_METADATA_URL.key(),
+                    mockECSMetadataService.getUrl() + "latest/meta-data/Ram/security-credentials/");
+            options.set(DLF_TOKEN_ECS_ROLE_NAME.key(), "CustomRole");
+            options.set(RESTCatalogOptions.URI.key(), "serverUrl");
+            options.set(DLF_REGION.key(), "cn-hangzhou");
+            options.set(TOKEN_REFRESH_TIME.key(), tokenRefreshInMills + "ms");
+            AuthProvider authProvider =
+                    AuthProviderFactory.createAuthProvider(
+                            AuthProviderEnum.DLF.identifier(), options);
+            ScheduledExecutorService executor =
+                    ThreadPoolUtils.createScheduledThreadPool(1, "refresh-token");
+            AuthSession session = AuthSession.fromRefreshAuthProvider(executor, authProvider);
+            DLFAuthProvider dlfAuthProvider = (DLFAuthProvider) session.getAuthProvider();
+            String theFirstFetchTokenStr =
+                    OBJECT_MAPPER_INSTANCE.writeValueAsString(dlfAuthProvider.token);
+            assertEquals(theFirstFetchTokenStr, theFirstMockTokenStr);
+
+            DLFToken theSecondMockToken = generateToken();
+            String theSecondMockTokenStr =
+                    OBJECT_MAPPER_INSTANCE.writeValueAsString(theSecondMockToken);
+            mockECSMetadataService.setMockToken(theSecondMockToken);
+            Thread.sleep(tokenRefreshInMills * 2);
+            String theSecondFetchTokenStr =
+                    OBJECT_MAPPER_INSTANCE.writeValueAsString(dlfAuthProvider.token);
+            assertEquals(theSecondFetchTokenStr, theSecondMockTokenStr);
+        } finally {
+            mockECSMetadataService.shutdown();
+        }
+    }
+
+    @Test
+    public void testCreateDlfAuthProviderByECSTokenProviderWithInvalidRole()
+            throws IOException, InterruptedException {
+        MockECSMetadataService mockECSMetadataService = new MockECSMetadataService("EcsTestRole");
+        mockECSMetadataService.start();
+        try {
+            DLFToken theFirstMockToken = generateToken();
+            mockECSMetadataService.setMockToken(theFirstMockToken);
+            // create options with token loader
+            Options options = new Options();
+            options.set(DLF_TOKEN_LOADER.key(), "ecs");
+            options.set(
+                    DLF_TOKEN_ECS_METADATA_URL.key(),
+                    mockECSMetadataService.getUrl() + "latest/meta-data/Ram/security-credentials/");
+            options.set(DLF_TOKEN_ECS_ROLE_NAME.key(), "CustomRole");
+            options.set(RESTCatalogOptions.URI.key(), "serverUrl");
+            options.set(DLF_REGION.key(), "cn-hangzhou");
+            assertThrows(
+                    RuntimeException.class,
+                    () -> {
+                        AuthProvider authProvider =
+                                AuthProviderFactory.createAuthProvider(
+                                        AuthProviderEnum.DLF.identifier(), options);
+                    });
+        } finally {
+            mockECSMetadataService.shutdown();
+        }
+    }
+
     @Test
     public void testDLFAuthProviderAuthHeaderWhenDataIsNotEmpty() throws Exception {
         String fileName = UUID.randomUUID().toString();