[#5973] feat(hadoop-catalog): Support credential when using fileset catalog with cloud storage

api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java

@@ -74,6 +74,7 @@ public long expireTimeInMs() {
   public Map<String, String> credentialInfo() {
     return (new ImmutableMap.Builder<String, String>())
         .put(GRAVITINO_ADLS_SAS_TOKEN, sasToken)
+        .put(GRAVITINO_AZURE_STORAGE_ACCOUNT_NAME, accountName)
         .build();
   }
 

bundles/aliyun/src/main/java/org/apache/gravitino/oss/fs/OSSCredentialsProvider.java

@@ -0,0 +1,113 @@
+/*
+ *  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.gravitino.oss.fs;
+
+import com.aliyun.oss.common.auth.BasicCredentials;
+import com.aliyun.oss.common.auth.Credentials;
+import com.aliyun.oss.common.auth.CredentialsProvider;
+import com.aliyun.oss.common.auth.DefaultCredentials;
+import java.net.URI;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
+import org.apache.gravitino.catalog.hadoop.fs.GravitinoFileSystemCredentialsProvider;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.OSSSecretKeyCredential;
+import org.apache.gravitino.credential.OSSTokenCredential;
+import org.apache.hadoop.conf.Configuration;
+
+public class OSSCredentialsProvider implements CredentialsProvider {
+
+  private GravitinoFileSystemCredentialsProvider gravitinoFileSystemCredentialsProvider;
+  private Credentials basicCredentials;
+  private long expirationTime = Long.MAX_VALUE;
+  private static final double EXPIRATION_TIME_FACTOR = 0.9D;
+
+  public OSSCredentialsProvider(URI uri, Configuration conf) {
+    this.gravitinoFileSystemCredentialsProvider = FileSystemUtils.getGvfsCredentialProvider(conf);
+  }
+
+  @Override
+  public void setCredentials(Credentials credentials) {}
+
+  @Override
+  public Credentials getCredentials() {
+    if (basicCredentials == null || System.currentTimeMillis() >= expirationTime) {
+      synchronized (this) {
+        refresh();
+      }
+    }
+
+    return basicCredentials;
+  }
+
+  private void refresh() {
+    Credential[] gravitinoCredentials = gravitinoFileSystemCredentialsProvider.getCredentials();
+    Credential credential = getSuitableCredential(gravitinoCredentials);
+    if (credential == null) {
+      throw new RuntimeException("No suitable credential for OSS found...");
+    }
+
+    if (credential instanceof OSSSecretKeyCredential) {
+      OSSSecretKeyCredential ossSecretKeyCredential = (OSSSecretKeyCredential) credential;
+      basicCredentials =
+          new DefaultCredentials(
+              ossSecretKeyCredential.accessKeyId(), ossSecretKeyCredential.secretAccessKey());
+    } else if (credential instanceof OSSTokenCredential) {
+      OSSTokenCredential ossTokenCredential = (OSSTokenCredential) credential;
+      basicCredentials =
+          new BasicCredentials(
+              ossTokenCredential.accessKeyId(),
+              ossTokenCredential.secretAccessKey(),
+              ossTokenCredential.securityToken());
+    }
+
+    if (credential.expireTimeInMs() > 0) {
+      expirationTime =
+          System.currentTimeMillis()
+              + (long)
+                  ((credential.expireTimeInMs() - System.currentTimeMillis())
+                      * EXPIRATION_TIME_FACTOR);
+    }
+  }
+
+  /**
+   * Get the credential from the credential array. Using dynamic credential first, if not found,
+   * uses static credential.
+   *
+   * @param credentials The credential array.
+   * @return A credential. Null if not found.
+   */
+  private Credential getSuitableCredential(Credential[] credentials) {
+    // Use dynamic credential if found.
+    for (Credential credential : credentials) {
+      if (credential instanceof OSSTokenCredential) {
+        return credential;
+      }
+    }
+
+    // If dynamic credential not found, use the static one
+    for (Credential credential : credentials) {
+      if (credential instanceof OSSSecretKeyCredential) {
+        return credential;
+      }
+    }
+
+    return null;
+  }
+}

bundles/aliyun/src/main/java/org/apache/gravitino/oss/fs/OSSFileSystemProvider.java

@@ -18,6 +18,8 @@
  */
 package org.apache.gravitino.oss.fs;
 
+import static org.apache.gravitino.catalog.hadoop.fs.CredentialUtils.enableGravitinoCredentialVending;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
@@ -60,7 +62,13 @@ public FileSystem getFileSystem(Path path, Map<String, String> config) throws IO
       hadoopConfMap.put(OSS_FILESYSTEM_IMPL, AliyunOSSFileSystem.class.getCanonicalName());
     }
 
