[#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/build.gradle.kts

@@ -39,6 +39,8 @@ dependencies {
   implementation(project(":catalogs:hadoop-common")) {
     exclude("*")
   }
+  implementation(project(":clients:client-java-runtime", configuration = "shadow"))
+  implementation(project(":clients:filesystem-hadoop3-runtime", configuration = "shadow"))
 
   implementation(libs.aliyun.credentials.sdk)
   implementation(libs.commons.collections3)

bundles/aliyun/src/main/java/org/apache/gravitino/oss/credential/OSSTokenProvider.java

@@ -156,7 +156,7 @@ private String createPolicy(Set<String> readLocations, Set<String> writeLocation
                   key ->
                       Statement.builder()
                           .effect(Effect.ALLOW)
-                          .addAction("oss:ListBucket")
+                          .addAction("oss:ListObjects")
                           .addResource(key)
                           .condition(getCondition(uri)));
               // GetBucketLocation

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

@@ -24,6 +24,7 @@
 import java.util.Map;
 import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
 import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
+import org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration;
 import org.apache.gravitino.storage.OSSProperties;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -60,7 +61,16 @@ public FileSystem getFileSystem(Path path, Map<String, String> config) throws IO
       hadoopConfMap.put(OSS_FILESYSTEM_IMPL, AliyunOSSFileSystem.class.getCanonicalName());
     }
 
+    if (!hadoopConfMap.containsKey(Constants.CREDENTIALS_PROVIDER_KEY)
+        && config.containsKey(
+            GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_SERVER_URI_KEY)) {
+      hadoopConfMap.put(
+          Constants.CREDENTIALS_PROVIDER_KEY,
+          OSSSessionCredentialProvider.class.getCanonicalName());
+    }
+
     hadoopConfMap.forEach(configuration::set);
+
     return AliyunOSSFileSystem.newInstance(path.toUri(), configuration);
   }
 

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

