[#5996] feat(python-client): Using credentail in python GVFS client. (#…

…5997)

### What changes were proposed in this pull request?

Support using credentail in GVFS python client for cloud storage.

### Why are the changes needed?

It's need.

Fix: #5996 

### Does this PR introduce _any_ user-facing change?

N/A

### How was this patch tested?

New it and test locally

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();
   }
 

clients/client-python/gravitino/client/metadata_object_credential_operations.py

@@ -48,7 +48,7 @@ def __init__(
         metadata_object_type = metadata_object.type().value
         metadata_object_name = metadata_object.name()
         self._request_path = (
-            f"api/metalakes/{metalake_name}objects/{metadata_object_type}/"
+            f"api/metalakes/{metalake_name}/objects/{metadata_object_type}/"
             f"{metadata_object_name}/credentials"
         )
 

clients/client-python/gravitino/filesystem/gvfs_config.py

@@ -44,3 +44,11 @@ class GVFSConfig:
 
     GVFS_FILESYSTEM_AZURE_ACCOUNT_NAME = "abs_account_name"
     GVFS_FILESYSTEM_AZURE_ACCOUNT_KEY = "abs_account_key"
+
+    # This configuration marks the expired time of the credential. For instance, if the credential
+    # fetched from Gravitino server has expired time of 3600 seconds, and the credential_expired_time_ration is 0.5
+    # then the credential will be considered as expired after 1800 seconds and will try to retrieve a new credential.
+    GVFS_FILESYSTEM_CREDENTIAL_EXPIRED_TIME_RATIO = "credential_expiration_ratio"
+
+    # The default value of the credential_expired_time_ratio is 0.5
+    DEFAULT_CREDENTIAL_EXPIRED_TIME_RATIO = 0.5

clients/client-python/tests/integration/test_gvfs_with_abs.py

@@ -33,8 +33,6 @@
 )
 from gravitino.exceptions.base import GravitinoRuntimeException
 from gravitino.filesystem.gvfs_config import GVFSConfig
-from gravitino.filesystem.gvfs import StorageType
-
 
 logger = logging.getLogger(__name__)
 
@@ -281,7 +279,7 @@ def test_mkdir(self):
         self.assertFalse(self.fs.exists(mkdir_actual_dir))
         self.assertFalse(fs.exists(mkdir_dir))
 
-        self.assertFalse(self.fs.exists(f"{StorageType.ABS.value}://{new_bucket}"))
+        self.assertFalse(self.fs.exists("abfss://{new_bucket}"))
 
     def test_makedirs(self):
         mkdir_dir = self.fileset_gvfs_location + "/test_mkdir"
@@ -309,7 +307,7 @@ def test_makedirs(self):
         self.assertFalse(self.fs.exists(mkdir_actual_dir))
 
         self.assertFalse(fs.exists(mkdir_dir))
-        self.assertFalse(self.fs.exists(f"{StorageType.ABS.value}://{new_bucket}"))
+        self.assertFalse(self.fs.exists(f"abfss://{new_bucket}"))
 
     def test_ls(self):
         ls_dir = self.fileset_gvfs_location + "/test_ls"

clients/client-python/tests/integration/test_gvfs_with_abs_credential.py