+    if (enableGravitinoCredentialVending(config)) {
+      hadoopConfMap.put(
+          Constants.CREDENTIALS_PROVIDER_KEY, OSSCredentialsProvider.class.getCanonicalName());
+    }
+
     hadoopConfMap.forEach(configuration::set);
+
     return AliyunOSSFileSystem.newInstance(path.toUri(), configuration);
   }
 

bundles/aws/src/main/java/org/apache/gravitino/s3/fs/S3CredentialsProvider.java

@@ -0,0 +1,112 @@
+/*
+ *  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.gravitino.s3.fs;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.auth.BasicSessionCredentials;
+import java.net.URI;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
+import org.apache.gravitino.catalog.hadoop.fs.GravitinoFileSystemCredentialsProvider;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.S3SecretKeyCredential;
+import org.apache.gravitino.credential.S3TokenCredential;
+import org.apache.hadoop.conf.Configuration;
+
+public class S3CredentialsProvider implements AWSCredentialsProvider {
+  private GravitinoFileSystemCredentialsProvider gravitinoFileSystemCredentialsProvider;
+
+  private AWSCredentials basicSessionCredentials;
+  private long expirationTime = Long.MAX_VALUE;
+  private static final double EXPIRATION_TIME_FACTOR = 0.9D;
+
+  public S3CredentialsProvider(final URI uri, final Configuration conf) {
+    this.gravitinoFileSystemCredentialsProvider = FileSystemUtils.getGvfsCredentialProvider(conf);
+  }
+
+  @Override
+  public AWSCredentials getCredentials() {
+    // Refresh credentials if they are null or about to expire.
+    if (basicSessionCredentials == null || System.currentTimeMillis() >= expirationTime) {
+      synchronized (this) {
+        refresh();
+      }
+    }
+
+    return basicSessionCredentials;
+  }
+
+  @Override
+  public void refresh() {
+    Credential[] gravitinoCredentials = gravitinoFileSystemCredentialsProvider.getCredentials();
+    Credential credential = getSuitableCredential(gravitinoCredentials);
+
+    if (credential == null) {
+      throw new RuntimeException("No suitable credential for S3 found...");
+    }
+
+    if (credential instanceof S3SecretKeyCredential) {
+      S3SecretKeyCredential s3SecretKeyCredential = (S3SecretKeyCredential) credential;
+      basicSessionCredentials =
+          new BasicAWSCredentials(
+              s3SecretKeyCredential.accessKeyId(), s3SecretKeyCredential.secretAccessKey());
+    } else if (credential instanceof S3TokenCredential) {
+      S3TokenCredential s3TokenCredential = (S3TokenCredential) credential;
+      basicSessionCredentials =
+          new BasicSessionCredentials(
+              s3TokenCredential.accessKeyId(),
+              s3TokenCredential.secretAccessKey(),
+              s3TokenCredential.sessionToken());
+    }
+
+    if (credential.expireTimeInMs() > 0) {
+      expirationTime =
+          System.currentTimeMillis()
+              + (long)
+                  ((credential.expireTimeInMs() - System.currentTimeMillis())
+                      * EXPIRATION_TIME_FACTOR);
+    }
+  }
+
+  /**
+   * Get the credential from the credential array. Using dynamic credential first, if not found,
+   * uses static credential.
+   *
+   * @param credentials The credential array.
+   * @return A credential. Null if not found.
+   */
+  private Credential getSuitableCredential(Credential[] credentials) {
+    // Use dynamic credential if found.
+    for (Credential credential : credentials) {
+      if (credential instanceof S3TokenCredential) {
+        return credential;
+      }
+    }
+
+    // If dynamic credential not found, use the static one
+    for (Credential credential : credentials) {
+      if (credential instanceof S3SecretKeyCredential) {
+        return credential;
+      }
+    }
+    return null;
+  }
+}

bundles/aws/src/main/java/org/apache/gravitino/s3/fs/S3FileSystemProvider.java

@@ -19,6 +19,8 @@
 
 package org.apache.gravitino.s3.fs;
 
