[#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

@@ -29,13 +29,17 @@ dependencies {
   compileOnly(project(":catalogs:catalog-common"))
   compileOnly(project(":catalogs:catalog-hadoop"))
   compileOnly(project(":core"))
+  compileOnly(project(":clients:client-java"))
   compileOnly(libs.hadoop3.client.api)
   compileOnly(libs.hadoop3.client.runtime)
   compileOnly(libs.hadoop3.oss)
 
   implementation(project(":catalogs:catalog-common")) {
     exclude("*")
   }
+  implementation(project(":clients:filesystem-hadoop3-common")) {
+    exclude("*")
+  }
   implementation(project(":catalogs:hadoop-common")) {
     exclude("*")
   }

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

@@ -0,0 +1,147 @@
+/*
+ *  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.NameIdentifier;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.OSSSecretKeyCredential;
+import org.apache.gravitino.credential.OSSTokenCredential;
+import org.apache.gravitino.file.Fileset;
+import org.apache.gravitino.file.FilesetCatalog;
+import org.apache.gravitino.filesystem.common.GravitinoVirtualFileSystemConfiguration;
+import org.apache.gravitino.filesystem.common.GravitinoVirtualFileSystemUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.aliyun.oss.Constants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class OSSCredentialsProvider implements CredentialsProvider {
+
+  private static final Logger LOG = LoggerFactory.getLogger(OSSCredentialsProvider.class);
+  private Credentials basicCredentials;
+  private final String filesetIdentifier;
+  private GravitinoClient client;
+  private final Configuration configuration;
+
+  private long expirationTime = Long.MAX_VALUE;
+  private static final double EXPIRATION_TIME_FACTOR = 0.9D;
+
+  public OSSCredentialsProvider(URI uri, Configuration conf) {
+    this.filesetIdentifier =
+        conf.get(GravitinoVirtualFileSystemConfiguration.GVFS_FILESET_IDENTIFIER);
+    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) {
+      synchronized (this) {
+        try {
+          refresh();
+        } finally {
+          if (null != this.client) {
+            this.client.close();
+          }
+        }
+      }
+    }
+
+    return basicCredentials;
+  }
+
+  private void refresh() {
+    String[] idents = filesetIdentifier.split("\\.");
+    String catalog = idents[1];
+
+    client = GravitinoVirtualFileSystemUtils.createClient(configuration);
+    FilesetCatalog filesetCatalog = client.loadCatalog(catalog).asFilesetCatalog();
+
+    Fileset fileset = filesetCatalog.loadFileset(NameIdentifier.of(idents[2], idents[3]));
+    Credential[] credentials = fileset.supportsCredentials().getCredentials();
+    Credential credential = getCredential(credentials);
+
+    if (credential == null) {
+      LOG.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;
+    }
+
+    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 getCredential(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

@@ -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.common.GravitinoVirtualFileSystemConfiguration;
 import org.apache.gravitino.storage.OSSProperties;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -60,10 +61,40 @@ public FileSystem getFileSystem(Path path, Map<String, String> config) throws IO
       hadoopConfMap.put(OSS_FILESYSTEM_IMPL, AliyunOSSFileSystem.class.getCanonicalName());
     }
 
+    //    if (shouldSetCredentialsProviderExplicitly(config)) {
+    //      hadoopConfMap.put(
+    //          Constants.CREDENTIALS_PROVIDER_KEY,
+    // OSSCredentialsProvider.class.getCanonicalName());
+    //    }
+
+    if (enableCredentialProvidedByGravitino(config)) {
+      hadoopConfMap.put(
+          Constants.CREDENTIALS_PROVIDER_KEY, TestOSSCredentialProvider.class.getCanonicalName());
+    }
+
     hadoopConfMap.forEach(configuration::set);
+
     return AliyunOSSFileSystem.newInstance(path.toUri(), configuration);
   }
 
+  private boolean enableCredentialProvidedByGravitino(Map<String, String> config) {
+    return null != config.get("fs.gvfs.provider.impl");
+  }
+
+  /**
+   * Check if the credential provider should be set explicitly.
+   *
+   * <p>When the credential provider is not set and the server URI is set (this means the call is
+   * from GVFS client), we need to manually set the credential provider
+   *
+   * @param config the configuration map
+   * @return true if the credential provider should be set explicitly
+   */
+  private boolean shouldSetCredentialsProviderExplicitly(Map<String, String> config) {
+    return !config.containsKey(Constants.CREDENTIALS_PROVIDER_KEY)
+        && config.containsKey(GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_SERVER_URI_KEY);
+  }
+
   @Override
   public String scheme() {
     return "oss";

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

@@ -0,0 +1,92 @@
+/*
+ *  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.GravitinoFileSystemCredentialProvider;
+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 TestOSSCredentialProvider implements CredentialsProvider {
+  private GravitinoFileSystemCredentialProvider gravitinoFileSystemCredentialProvider;
+  private Credentials basicCredentials;
+  private long expirationTime = Long.MAX_VALUE;
+  private static final double EXPIRATION_TIME_FACTOR = 0.9D;
+
+  public TestOSSCredentialProvider(URI uri, Configuration conf) {
+    try {
+      gravitinoFileSystemCredentialProvider =
+          (GravitinoFileSystemCredentialProvider)
+              Class.forName(conf.get("fs.gvfs.credential.provider"))
+                  .getDeclaredConstructor()
+                  .newInstance();
+      gravitinoFileSystemCredentialProvider.setConf(conf);
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to create GravitinoFileSystemCredentialProvider", e);
+    }
+  }
+
+  @Override
+  public void setCredentials(Credentials credentials) {}
+
+  @Override
+  public Credentials getCredentials() {
+    if (basicCredentials == null || System.currentTimeMillis() >= expirationTime) {
+      Credential[] gravitinoCredentials = gravitinoFileSystemCredentialProvider.getCredentials();
+      if (gravitinoCredentials.length == 0) {
+        throw new RuntimeException("No credentials found");
+      }
+
+      // Get dynamic credentials from Gravitino
+      Credential gravitinoCredential = gravitinoCredentials[0];
+
+      if (gravitinoCredential instanceof OSSSecretKeyCredential) {
+        OSSSecretKeyCredential ossSecretKeyCredential =
+            (OSSSecretKeyCredential) gravitinoCredential;
+        basicCredentials =
+            new DefaultCredentials(
+                ossSecretKeyCredential.accessKeyId(), ossSecretKeyCredential.secretAccessKey());
+      } else if (gravitinoCredential instanceof OSSTokenCredential) {
+        OSSTokenCredential ossTokenCredential = (OSSTokenCredential) gravitinoCredential;
+        basicCredentials =
+            new BasicCredentials(
+                ossTokenCredential.accessKeyId(),
+                ossTokenCredential.secretAccessKey(),
+                ossTokenCredential.securityToken());
+      }
+
+      if (gravitinoCredential.expireTimeInMs() > 0) {
+        expirationTime =
+            System.currentTimeMillis()
+                + (long)
+                    ((gravitinoCredential.expireTimeInMs() - System.currentTimeMillis())
+                        * EXPIRATION_TIME_FACTOR);
+      }
+    }
+
+    return basicCredentials;
+  }
+}

bundles/aws/build.gradle.kts

@@ -29,13 +29,17 @@ dependencies {
   compileOnly(project(":catalogs:catalog-common"))
   compileOnly(project(":catalogs:catalog-hadoop"))
   compileOnly(project(":core"))
+  compileOnly(project(":clients:client-java"))
   compileOnly(libs.hadoop3.aws)
   compileOnly(libs.hadoop3.client.api)
   compileOnly(libs.hadoop3.client.runtime)
 
   implementation(project(":catalogs:catalog-common")) {
     exclude("*")
   }
+  implementation(project(":clients:filesystem-hadoop3-common")) {
+    exclude("*")
+  }
   implementation(project(":catalogs:hadoop-common")) {
     exclude("*")
   }