@@ -0,0 +1,171 @@
+# 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.
+
+import logging
+import os
+from random import randint
+import unittest
+
+
+from adlfs import AzureBlobFileSystem
+
+from gravitino import (
+    gvfs,
+    GravitinoClient,
+    Catalog,
+    Fileset,
+)
+from gravitino.filesystem.gvfs_config import GVFSConfig
+from tests.integration.test_gvfs_with_abs import TestGvfsWithABS
+
+
+logger = logging.getLogger(__name__)
+
+
+def azure_abs_with_credential_is_prepared():
+    return (
+        os.environ.get("ABS_ACCOUNT_NAME_FOR_CREDENTIAL")
+        and os.environ.get("ABS_ACCOUNT_KEY_FOR_CREDENTIAL")
+        and os.environ.get("ABS_CONTAINER_NAME_FOR_CREDENTIAL")
+        and os.environ.get("ABS_TENANT_ID_FOR_CREDENTIAL")
+        and os.environ.get("ABS_CLIENT_ID_FOR_CREDENTIAL")
+        and os.environ.get("ABS_CLIENT_SECRET_FOR_CREDENTIAL")
+    )
+
+
+@unittest.skipUnless(
+    azure_abs_with_credential_is_prepared(),
+    "Azure Blob Storage credential test is not prepared.",
+)
+class TestGvfsWithCredentialABS(TestGvfsWithABS):
+    # Before running this test, please set the make sure azure-bundle-xxx.jar has been
+    # copy to the $GRAVITINO_HOME/catalogs/hadoop/libs/ directory
+    azure_abs_account_key = os.environ.get("ABS_ACCOUNT_KEY_FOR_CREDENTIAL")
+    azure_abs_account_name = os.environ.get("ABS_ACCOUNT_NAME_FOR_CREDENTIAL")
+    azure_abs_container_name = os.environ.get("ABS_CONTAINER_NAME_FOR_CREDENTIAL")
+    azure_abs_tenant_id = os.environ.get("ABS_TENANT_ID_FOR_CREDENTIAL")
+    azure_abs_client_id = os.environ.get("ABS_CLIENT_ID_FOR_CREDENTIAL")
+    azure_abs_client_secret = os.environ.get("ABS_CLIENT_SECRET_FOR_CREDENTIAL")
+
+    metalake_name: str = "TestGvfsWithCredentialABS_metalake" + str(randint(1, 10000))
+
+    def setUp(self):
+        self.options = {
+            GVFSConfig.GVFS_FILESYSTEM_AZURE_ACCOUNT_NAME: self.azure_abs_account_name,
+            GVFSConfig.GVFS_FILESYSTEM_AZURE_ACCOUNT_KEY: self.azure_abs_account_key,
+        }
+
+    @classmethod
+    def _init_test_entities(cls):
+        cls.gravitino_admin_client.create_metalake(
+            name=cls.metalake_name, comment="", properties={}
+        )
+        cls.gravitino_client = GravitinoClient(
+            uri="http://localhost:8090", metalake_name=cls.metalake_name
+        )
+
+        cls.config = {}
+        cls.conf = {}
+        catalog = cls.gravitino_client.create_catalog(
+            name=cls.catalog_name,
+            catalog_type=Catalog.Type.FILESET,
+            provider=cls.catalog_provider,
+            comment="",
+            properties={
+                "filesystem-providers": "abs",
+                "azure-storage-account-name": cls.azure_abs_account_name,
+                "azure-storage-account-key": cls.azure_abs_account_key,
+                "azure-tenant-id": cls.azure_abs_tenant_id,
+                "azure-client-id": cls.azure_abs_client_id,
+                "azure-client-secret": cls.azure_abs_client_secret,
+                "credential-providers": "adls-token",
+            },
+        )
+        catalog.as_schemas().create_schema(
+            schema_name=cls.schema_name, comment="", properties={}
+        )
+
+        cls.fileset_storage_location: str = (
+            f"{cls.azure_abs_container_name}/{cls.catalog_name}/{cls.schema_name}/{cls.fileset_name}"
+        )
+        cls.fileset_gvfs_location = (
+            f"gvfs://fileset/{cls.catalog_name}/{cls.schema_name}/{cls.fileset_name}"
+        )
+        catalog.as_fileset_catalog().create_fileset(
+            ident=cls.fileset_ident,
+            fileset_type=Fileset.Type.MANAGED,
+            comment=cls.fileset_comment,
+            storage_location=(
+                f"abfss://{cls.azure_abs_container_name}@{cls.azure_abs_account_name}.dfs.core.windows.net/"
+                f"{cls.catalog_name}/{cls.schema_name}/{cls.fileset_name}"
+            ),
+            properties=cls.fileset_properties,
+        )
+
+        cls.fs = AzureBlobFileSystem(
+            account_name=cls.azure_abs_account_name,
+            account_key=cls.azure_abs_account_key,
+        )
+
+    # As the permission provided by the dynamic token is smaller compared to the permission provided by the static token
+    # like account key and account name, the test case will fail if we do not override the test case.
+    def test_mkdir(self):
+        mkdir_dir = self.fileset_gvfs_location + "/test_mkdir"
+        mkdir_actual_dir = self.fileset_storage_location + "/test_mkdir"
+        fs = gvfs.GravitinoVirtualFileSystem(
+            server_uri="http://localhost:8090",
+            metalake_name=self.metalake_name,
+            options=self.options,
+            **self.conf,
+        )
+
+        # it actually takes no effect.
+        self.check_mkdir(mkdir_dir, mkdir_actual_dir, fs)
+
+        # check whether it will automatically create the bucket if 'create_parents'
+        # is set to True.
+        new_bucket = self.azure_abs_container_name + "2"
+        mkdir_actual_dir = mkdir_actual_dir.replace(
+            self.azure_abs_container_name, new_bucket
+        )
+        self.fs.mkdir(mkdir_actual_dir, create_parents=True)
+
+        self.assertFalse(self.fs.exists(mkdir_actual_dir))
+
+        self.assertTrue(self.fs.exists(f"abfss://{new_bucket}"))
+
+    def test_makedirs(self):
+        mkdir_dir = self.fileset_gvfs_location + "/test_mkdir"
+        mkdir_actual_dir = self.fileset_storage_location + "/test_mkdir"
+        fs = gvfs.GravitinoVirtualFileSystem(
+            server_uri="http://localhost:8090",
+            metalake_name=self.metalake_name,
+            options=self.options,
+            **self.conf,
+        )
+
+        # it actually takes no effect.
+        self.check_mkdir(mkdir_dir, mkdir_actual_dir, fs)
+
+        # check whether it will automatically create the bucket if 'create_parents'
+        # is set to True.
+        new_bucket = self.azure_abs_container_name + "1"
+        new_mkdir_actual_dir = mkdir_actual_dir.replace(
+            self.azure_abs_container_name, new_bucket
+        )
+        self.fs.makedirs(new_mkdir_actual_dir)
+        self.assertFalse(self.fs.exists(mkdir_actual_dir))