+import static org.apache.gravitino.catalog.hadoop.fs.CredentialUtils.enableGravitinoCredentialVending;
+
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
@@ -41,7 +43,7 @@
 
 public class S3FileSystemProvider implements FileSystemProvider {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(S3FileSystemProvider.class);
+  private static final Logger LOG = LoggerFactory.getLogger(S3FileSystemProvider.class);
 
   @VisibleForTesting
   public static final Map<String, String> GRAVITINO_KEY_TO_S3_HADOOP_KEY =
@@ -61,11 +63,15 @@ public FileSystem getFileSystem(Path path, Map<String, String> config) throws IO
     Map<String, String> hadoopConfMap =
         FileSystemUtils.toHadoopConfigMap(config, GRAVITINO_KEY_TO_S3_HADOOP_KEY);
 
+    hadoopConfMap.forEach(configuration::set);
     if (!hadoopConfMap.containsKey(S3_CREDENTIAL_KEY)) {
-      hadoopConfMap.put(S3_CREDENTIAL_KEY, S3_SIMPLE_CREDENTIAL);
+      configuration.set(S3_CREDENTIAL_KEY, S3_SIMPLE_CREDENTIAL);
     }
 
-    hadoopConfMap.forEach(configuration::set);
+    if (enableGravitinoCredentialVending(config)) {
+      configuration.set(
+          Constants.AWS_CREDENTIALS_PROVIDER, S3CredentialsProvider.class.getCanonicalName());
+    }
 
     // Hadoop-aws 2 does not support IAMInstanceCredentialsProvider
     checkAndSetCredentialProvider(configuration);
@@ -91,12 +97,12 @@ private void checkAndSetCredentialProvider(Configuration configuration) {
         if (AWSCredentialsProvider.class.isAssignableFrom(c)) {
           validProviders.add(provider);
         } else {
-          LOGGER.warn(
+          LOG.warn(
               "Credential provider {} is not a subclass of AWSCredentialsProvider, skipping",
               provider);
         }
       } catch (Exception e) {
-        LOGGER.warn(
+        LOG.warn(
             "Credential provider {} not found in the Hadoop runtime, falling back to default",
             provider);
         configuration.set(S3_CREDENTIAL_KEY, S3_SIMPLE_CREDENTIAL);

bundles/azure/build.gradle.kts

@@ -28,7 +28,6 @@ dependencies {
   compileOnly(project(":api"))
   compileOnly(project(":catalogs:catalog-hadoop"))
   compileOnly(project(":core"))
-
   compileOnly(libs.hadoop3.abs)
   compileOnly(libs.hadoop3.client.api)
   compileOnly(libs.hadoop3.client.runtime)

bundles/azure/src/main/java/org/apache/gravitino/abs/fs/AzureFileSystemProvider.java

@@ -19,6 +19,11 @@
 
 package org.apache.gravitino.abs.fs;
 
+import static org.apache.gravitino.catalog.hadoop.fs.CredentialUtils.enableGravitinoCredentialVending;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_AUTH_TYPE_PROPERTY_NAME;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_IS_HNS_ENABLED;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_SAS_TOKEN_PROVIDER_TYPE;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
@@ -30,6 +35,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.services.AuthType;
 
 public class AzureFileSystemProvider implements FileSystemProvider {
 
@@ -58,12 +64,42 @@ public FileSystem getFileSystem(@Nonnull Path path, @Nonnull Map<String, String>
           config.get(AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_KEY));
     }
 
-    if (!config.containsKey(ABFS_IMPL_KEY)) {
+    if (!hadoopConfMap.containsKey(ABFS_IMPL_KEY)) {
       configuration.set(ABFS_IMPL_KEY, ABFS_IMPL);
     }
 
     hadoopConfMap.forEach(configuration::set);
 
+    if (enableGravitinoCredentialVending(hadoopConfMap)) {
+      try {
+        AzureSasCredentialsProvider azureSasCredentialsProvider = new AzureSasCredentialsProvider();
+        azureSasCredentialsProvider.initialize(configuration, null);
+        String sas = azureSasCredentialsProvider.getSASToken(null, null, null, null);
+        if (sas != null) {
+          String accountName =
+              String.format(
+                  "%s.dfs.core.windows.net",
+                  config.get(AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_NAME));
+
+          configuration.set(
+              FS_AZURE_ACCOUNT_AUTH_TYPE_PROPERTY_NAME + "." + accountName, AuthType.SAS.name());
+          configuration.set(
+              FS_AZURE_SAS_TOKEN_PROVIDER_TYPE + "." + accountName,
+              AzureSasCredentialsProvider.class.getName());
+          configuration.set(FS_AZURE_ACCOUNT_IS_HNS_ENABLED, "true");
+        } else if (azureSasCredentialsProvider.getAzureStorageAccountKey() != null
+            && azureSasCredentialsProvider.getAzureStorageAccountName() != null) {
+          configuration.set(
+              String.format(
+                  "fs.azure.account.key.%s.dfs.core.windows.net",
+                  azureSasCredentialsProvider.getAzureStorageAccountName()),
+              azureSasCredentialsProvider.getAzureStorageAccountKey());
+        }
+      } catch (Exception e) {
+        throw new IOException("Failed to get SAS token from AzureSasCredentialsProvider", e);
+      }
+    }
+
     return FileSystem.get(path.toUri(), configuration);
   }