@@ -0,0 +1,116 @@
+/*
+ *  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 static org.apache.gravitino.credential.OSSTokenCredential.GRAVITINO_OSS_SESSION_ACCESS_KEY_ID;
+import static org.apache.gravitino.credential.OSSTokenCredential.GRAVITINO_OSS_SESSION_SECRET_ACCESS_KEY;
+import static org.apache.gravitino.credential.OSSTokenCredential.GRAVITINO_OSS_TOKEN;
+
+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 java.util.Map;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.OSSTokenCredential;
+import org.apache.gravitino.file.Fileset;
+import org.apache.gravitino.file.FilesetCatalog;
+import org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystem;
+import org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.aliyun.oss.Constants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class OSSSessionCredentialProvider implements CredentialsProvider {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(OSSSessionCredentialProvider.class);
+  private Credentials basicCredentials;
+  private final String filesetIdentifier;
+  private long expirationTime;
+  private final GravitinoClient client;
+  private final Configuration configuration;
+
+  public OSSSessionCredentialProvider(URI uri, Configuration conf) {
+    this.filesetIdentifier =
+        conf.get(GravitinoVirtualFileSystemConfiguration.GVFS_FILESET_IDENTIFIER);
+    // extra value and init Gravitino client here
+    GravitinoVirtualFileSystem gravitinoVirtualFileSystem = new GravitinoVirtualFileSystem();
+    this.client = gravitinoVirtualFileSystem.initializeClient(conf);
+    this.configuration = conf;
+  }
+
+  @Override
+  public void setCredentials(Credentials credentials) {}
+
+  @Override
+  public Credentials getCredentials() {
+    // If the credentials are null or about to expire, refresh the credentials.
+    if (basicCredentials == null || System.currentTimeMillis() > expirationTime - 5 * 60 * 1000) {
+      synchronized (this) {
+        refresh();
+      }
+    }
+
+    return basicCredentials;
+  }
+
+  private void refresh() {
+    String[] idents = filesetIdentifier.split("\\.");
+    String catalog = idents[1];
+
+    FilesetCatalog filesetCatalog = client.loadCatalog(catalog).asFilesetCatalog();
+
+    Fileset fileset = filesetCatalog.loadFileset(NameIdentifier.of(idents[2], idents[3]));
+    Credential[] credentials = fileset.supportsCredentials().getCredentials();
+    if (credentials.length == 0) {
+      LOGGER.warn("No credential found for fileset: {}, try to use static AKSK", filesetIdentifier);
+      expirationTime = Long.MAX_VALUE;
+      this.basicCredentials =
+          new DefaultCredentials(
+              configuration.get(Constants.ACCESS_KEY_ID),
+              configuration.get(Constants.ACCESS_KEY_SECRET));
+      return;
+    }
+
+    // Use the first one.
+    Credential credential = credentials[0];
+    Map<String, String> credentialMap = credential.toProperties();
+
+    String accessKeyId = credentialMap.get(GRAVITINO_OSS_SESSION_ACCESS_KEY_ID);
+    String secretAccessKey = credentialMap.get(GRAVITINO_OSS_SESSION_SECRET_ACCESS_KEY);
+
+    if (OSSTokenCredential.OSS_TOKEN_CREDENTIAL_TYPE.equals(
+        credentialMap.get(Credential.CREDENTIAL_TYPE))) {
+      String sessionToken = credentialMap.get(GRAVITINO_OSS_TOKEN);
+      this.basicCredentials = new BasicCredentials(accessKeyId, secretAccessKey, sessionToken);
+    } else {
+      this.basicCredentials = new DefaultCredentials(accessKeyId, secretAccessKey);
+    }
+
+    this.expirationTime = credential.expireTimeInMs();
+    if (expirationTime <= 0) {
+      expirationTime = Long.MAX_VALUE;
+    }
+  }
+}

bundles/aws/build.gradle.kts

@@ -39,6 +39,8 @@ dependencies {
   implementation(project(":catalogs:hadoop-common")) {
     exclude("*")
   }
+  implementation(project(":clients:client-java-runtime", configuration = "shadow"))
+  implementation(project(":clients:filesystem-hadoop3-runtime", configuration = "shadow"))
 
   implementation(libs.aws.iam)
   implementation(libs.aws.policy)

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

@@ -30,6 +30,7 @@
 import java.util.Map;
 import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
 import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
+import org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration;
 import org.apache.gravitino.storage.S3Properties;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -61,11 +62,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 (config.containsKey(GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_SERVER_URI_KEY)) {
+      configuration.set(
+          Constants.AWS_CREDENTIALS_PROVIDER, S3SessionCredentialProvider.class.getCanonicalName());
+    }
 
     // Hadoop-aws 2 does not support IAMInstanceCredentialsProvider
     checkAndSetCredentialProvider(configuration);

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

@@ -0,0 +1,119 @@
+/*
+ *  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 static org.apache.gravitino.credential.S3TokenCredential.GRAVITINO_S3_SESSION_ACCESS_KEY_ID;
+import static org.apache.gravitino.credential.S3TokenCredential.GRAVITINO_S3_SESSION_SECRET_ACCESS_KEY;
+import static org.apache.gravitino.credential.S3TokenCredential.GRAVITINO_S3_TOKEN;
+
+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 java.util.Map;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.S3TokenCredential;
+import org.apache.gravitino.file.Fileset;
+import org.apache.gravitino.file.FilesetCatalog;
+import org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystem;
+import org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.Constants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class S3SessionCredentialProvider implements AWSCredentialsProvider {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(S3SessionCredentialProvider.class);
+  private final GravitinoClient client;
+  private final String filesetIdentifier;
+  private final Configuration configuration;
+
+  private AWSCredentials basicSessionCredentials;
+  private long expirationTime;
+
+  public S3SessionCredentialProvider(final URI uri, final Configuration conf) {
+    this.filesetIdentifier =
+        conf.get(GravitinoVirtualFileSystemConfiguration.GVFS_FILESET_IDENTIFIER);
+    this.configuration = conf;
+
+    // extra value and init Gravitino client here
+    GravitinoVirtualFileSystem gravitinoVirtualFileSystem = new GravitinoVirtualFileSystem();
+    this.client = gravitinoVirtualFileSystem.initializeClient(conf);
+  }
+
+  @Override
+  public AWSCredentials getCredentials() {
+    // Refresh credentials if they are null or about to expire in 5 minutes
+    if (basicSessionCredentials == null
+        || System.currentTimeMillis() > expirationTime - 5 * 60 * 1000) {
+      synchronized (this) {
+        refresh();
+      }
+    }
+
+    return basicSessionCredentials;
+  }
+
+  @Override
+  public void refresh() {
+    // The format of filesetIdentifier is "metalake.catalog.fileset.schema"
+    String[] idents = filesetIdentifier.split("\\.");
+    String catalog = idents[1];
+
+    FilesetCatalog filesetCatalog = client.loadCatalog(catalog).asFilesetCatalog();
+
+    Fileset fileset = filesetCatalog.loadFileset(NameIdentifier.of(idents[2], idents[3]));
+    Credential[] credentials = fileset.supportsCredentials().getCredentials();
+
+    // Can't find any credential, use the default one.
+    if (credentials.length == 0) {
+      LOGGER.warn("No credential found for fileset: {}, try to use static AKSK", filesetIdentifier);
+      expirationTime = Long.MAX_VALUE;
+      this.basicSessionCredentials =
+          new BasicAWSCredentials(
+              configuration.get(Constants.ACCESS_KEY), configuration.get(Constants.SECRET_KEY));
+      return;
+    }
+
+    Credential credential = credentials[0];
+    Map<String, String> credentialMap = credential.toProperties();
+
+    String accessKeyId = credentialMap.get(GRAVITINO_S3_SESSION_ACCESS_KEY_ID);
+    String secretAccessKey = credentialMap.get(GRAVITINO_S3_SESSION_SECRET_ACCESS_KEY);
+
+    if (S3TokenCredential.S3_TOKEN_CREDENTIAL_TYPE.equals(
+        credentialMap.get(Credential.CREDENTIAL_TYPE))) {
+      String sessionToken = credentialMap.get(GRAVITINO_S3_TOKEN);
+      this.basicSessionCredentials =
+          new BasicSessionCredentials(accessKeyId, secretAccessKey, sessionToken);
+    } else {
+      this.basicSessionCredentials = new BasicAWSCredentials(accessKeyId, secretAccessKey);
+    }
+
+    this.expirationTime = credential.expireTimeInMs();
+    if (expirationTime <= 0) {
+      expirationTime = Long.MAX_VALUE;
+    }
+  }
+}

bundles/azure/build.gradle.kts

@@ -39,6 +39,8 @@ dependencies {
   implementation(project(":catalogs:hadoop-common")) {
     exclude("*")
   }
+  implementation(project(":clients:client-java-runtime", configuration = "shadow"))
+  implementation(project(":clients:filesystem-hadoop3-runtime", configuration = "shadow"))
 
   implementation(libs.azure.identity)
   implementation(libs.azure.storage.file.datalake)