clients/client-python/tests/integration/test_gvfs_with_gcs.py

@@ -36,7 +36,7 @@
 logger = logging.getLogger(__name__)
 
 
-def oss_is_configured():
+def gcs_is_configured():
     return all(
         [
             os.environ.get("GCS_SERVICE_ACCOUNT_JSON_PATH") is not None,
@@ -45,7 +45,7 @@ def oss_is_configured():
     )
 
 
-@unittest.skipUnless(oss_is_configured(), "GCS is not configured.")
+@unittest.skipUnless(gcs_is_configured(), "GCS is not configured.")
 class TestGvfsWithGCS(TestGvfsWithHDFS):
     # Before running this test, please set the make sure gcp-bundle-x.jar has been
     # copy to the $GRAVITINO_HOME/catalogs/hadoop/libs/ directory
@@ -254,11 +254,10 @@ def test_mkdir(self):
         new_bucket = self.bucket_name + "1"
         mkdir_dir = mkdir_dir.replace(self.bucket_name, new_bucket)
         mkdir_actual_dir = mkdir_actual_dir.replace(self.bucket_name, new_bucket)
-        fs.mkdir(mkdir_dir, create_parents=True)
 
+        with self.assertRaises(OSError):
+            fs.mkdir(mkdir_dir, create_parents=True)
         self.assertFalse(self.fs.exists(mkdir_actual_dir))
-        self.assertFalse(fs.exists(mkdir_dir))
-        self.assertFalse(self.fs.exists("gs://" + new_bucket))
 
     def test_makedirs(self):
         mkdir_dir = self.fileset_gvfs_location + "/test_mkdir"

clients/client-python/tests/integration/test_gvfs_with_gcs_credential.py

@@ -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.
+
+import logging
+import os
+from random import randint
+import unittest
+
+from gcsfs import GCSFileSystem
+
+from gravitino import Catalog, Fileset, GravitinoClient
+from gravitino.filesystem import gvfs
+from tests.integration.test_gvfs_with_gcs import TestGvfsWithGCS
+
+logger = logging.getLogger(__name__)
+
+
+def gcs_with_credential_is_configured():
+    return all(
+        [
+            os.environ.get("GCS_SERVICE_ACCOUNT_JSON_PATH_FOR_CREDENTIAL") is not None,
+            os.environ.get("GCS_BUCKET_NAME_FOR_CREDENTIAL") is not None,
+        ]
+    )
+
+
+@unittest.skipUnless(gcs_with_credential_is_configured(), "GCS is not configured.")
+class TestGvfsWithGCSCredential(TestGvfsWithGCS):
+    # Before running this test, please set the make sure gcp-bundle-x.jar has been
+    # copy to the $GRAVITINO_HOME/catalogs/hadoop/libs/ directory
+    key_file = os.environ.get("GCS_SERVICE_ACCOUNT_JSON_PATH_FOR_CREDENTIAL")
+    bucket_name = os.environ.get("GCS_BUCKET_NAME_FOR_CREDENTIAL")
+    metalake_name: str = "TestGvfsWithGCSCredential_metalake" + str(randint(1, 10000))
+
+    @classmethod
+    def _init_test_entities(cls):
+        cls.gravitino_admin_client.create_metalake(
+            name=cls.metalake_name, comment="", properties={}
+        )
+        cls.gravitino_client = GravitinoClient(
+            uri="http://localhost:8090", metalake_name=cls.metalake_name
+        )
+
+        cls.config = {}
+        cls.conf = {}
+        catalog = cls.gravitino_client.create_catalog(
+            name=cls.catalog_name,
+            catalog_type=Catalog.Type.FILESET,
+            provider=cls.catalog_provider,
+            comment="",
+            properties={
+                "filesystem-providers": "gcs",
+                "gcs-credential-file-path": cls.key_file,
+                "gcs-service-account-file": cls.key_file,
+                "credential-providers": "gcs-token",
+            },
+        )
+        catalog.as_schemas().create_schema(
+            schema_name=cls.schema_name, comment="", properties={}
+        )
+
+        cls.fileset_storage_location: str = (
+            f"gs://{cls.bucket_name}/{cls.catalog_name}/{cls.schema_name}/{cls.fileset_name}"
+        )
+        cls.fileset_gvfs_location = (
+            f"gvfs://fileset/{cls.catalog_name}/{cls.schema_name}/{cls.fileset_name}"
+        )
+        catalog.as_fileset_catalog().create_fileset(
+            ident=cls.fileset_ident,
+            fileset_type=Fileset.Type.MANAGED,
+            comment=cls.fileset_comment,
+            storage_location=cls.fileset_storage_location,
+            properties=cls.fileset_properties,
+        )
+
+        cls.fs = GCSFileSystem(token=cls.key_file)
+
+    def test_mkdir(self):
+        mkdir_dir = self.fileset_gvfs_location + "/test_mkdir"
+        mkdir_actual_dir = self.fileset_storage_location + "/test_mkdir"
+        fs = gvfs.GravitinoVirtualFileSystem(
+            server_uri="http://localhost:8090",
+            metalake_name=self.metalake_name,
+            options=self.options,
+            **self.conf,
+        )
+
+        # it actually takes no effect.
+        self.check_mkdir(mkdir_dir, mkdir_actual_dir, fs)
+
+        # check whether it will automatically create the bucket if 'create_parents'
+        # is set to True.
+        new_bucket = self.bucket_name + "1"
+        mkdir_dir = mkdir_dir.replace(self.bucket_name, new_bucket)
+        mkdir_actual_dir = mkdir_actual_dir.replace(self.bucket_name, new_bucket)
+
+        fs.mkdir(mkdir_dir, create_parents=True)
+        self.assertFalse(self.fs.exists(mkdir_actual_dir))