This is an automated email from the ASF dual-hosted git repository.
jshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 6bea0e9320 [#6942] feat(clients): clients and gvfs support config
request headers (#6976)
6bea0e9320 is described below
commit 6bea0e9320ed34584fff639b377de1aa9bf36386
Author: mchades <[email protected]>
AuthorDate: Thu Apr 17 17:58:15 2025 +0800
[#6942] feat(clients): clients and gvfs support config request headers
(#6976)
### What changes were proposed in this pull request?
clients and gvfs support config request headers
### Why are the changes needed?
Fix: #6942
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
tests added
---
.../gravitino/client/gravitino_client.py | 4 +-
.../gravitino/client/gravitino_client_base.py | 5 ++-
.../gravitino/filesystem/gvfs_base_operations.py | 17 +++++++-
.../gravitino/filesystem/gvfs_config.py | 3 ++
.../gravitino/filesystem/gvfs_utils.py | 2 +
.../tests/unittests/test_gravitino_client.py | 45 +++++++++++++++++++++
.../tests/unittests/test_gvfs_with_local.py | 13 ++++++
.../GravitinoVirtualFileSystemConfiguration.java | 4 ++
.../hadoop/GravitinoVirtualFileSystemUtils.java | 14 +++++++
.../filesystem/hadoop/GravitinoMockServerBase.java | 8 ++++
.../gravitino/filesystem/hadoop/TestGvfsBase.java | 47 ++++++++++++++++++++++
docs/how-to-use-gvfs.md | 2 +
12 files changed, 161 insertions(+), 3 deletions(-)
diff --git a/clients/client-python/gravitino/client/gravitino_client.py
b/clients/client-python/gravitino/client/gravitino_client.py
index e182f5a200..a8c3cd8535 100644
--- a/clients/client-python/gravitino/client/gravitino_client.py
+++ b/clients/client-python/gravitino/client/gravitino_client.py
@@ -39,6 +39,7 @@ class GravitinoClient(GravitinoClientBase):
metalake_name: str,
check_version: bool = True,
auth_data_provider: AuthDataProvider = None,
+ request_headers: dict = None,
):
"""Constructs a new GravitinoClient with the given URI, authenticator
and AuthDataProvider.
@@ -46,11 +47,12 @@ class GravitinoClient(GravitinoClientBase):
uri: The base URI for the Gravitino API.
metalake_name: The specified metalake name.
auth_data_provider: The provider of the data which is used for
authentication.
+ request_headers: The headers to be included in the HTTP requests.
Raises:
NoSuchMetalakeException if the metalake with specified name does
not exist.
"""
- super().__init__(uri, check_version, auth_data_provider)
+ super().__init__(uri, check_version, auth_data_provider,
request_headers)
self.check_metalake_name(metalake_name)
self._metalake = super().load_metalake(metalake_name)
diff --git a/clients/client-python/gravitino/client/gravitino_client_base.py
b/clients/client-python/gravitino/client/gravitino_client_base.py
index b61637f6a0..7a1744595f 100644
--- a/clients/client-python/gravitino/client/gravitino_client_base.py
+++ b/clients/client-python/gravitino/client/gravitino_client_base.py
@@ -56,8 +56,11 @@ class GravitinoClientBase:
uri: str,
check_version: bool = True,
auth_data_provider: AuthDataProvider = None,
+ request_headers: dict = None,
):
- self._rest_client = HTTPClient(uri,
auth_data_provider=auth_data_provider)
+ self._rest_client = HTTPClient(
+ uri, auth_data_provider=auth_data_provider,
request_headers=request_headers
+ )
if check_version:
self.check_version()
diff --git a/clients/client-python/gravitino/filesystem/gvfs_base_operations.py
b/clients/client-python/gravitino/filesystem/gvfs_base_operations.py
index b847cccf59..62cc40934c 100644
--- a/clients/client-python/gravitino/filesystem/gvfs_base_operations.py
+++ b/clients/client-python/gravitino/filesystem/gvfs_base_operations.py
@@ -86,7 +86,22 @@ class BaseGVFSOperations(ABC):
self._metalake = metalake_name
self._options = options
- self._client = create_client(options, server_uri, metalake_name)
+ request_headers = (
+ None
+ if options is None
+ else {
+ key[
+
len(GVFSConfig.GVFS_FILESYSTEM_CLIENT_REQUEST_HEADER_PREFIX) :
+ ]: value
+ for key, value in options.items()
+ if key.startswith(
+ GVFSConfig.GVFS_FILESYSTEM_CLIENT_REQUEST_HEADER_PREFIX
+ )
+ }
+ )
+ self._client = create_client(
+ options, server_uri, metalake_name, request_headers
+ )
cache_size = (
GVFSConfig.DEFAULT_CACHE_SIZE
diff --git a/clients/client-python/gravitino/filesystem/gvfs_config.py
b/clients/client-python/gravitino/filesystem/gvfs_config.py
index 47aa55b763..159b5ab986 100644
--- a/clients/client-python/gravitino/filesystem/gvfs_config.py
+++ b/clients/client-python/gravitino/filesystem/gvfs_config.py
@@ -67,3 +67,6 @@ class GVFSConfig:
# The hook class that will be used to intercept file system operations.
GVFS_FILESYSTEM_HOOK = "hook_class"
+
+ # The configuration key prefix for the client request headers.
+ GVFS_FILESYSTEM_CLIENT_REQUEST_HEADER_PREFIX = "client_request_header_"
diff --git a/clients/client-python/gravitino/filesystem/gvfs_utils.py
b/clients/client-python/gravitino/filesystem/gvfs_utils.py
index b3b7f8f385..bb4553288a 100644
--- a/clients/client-python/gravitino/filesystem/gvfs_utils.py
+++ b/clients/client-python/gravitino/filesystem/gvfs_utils.py
@@ -67,6 +67,7 @@ def create_client(
uri=server_uri,
metalake_name=metalake_name,
auth_data_provider=SimpleAuthProvider(),
+ request_headers=request_headers,
)
if auth_type == GVFSConfig.OAUTH2_AUTH_TYPE:
@@ -90,6 +91,7 @@ def create_client(
uri=server_uri,
metalake_name=metalake_name,
auth_data_provider=oauth2_token_provider,
+ request_headers=request_headers,
)
raise GravitinoRuntimeException(
diff --git a/clients/client-python/tests/unittests/test_gravitino_client.py
b/clients/client-python/tests/unittests/test_gravitino_client.py
new file mode 100644
index 0000000000..ef8e87615a
--- /dev/null
+++ b/clients/client-python/tests/unittests/test_gravitino_client.py
@@ -0,0 +1,45 @@
+# 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 unittest
+
+from gravitino import GravitinoAdminClient, GravitinoClient
+from tests.unittests import mock_base
+
+
+@mock_base.mock_data
+class TestMetalake(unittest.TestCase):
+ # pylint: disable=W0212
+ def test_gravitino_client_headers(self, *mock_methods):
+ expected_headers = {
+ "k1": "v1",
+ }
+ gravitino_admin_client = GravitinoAdminClient(
+ uri="http://localhost:8090";,
+ request_headers=expected_headers,
+ )
+ self.assertEqual(
+ expected_headers,
gravitino_admin_client._rest_client.request_headers
+ )
+
+ gravitino_client = GravitinoClient(
+ uri="http://localhost:8090";,
+ metalake_name="test",
+ request_headers=expected_headers,
+ )
+ self.assertEqual(
+ expected_headers, gravitino_client._rest_client.request_headers
+ )
diff --git a/clients/client-python/tests/unittests/test_gvfs_with_local.py
b/clients/client-python/tests/unittests/test_gvfs_with_local.py
index 8090f051ed..30b5c0f436 100644
--- a/clients/client-python/tests/unittests/test_gvfs_with_local.py
+++ b/clients/client-python/tests/unittests/test_gvfs_with_local.py
@@ -82,6 +82,19 @@ class TestLocalFilesystem(unittest.TestCase):
if local_fs.exists(self._local_base_dir_path):
local_fs.rm(self._local_base_dir_path, recursive=True)
+ def test_request_headers(self, *mock_methods):
+ options = {
+ f"{GVFSConfig.GVFS_FILESYSTEM_CLIENT_REQUEST_HEADER_PREFIX}k1":
"v1",
+ }
+ fs = gvfs.GravitinoVirtualFileSystem(
+ server_uri="http://localhost:9090";,
+ metalake_name="metalake_demo",
+ options=options,
+ skip_instance_cache=True,
+ )
+ headers = fs._operations._client._rest_client.request_headers
+ self.assertEqual(headers["k1"], "v1")
+
def test_cache(self, *mock_methods):
fileset_storage_location = f"{self._fileset_dir}/test_cache"
fileset_virtual_location = "fileset/fileset_catalog/tmp/test_cache"
diff --git
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java
index c79cd6e890..f541a3492c 100644
---
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java
+++
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java
@@ -135,5 +135,9 @@ public class GravitinoVirtualFileSystemConfiguration {
/** The default value for the Gravitino hook class. */
public static final String FS_GRAVITINO_HOOK_CLASS_DEFAULT =
NoOpHook.class.getCanonicalName();
+ /** The configuration key prefix for the Gravitino client request header. */
+ public static final String FS_GRAVITINO_CLIENT_REQUEST_HEADER_PREFIX =
+ "fs.gravitino.client.request.header.";
+
private GravitinoVirtualFileSystemConfiguration() {}
}
diff --git
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemUtils.java
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemUtils.java
index 181843cfd5..926637072b 100644
---
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemUtils.java
+++
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemUtils.java
@@ -19,12 +19,15 @@
package org.apache.gravitino.filesystem.hadoop;
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CLIENT_REQUEST_HEADER_PREFIX;
+
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import java.io.File;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.client.DefaultOAuth2TokenProvider;
@@ -83,6 +86,14 @@ public class GravitinoVirtualFileSystemUtils {
"'%s' is not set in the configuration",
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_SERVER_URI_KEY);
+ Map<String, String> requestHeaders =
+ configuration.entrySet().stream()
+ .filter(e ->
e.getKey().startsWith(FS_GRAVITINO_CLIENT_REQUEST_HEADER_PREFIX))
+ .collect(
+ Collectors.toMap(
+ e ->
e.getKey().substring(FS_GRAVITINO_CLIENT_REQUEST_HEADER_PREFIX.length()),
+ Map.Entry::getValue));
+
String authType =
configuration.getOrDefault(
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CLIENT_AUTH_TYPE_KEY,
@@ -91,6 +102,7 @@ public class GravitinoVirtualFileSystemUtils {
return GravitinoClient.builder(serverUri)
.withMetalake(metalakeValue)
.withSimpleAuth()
+ .withHeaders(requestHeaders)
.build();
} else if (authType.equalsIgnoreCase(
GravitinoVirtualFileSystemConfiguration.OAUTH2_AUTH_TYPE)) {
@@ -137,6 +149,7 @@ public class GravitinoVirtualFileSystemUtils {
return GravitinoClient.builder(serverUri)
.withMetalake(metalakeValue)
.withOAuth(authDataProvider)
+ .withHeaders(requestHeaders)
.build();
} else if (authType.equalsIgnoreCase(
GravitinoVirtualFileSystemConfiguration.KERBEROS_AUTH_TYPE)) {
@@ -167,6 +180,7 @@ public class GravitinoVirtualFileSystemUtils {
return GravitinoClient.builder(serverUri)
.withMetalake(metalakeValue)
.withKerberosAuth(authDataProvider)
+ .withHeaders(requestHeaders)
.build();
} else {
throw new IllegalArgumentException(
diff --git
a/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/GravitinoMockServerBase.java
b/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/GravitinoMockServerBase.java
index ba5ca5a161..b950536d12 100644
---
a/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/GravitinoMockServerBase.java
+++
b/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/GravitinoMockServerBase.java
@@ -203,6 +203,14 @@ public abstract class GravitinoMockServerBase {
}
}
+ protected String getJsonString(Object obj) {
+ try {
+ return MAPPER.writeValueAsString(obj);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
public static ClientAndServer mockServer() {
return mockServer;
}
diff --git
a/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/TestGvfsBase.java
b/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/TestGvfsBase.java
index eb6c674961..d9084f90f3 100644
---
a/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/TestGvfsBase.java
+++
b/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/TestGvfsBase.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.filesystem.hadoop;
import static org.apache.gravitino.file.Fileset.LOCATION_NAME_UNKNOWN;
import static org.apache.gravitino.file.Fileset.PROPERTY_DEFAULT_LOCATION_NAME;
import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_BLOCK_SIZE_DEFAULT;
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CLIENT_REQUEST_HEADER_PREFIX;
import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.extractIdentifier;
import static org.apache.hc.core5.http.HttpStatus.SC_NOT_FOUND;
import static org.apache.hc.core5.http.HttpStatus.SC_OK;
@@ -35,6 +36,8 @@ import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyShort;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.collect.ImmutableMap;
@@ -52,6 +55,7 @@ import java.util.Objects;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Version;
import org.apache.gravitino.dto.AuditDTO;
import org.apache.gravitino.dto.credential.CredentialDTO;
import org.apache.gravitino.dto.file.FilesetDTO;
@@ -59,6 +63,7 @@ import org.apache.gravitino.dto.responses.CredentialResponse;
import org.apache.gravitino.dto.responses.ErrorResponse;
import org.apache.gravitino.dto.responses.FileLocationResponse;
import org.apache.gravitino.dto.responses.FilesetResponse;
+import org.apache.gravitino.dto.responses.VersionResponse;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchFilesetException;
import org.apache.gravitino.exceptions.NoSuchLocationNameException;
@@ -80,6 +85,9 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
+import org.mockserver.matchers.Times;
+import org.mockserver.model.HttpRequest;
+import org.mockserver.verify.VerificationTimes;
public class TestGvfsBase extends GravitinoMockServerBase {
protected static final String GVFS_IMPL_CLASS =
GravitinoVirtualFileSystem.class.getName();
@@ -238,6 +246,45 @@ public class TestGvfsBase extends GravitinoMockServerBase {
}
}
+ @Test
+ public void testRequestHeaders()
+ throws NoSuchFieldException, IllegalAccessException, IOException {
+ String envKey = "GRAVITINO_TEST_HEADER";
+ String envValue = "v1";
+ String headerKey1 = "k1";
+ // prepare the env variable
+ Map<String, String> env = System.getenv();
+ Field field = env.getClass().getDeclaredField("m");
+ field.setAccessible(true);
+ Map<String, String> writableEnv = (Map<String, String>) field.get(env);
+ writableEnv.put(envKey, envValue);
+
+ // test the request headers
+ String headerKey2 = "k2";
+ String headerValue2 = "v2";
+ Configuration configuration = new Configuration(conf);
+ configuration.set(
+ FS_GRAVITINO_CLIENT_REQUEST_HEADER_PREFIX + headerKey1, "${env." +
envKey + "}");
+ configuration.set(FS_GRAVITINO_CLIENT_REQUEST_HEADER_PREFIX + headerKey2,
headerValue2);
+
+ mockServer().clear(request().withPath("/api/version"));
+ HttpRequest req =
+ request()
+ .withHeader(headerKey1, envValue)
+ .withHeader(headerKey2, headerValue2)
+ .withPath("/api/version");
+ mockServer()
+ .when(req, Times.once())
+ .respond(
+ response()
+ .withStatusCode(SC_OK)
+ .withBody(getJsonString(new
VersionResponse(Version.getCurrentVersionDTO()))));
+
+ try (FileSystem fs = new
Path("gvfs://fileset/").getFileSystem(configuration)) {
+ mockServer().verify(req, VerificationTimes.once());
+ }
+ }
+
@Test
public void testFSCache() throws IOException {
String filesetName = "testFSCache";
diff --git a/docs/how-to-use-gvfs.md b/docs/how-to-use-gvfs.md
index caaa2e92e8..996c61f947 100644
--- a/docs/how-to-use-gvfs.md
+++ b/docs/how-to-use-gvfs.md
@@ -69,6 +69,7 @@ the path mapping and convert automatically.
| `fs.gravitino.current.location.name.env.var` | The environment
variable name to get the current location name.
| `CURRENT_LOCATION_NAME`
| No |
0.9.0-incubating |
| `fs.gravitino.operations.class` | The operations class
to provide the FS operations for the Gravitino Virtual File System. Users can
extends `BaseGVFSOperations` to implement their own operations and configure
the class name in this conf to use custom FS operations.
|
`org.apache.gravitino.filesystem.hadoop.DefaultGVFSOperations` | No
| 0.9.0-incubating |
| `fs.gravitino.hook.class` | The hook class to
inject into the <br/>Gravitino Virtual File System. Users can implement their
own `GravitinoVirtualFileSystemHook` and configure the class name in this conf
to inject custom code.
|
`org.apache.gravitino.filesystem.hadoop.NoOpHook` | No
| 0.9.0-incubating |
+| `fs.gravitino.client.request.header.` | The configuration
key prefix for the Gravitino client request header. You can set the request
header for the Gravitino client.
| (none)
| No |
0.9.0-incubating |
Apart from the above properties, to access fileset like S3, GCS, OSS and
custom fileset, extra properties are needed, please see
[S3 GVFS Java client
configurations](./hadoop-catalog-with-s3.md#using-the-gvfs-java-client-to-access-the-fileset),
[GCS GVFS Java client
configurations](./hadoop-catalog-with-gcs.md#using-the-gvfs-java-client-to-access-the-fileset),
[OSS GVFS Java client
configurations](./hadoop-catalog-with-oss.md#using-the-gvfs-java-client-to-access-the-fileset)
and [Azure Blob Storage GVFS Java client
configurations](./hadoop-catalog-with-adls.md#using-the-gvfs-java-client-to-access-the-fileset)
for [...]
@@ -363,6 +364,7 @@ to recompile the native libraries like `libhdfs` and
others, and completely repl
| `current_location_name_env_var` | The environment variable name to get the
current location name.
| `CURRENT_LOCATION_NAME`
| No | 0.9.0-incubating |
| `operations_class` | The operations class to provide the FS
operations for the Gravitino Virtual File System. Users can extends
`BaseGVFSOperations` to implement their own operations and configure the class
name in this conf to use custom FS operations.
|
`gravitino.filesystem.gvfs_default_operations.DefaultGVFSOperations` | No
| 0.9.0-incubating |
| `hook_class` | The hook class to inject into the
Gravitino Virtual File System. Users can implement their own
`GravitinoVirtualFileSystemHook` and configure the class name in this conf to
inject custom code.
|
`gravitino.filesystem.gvfs_hook.NoOpHook` | No
| 0.9.0-incubating |
+| `client_request_header_` | The configuration key prefix for the
Gravitino client request header. You can set the request header for the
Gravitino client.
| (none)
| No | 0.9.0-incubating |
#### Configurations for S3, GCS, OSS and Azure Blob storage fileset
