[filesystem/oss] OssFileSystem should allow to configure credential provider (#939)

Author: luoyuxia

fluss-filesystems/fluss-fs-oss/src/main/java/com/alibaba/fluss/fs/oss/OSSFileSystemPlugin.java

@@ -68,11 +68,19 @@ public FileSystem create(URI fsUri, Configuration flussConfig) throws IOExceptio
 
         // set credential provider
         if (hadoopConfig.get(ACCESS_KEY_ID) == null) {
-            LOG.info(
-                    "{} is not set, using credential provider {}.",
-                    ACCESS_KEY_ID,
-                    hadoopConfig.get(CREDENTIALS_PROVIDER_KEY));
-            setCredentialProvider(flussConfig, hadoopConfig);
+            String credentialsProvider = hadoopConfig.get(CREDENTIALS_PROVIDER_KEY);
+            if (credentialsProvider != null) {
+                LOG.info(
+                        "{} is not set, but {} is set, using credential provider {}.",
+                        ACCESS_KEY_ID,
+                        CREDENTIALS_PROVIDER_KEY,
+                        credentialsProvider);
+            } else {
+                // no ak, no credentialsProvider,
+                // set default credential provider which will get token from
+                // OSSSecurityTokenReceiver
+                setDefaultCredentialProvider(flussConfig, hadoopConfig);
+            }
         } else {
             LOG.info("{} is set, using provided access key id and secret.", ACCESS_KEY_ID);
         }
@@ -101,6 +109,12 @@ protected org.apache.hadoop.fs.FileSystem initFileSystem(
         return fileSystem;
     }
 
+    protected void setDefaultCredentialProvider(
+            Configuration flussConfig, org.apache.hadoop.conf.Configuration hadoopConfig) {
+        // use OSSSecurityTokenReceiver to update hadoop config to set credentialsProvider
+        OSSSecurityTokenReceiver.updateHadoopConfig(hadoopConfig);
+    }
+
     @VisibleForTesting
     org.apache.hadoop.conf.Configuration getHadoopConfiguration(Configuration flussConfig) {
         org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
@@ -127,11 +141,6 @@ org.apache.hadoop.conf.Configuration getHadoopConfiguration(Configuration flussC
         return conf;
     }
 
-    protected void setCredentialProvider(
-            Configuration flussConfig, org.apache.hadoop.conf.Configuration hadoopConfig) {
-        OSSSecurityTokenReceiver.updateHadoopConfig(hadoopConfig);
-    }
-
     private void setSignatureVersion4(
             AliyunOSSFileSystem aliyunOSSFileSystem,
             org.apache.hadoop.conf.Configuration hadoopConfig) {

fluss-filesystems/fluss-fs-oss/src/test/java/com/alibaba/fluss/fs/oss/OSSWithCredentialsProviderFileSystemBehaviorITCase.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.fs.oss;
+
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.fs.FileSystem;
+import com.alibaba.fluss.fs.FileSystemBehaviorTestSuite;
+import com.alibaba.fluss.fs.FsPath;
+
+import com.aliyun.oss.common.auth.CredentialsProvider;
+import com.aliyun.oss.common.auth.SystemPropertiesCredentialsProvider;
+import org.junit.jupiter.api.BeforeAll;
+
+import java.util.UUID;
+
+import static org.apache.hadoop.fs.aliyun.oss.Constants.CREDENTIALS_PROVIDER_KEY;
+
+/** IT case for access oss via set {@link CredentialsProvider}. */
+class OSSWithCredentialsProviderFileSystemBehaviorITCase extends FileSystemBehaviorTestSuite {
+
+    private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
+
+    @BeforeAll
+    static void setup() {
+        OSSTestCredentials.assumeCredentialsAvailable();
+
+        // use SystemPropertiesCredentialsProvider
+        final Configuration conf = new Configuration();
+        conf.setString(
+                CREDENTIALS_PROVIDER_KEY,
+                SystemPropertiesCredentialsProvider.class.getCanonicalName());
+        conf.setString("fs.oss.endpoint", OSSTestCredentials.getOSSEndpoint());
+        conf.setString("fs.oss.region", OSSTestCredentials.getOSSRegion());
+
+        // now, we need to set oss config to system properties
+        System.setProperty("oss.accessKeyId", OSSTestCredentials.getOSSAccessKey());
+        System.setProperty("oss.accessKeySecret", OSSTestCredentials.getOSSSecretKey());
+        FileSystem.initialize(conf, null);
+    }
+
+    @Override
+    protected FileSystem getFileSystem() throws Exception {
+        return getBasePath().getFileSystem();
+    }
+
+    @Override
+    protected FsPath getBasePath() throws Exception {
+        return new FsPath(OSSTestCredentials.getTestBucketUri() + TEST_DATA_DIR);
+    }
+}

website/docs/maintenance/filesystems/oss.md

@@ -35,16 +35,34 @@ To enabled OSS as remote storage, there are some required configurations that mu
 remote.data.dir: oss://<your-bucket>/path/to/remote/storage
 # Aliyun OSS endpoint to connect to, such as: oss-cn-hangzhou.aliyuncs.com
 fs.oss.endpoint: <your-endpoint>
+# Aliyun STS endpoint to connect to obtain a STS token, such as: sts.cn-hangzhou.aliyuncs.com
+fs.oss.sts.endpoint: <your-sts-endpoint>
+# For the role of the STS token obtained from the STS endpoint, such as: acs:ram::123456789012:role/testrole
+fs.oss.roleArn: <your-role-arn>
+
+# Authentication (choose one option below)
+
+# Option 1: Direct credentials
 # Aliyun access key ID
 fs.oss.accessKeyId: <your-access-key>
 # Aliyun access key secret
 fs.oss.accessKeySecret: <your-secret-key>
 
-# Aliyun STS endpoint to connect to obtain a STS token, such as: sts.cn-hangzhou.aliyuncs.com
-fs.oss.sts.endpoint: <your-sts-endpoint>
-# For the role of the STS token obtained from the STS endpoint, such as: acs:ram::123456789012:role/testrole
-fs.oss.roleArn: <your-role-arn>
+# Option 2: Secure credential provider
+fs.oss.credentials.provider: <your-credentials-provider>
+```
+To avoid exposing sensitive access key information directly in the `server.yaml`, you can choose option2 to use a credential provider by setting the `fs.oss.credentials.provider` property.
+
+For example, to use environment variables for credential management:
+```yaml
+fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider
+```
+Then, set the following environment variables before starting the Fluss service:
+```bash
+export OSS_ACCESS_KEY_ID=<your-access-key>
+export OSS_ACCESS_KEY_SECRET=<your-secret-key>
 ```
+This approach enhances security by keeping sensitive credentials out of configuration files.
 
 ## Token-based Authentication