This is an automated email from the ASF dual-hosted git repository. roryqi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 33151dbee715259737cde8c7763b4bb1cc9923b5 Author: MaSai <[email protected]> AuthorDate: Tue Oct 20 15:13:04 2026 +0800 [#13797] feat(tms): Add cluster-static gravitinoAuth/ircAuth overlay bags (#13865) ### What changes were proposed in this pull request? - Add `MaintenanceAuthConfigs` with optional `ConfigEntry` keys under `gravitino.maintenance.gravitinoAuth.*` and `gravitino.maintenance.ircAuth.*` - Load suffix-keyed auth bags from `gravitino.conf` and overlay onto policy `jobOptions` / `updateOptions` (auth wins on duplicate keys) - Docs in `docs/table-maintenance-service/optimizer-configuration.md` - Unit tests for config load and overlay behavior **Out of scope:** wiring the overlay into the live TMS scheduler `runJob` call site. ### Why are the changes needed? Fourth subtask of Epic #13760: cluster-static client auth must live in `gravitino.conf`, not in policy content. Fix: #13797 ### Does this PR introduce _any_ user-facing change? Yes. Operators can declare optional `gravitino.maintenance.gravitinoAuth.*` / `ircAuth.*` keys in `gravitino.conf`. Overlay into live `runJob` is not wired yet. ### How was this patch tested? ```bash ./gradlew :core:test --tests org.apache.gravitino.maintenance.policy.TestMaintenanceAuthConfigs -PskipITs ``` Made with [Cursor](https://cursor.com) --------- Co-authored-by: Cursor <[email protected]> --- .../maintenance/policy/MaintenanceAuthConfigs.java | 321 +++++++++++++++++++++ .../policy/TestMaintenanceAuthConfigs.java | 293 +++++++++++++++++++ .../optimizer-configuration.md | 161 +++++++++-- 3 files changed, 756 insertions(+), 19 deletions(-) diff --git a/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenanceAuthConfigs.java b/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenanceAuthConfigs.java new file mode 100644 index 0000000000..a90451bbd3 --- /dev/null +++ b/core/src/main/java/org/apache/gravitino/maintenance/policy/MaintenanceAuthConfigs.java @@ -0,0 +1,321 @@ +/* + * 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.maintenance.policy; + +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import javax.annotation.Nullable; +import org.apache.gravitino.Config; +import org.apache.gravitino.config.ConfigBuilder; +import org.apache.gravitino.config.ConfigConstants; +import org.apache.gravitino.config.ConfigEntry; + +/** + * Cluster-static TMS client auth in {@code gravitino.conf}. + * + * <p>{@link #GRAVITINO_AUTH_PREFIX} is a <strong>fixed</strong> camelCase bag in {@code + * gravitino.conf} ({@code authType}, {@code oauthServerUri}, …). {@link #loadGravitinoAuth} maps + * those suffixes onto built-in updater short names ({@code auth_type}, {@code oauth_server_uri}, …) + * before overlaying onto policy {@code updaterOptions}. Unknown or blank suffixes are rejected, and + * {@code authType} plus mode-required fields are validated at load time. + * + * <p>{@link #IRC_AUTH_PREFIX} is an <strong>open</strong> passthrough for Spark / Iceberg REST + * properties (for example {@code rest.auth.type}); any non-empty suffix under the prefix is loaded + * and overlaid onto policy {@code jobOptions}. Auth wins on duplicate keys. Live {@code runJob} + * wiring is a follow-up. + */ +public final class MaintenanceAuthConfigs { + + /** Prefix for Gravitino control-plane client auth (overlaid onto {@code updaterOptions}). */ + public static final String GRAVITINO_AUTH_PREFIX = "gravitino.maintenance.gravitinoAuth."; + + /** + * Prefix for Iceberg REST / Spark client properties overlaid onto {@code jobOptions}. Suffixes + * are not enumerated; operators may pass arbitrary catalog auth keys such as {@code + * rest.auth.type}. + */ + public static final String IRC_AUTH_PREFIX = "gravitino.maintenance.ircAuth."; + + /** Auth mode: {@code none}, {@code simple}, {@code basic}, or {@code oauth}. */ + public static final String AUTH_TYPE_SUFFIX = "authType"; + + /** HTTP basic / simple user principal. */ + public static final String USERNAME_SUFFIX = "username"; + + /** HTTP basic password or secret reference. */ + public static final String PASSWORD_SUFFIX = "password"; + + /** OAuth2 token server base URI. */ + public static final String OAUTH_SERVER_URI_SUFFIX = "oauthServerUri"; + + /** OAuth2 token path relative to the server URI. */ + public static final String OAUTH_PATH_SUFFIX = "oauthPath"; + + /** OAuth2 client credential, often {@code client_id:client_secret}. */ + public static final String OAUTH_CREDENTIAL_SUFFIX = "oauthCredential"; + + /** OAuth2 scope. */ + public static final String OAUTH_SCOPE_SUFFIX = "oauthScope"; + + private static final String AUTH_TYPE_NONE = "none"; + private static final String AUTH_TYPE_SIMPLE = "simple"; + private static final String AUTH_TYPE_BASIC = "basic"; + private static final String AUTH_TYPE_OAUTH = "oauth"; + + private static final Set<String> SUPPORTED_AUTH_TYPES = + ImmutableSet.of(AUTH_TYPE_NONE, AUTH_TYPE_SIMPLE, AUTH_TYPE_BASIC, AUTH_TYPE_OAUTH); + + /** + * Maps camelCase {@code gravitino.conf} suffixes to {@code updaterOptions} / updater-option short + * names consumed by {@code GravitinoAuthSettings}. + */ + private static final Map<String, String> GRAVITINO_AUTH_TO_UPDATER_OPTIONS = + ImmutableMap.<String, String>builder() + .put(AUTH_TYPE_SUFFIX, "auth_type") + .put(USERNAME_SUFFIX, "username") + .put(PASSWORD_SUFFIX, "password") + .put(OAUTH_SERVER_URI_SUFFIX, "oauth_server_uri") + .put(OAUTH_PATH_SUFFIX, "oauth_path") + .put(OAUTH_CREDENTIAL_SUFFIX, "oauth_credential") + .put(OAUTH_SCOPE_SUFFIX, "oauth_scope") + .build(); + + public static final ConfigEntry<Optional<String>> GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_TYPE = + optionalAuth( + GRAVITINO_AUTH_PREFIX, + AUTH_TYPE_SUFFIX, + "TMS Gravitino client auth mode (none, simple, basic, or oauth)."); + + public static final ConfigEntry<Optional<String>> GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_USERNAME = + optionalAuth(GRAVITINO_AUTH_PREFIX, USERNAME_SUFFIX, "TMS Gravitino client user principal."); + + public static final ConfigEntry<Optional<String>> GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_PASSWORD = + optionalAuth( + GRAVITINO_AUTH_PREFIX, PASSWORD_SUFFIX, "TMS Gravitino client HTTP basic password."); + + public static final ConfigEntry<Optional<String>> + GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_OAUTH_SERVER_URI = + optionalAuth( + GRAVITINO_AUTH_PREFIX, + OAUTH_SERVER_URI_SUFFIX, + "TMS Gravitino client OAuth2 token server URI."); + + public static final ConfigEntry<Optional<String>> + GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_OAUTH_PATH = + optionalAuth( + GRAVITINO_AUTH_PREFIX, OAUTH_PATH_SUFFIX, "TMS Gravitino client OAuth2 token path."); + + public static final ConfigEntry<Optional<String>> + GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_OAUTH_CREDENTIAL = + optionalAuth( + GRAVITINO_AUTH_PREFIX, + OAUTH_CREDENTIAL_SUFFIX, + "TMS Gravitino client OAuth2 client credential."); + + public static final ConfigEntry<Optional<String>> + GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_OAUTH_SCOPE = + optionalAuth( + GRAVITINO_AUTH_PREFIX, OAUTH_SCOPE_SUFFIX, "TMS Gravitino client OAuth2 scope."); + + private MaintenanceAuthConfigs() {} + + /** + * Loads fixed Gravitino auth entries from server config and remaps camelCase conf suffixes to + * updater short names for {@code updaterOptions}. + * + * @param config gravitino server config + * @return unmodifiable map keyed by updater short names; empty when unset + */ + public static Map<String, String> loadGravitinoAuth(Config config) { + Map<String, String> loaded = loadAuthBag(config, GRAVITINO_AUTH_PREFIX, /*fixedSchema=*/ true); + if (loaded.isEmpty()) { + return Collections.emptyMap(); + } + validateGravitinoAuth(loaded); + Map<String, String> updaterOptions = new LinkedHashMap<>(); + GRAVITINO_AUTH_TO_UPDATER_OPTIONS.forEach( + (suffix, updateKey) -> { + String value = loaded.get(suffix); + if (value != null) { + updaterOptions.put(updateKey, value); + } + }); + return ImmutableMap.copyOf(updaterOptions); + } + + /** + * Loads arbitrary IRC / Spark property suffixes under {@link #IRC_AUTH_PREFIX}. + * + * @param config gravitino server config + * @return unmodifiable map; empty when unset + */ + public static Map<String, String> loadIrcAuth(Config config) { + return loadAuthBag(config, IRC_AUTH_PREFIX, /*fixedSchema=*/ false); + } + + /** + * Merges policy {@code jobOptions} with IRC auth passthrough; auth wins on duplicate keys. + * + * @param policyJobOptions options from the effective policy (may be null) + * @param ircAuth suffix-keyed IRC / Spark properties (may be null) + * @return merged job options for {@code runJob} + */ + public static Map<String, String> overlayJobOptions( + @Nullable Map<String, String> policyJobOptions, @Nullable Map<String, String> ircAuth) { + return overlay(policyJobOptions, ircAuth); + } + + /** + * Merges policy {@code updaterOptions} with Gravitino auth; auth wins on duplicate keys. + * + * @param policyUpdaterOptions options from the effective policy (may be null) + * @param gravitinoAuth updater-short-name Gravitino auth bag (may be null) + * @return merged updater options for {@code runJob} + */ + public static Map<String, String> overlayUpdaterOptions( + @Nullable Map<String, String> policyUpdaterOptions, + @Nullable Map<String, String> gravitinoAuth) { + return overlay(policyUpdaterOptions, gravitinoAuth); + } + + /** + * Loads IRC passthrough from {@code config} and overlays it onto policy {@code jobOptions}. + * + * @param config gravitino server config + * @param policyJobOptions options from the effective policy + * @return merged job options + */ + public static Map<String, String> overlayJobOptions( + Config config, @Nullable Map<String, String> policyJobOptions) { + return overlayJobOptions(policyJobOptions, loadIrcAuth(config)); + } + + /** + * Loads Gravitino auth from {@code config} and overlays it onto policy {@code updaterOptions}. + * + * @param config gravitino server config + * @param policyUpdaterOptions options from the effective policy + * @return merged updater options + */ + public static Map<String, String> overlayUpdaterOptions( + Config config, @Nullable Map<String, String> policyUpdaterOptions) { + return overlayUpdaterOptions(policyUpdaterOptions, loadGravitinoAuth(config)); + } + + private static Map<String, String> loadAuthBag( + Config config, String prefix, boolean fixedSchema) { + Map<String, String> loaded = config.getConfigsWithPrefix(prefix); + if (loaded.isEmpty()) { + return Collections.emptyMap(); + } + Map<String, String> sanitized = new LinkedHashMap<>(); + loaded.forEach( + (key, value) -> { + if (key == null || key.trim().isEmpty()) { + throw new IllegalArgumentException( + String.format("Configuration under %s must not use a blank suffix", prefix)); + } + String suffix = key.trim(); + if (fixedSchema && !GRAVITINO_AUTH_TO_UPDATER_OPTIONS.containsKey(suffix)) { + throw new IllegalArgumentException( + String.format( + "Unknown Gravitino auth configuration %s%s; supported suffixes are: %s", + prefix, suffix, String.join(", ", GRAVITINO_AUTH_TO_UPDATER_OPTIONS.keySet()))); + } + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException( + String.format("Configuration %s%s must have a non-empty value", prefix, suffix)); + } + sanitized.put(suffix, value.trim()); + }); + return ImmutableMap.copyOf(sanitized); + } + + private static void validateGravitinoAuth(Map<String, String> loaded) { + String authType = loaded.get(AUTH_TYPE_SUFFIX); + if (authType == null) { + throw new IllegalArgumentException( + String.format( + "Configuration %s%s is required when any Gravitino auth setting is present", + GRAVITINO_AUTH_PREFIX, AUTH_TYPE_SUFFIX)); + } + String normalized = authType.toLowerCase(Locale.ROOT); + if (!SUPPORTED_AUTH_TYPES.contains(normalized)) { + throw new IllegalArgumentException( + String.format( + "Unsupported %s%s value '%s'; supported values are: none, simple, basic, oauth", + GRAVITINO_AUTH_PREFIX, AUTH_TYPE_SUFFIX, authType)); + } + switch (normalized) { + case AUTH_TYPE_BASIC: + requireSuffix(loaded, USERNAME_SUFFIX, AUTH_TYPE_BASIC); + requireSuffix(loaded, PASSWORD_SUFFIX, AUTH_TYPE_BASIC); + break; + case AUTH_TYPE_OAUTH: + requireSuffix(loaded, OAUTH_SERVER_URI_SUFFIX, AUTH_TYPE_OAUTH); + requireSuffix(loaded, OAUTH_PATH_SUFFIX, AUTH_TYPE_OAUTH); + requireSuffix(loaded, OAUTH_CREDENTIAL_SUFFIX, AUTH_TYPE_OAUTH); + requireSuffix(loaded, OAUTH_SCOPE_SUFFIX, AUTH_TYPE_OAUTH); + break; + case AUTH_TYPE_NONE: + case AUTH_TYPE_SIMPLE: + default: + break; + } + } + + private static void requireSuffix(Map<String, String> loaded, String suffix, String authType) { + if (!loaded.containsKey(suffix)) { + throw new IllegalArgumentException( + String.format( + "Configuration %s%s is required when %s%s is '%s'", + GRAVITINO_AUTH_PREFIX, suffix, GRAVITINO_AUTH_PREFIX, AUTH_TYPE_SUFFIX, authType)); + } + } + + private static Map<String, String> overlay( + @Nullable Map<String, String> base, @Nullable Map<String, String> auth) { + if ((base == null || base.isEmpty()) && (auth == null || auth.isEmpty())) { + return Collections.emptyMap(); + } + Map<String, String> merged = new LinkedHashMap<>(); + if (base != null) { + merged.putAll(base); + } + if (auth != null) { + merged.putAll(auth); + } + return ImmutableMap.copyOf(merged); + } + + private static ConfigEntry<Optional<String>> optionalAuth( + String prefix, String suffix, String doc) { + return new ConfigBuilder(prefix + suffix) + .doc(doc) + .version(ConfigConstants.VERSION_2_0_0) + .stringConf() + .createWithOptional(); + } +} diff --git a/core/src/test/java/org/apache/gravitino/maintenance/policy/TestMaintenanceAuthConfigs.java b/core/src/test/java/org/apache/gravitino/maintenance/policy/TestMaintenanceAuthConfigs.java new file mode 100644 index 0000000000..fed3289e83 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/maintenance/policy/TestMaintenanceAuthConfigs.java @@ -0,0 +1,293 @@ +/* + * 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.maintenance.policy; + +import com.google.common.collect.ImmutableMap; +import java.util.Collections; +import java.util.Map; +import org.apache.gravitino.Config; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestMaintenanceAuthConfigs { + + @Test + void gravitinoAuthConfigKeysUseCamelCase() { + Assertions.assertEquals( + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + MaintenanceAuthConfigs.AUTH_TYPE_SUFFIX, + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_TYPE.getKey()); + Assertions.assertEquals( + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + MaintenanceAuthConfigs.OAUTH_PATH_SUFFIX, + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_OAUTH_PATH.getKey()); + Assertions.assertEquals( + "gravitino.maintenance.gravitinoAuth.authType", + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_TYPE.getKey()); + Assertions.assertEquals( + "gravitino.maintenance.gravitinoAuth.oauthServerUri", + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_OAUTH_SERVER_URI.getKey()); + } + + @Test + void optionalGravitinoAuthValuesLoadFromConfig() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of( + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_TYPE.getKey(), + "oauth", + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_OAUTH_PATH.getKey(), + "oauth2/token"), + key -> key.startsWith("gravitino.")); + + Assertions.assertEquals( + "oauth", + config.get(MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_TYPE).get()); + Assertions.assertEquals( + "oauth2/token", + config.get(MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_OAUTH_PATH).get()); + Assertions.assertTrue( + config.get(MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_USERNAME).isEmpty()); + } + + @Test + void loadGravitinoAuthMapsCamelCaseToUpdaterShortNames() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of( + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "authType", + "oauth", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "username", + "tms", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "password", + "secret", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "oauthServerUri", + "https://idp.example.com";, + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "oauthPath", + "oauth2/token", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "oauthCredential", + "id:secret", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "oauthScope", + "catalog"), + key -> key.startsWith("gravitino.")); + + Assertions.assertEquals( + ImmutableMap.<String, String>builder() + .put("auth_type", "oauth") + .put("username", "tms") + .put("password", "secret") + .put("oauth_server_uri", "https://idp.example.com";) + .put("oauth_path", "oauth2/token") + .put("oauth_credential", "id:secret") + .put("oauth_scope", "catalog") + .build(), + MaintenanceAuthConfigs.loadGravitinoAuth(config)); + } + + @Test + void loadIrcAuthPassesThroughArbitrarySparkKeys() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of( + MaintenanceAuthConfigs.IRC_AUTH_PREFIX + "rest.auth.type", + "oauth2", + MaintenanceAuthConfigs.IRC_AUTH_PREFIX + "rest.auth.oauth2.token", + "tok", + MaintenanceAuthConfigs.IRC_AUTH_PREFIX + "spark.executor.memory", + "4g"), + key -> key.startsWith("gravitino.")); + + Assertions.assertEquals( + ImmutableMap.of( + "rest.auth.type", + "oauth2", + "rest.auth.oauth2.token", + "tok", + "spark.executor.memory", + "4g"), + MaintenanceAuthConfigs.loadIrcAuth(config)); + } + + @Test + void overlayAuthWinsOnDuplicateKeys() { + Map<String, String> policyJob = ImmutableMap.of("uri", "http://irc";, "rest.auth.type", "basic"); + Map<String, String> policyUpdater = ImmutableMap.of("metalake", "ml", "auth_type", "none"); + Map<String, String> ircAuth = + ImmutableMap.of("rest.auth.type", "oauth2", "rest.auth.oauth2.token", "tok"); + Map<String, String> gravitinoAuth = ImmutableMap.of("auth_type", "oauth"); + + Map<String, String> job = MaintenanceAuthConfigs.overlayJobOptions(policyJob, ircAuth); + Map<String, String> updater = + MaintenanceAuthConfigs.overlayUpdaterOptions(policyUpdater, gravitinoAuth); + + Assertions.assertEquals("oauth2", job.get("rest.auth.type")); + Assertions.assertEquals("tok", job.get("rest.auth.oauth2.token")); + Assertions.assertEquals("http://irc";, job.get("uri")); + Assertions.assertEquals("oauth", updater.get("auth_type")); + Assertions.assertEquals("ml", updater.get("metalake")); + } + + @Test + void emptyAuthLeavesPolicyOnly() { + Assertions.assertEquals( + ImmutableMap.of("a", "1"), + MaintenanceAuthConfigs.overlayJobOptions( + ImmutableMap.of("a", "1"), Collections.emptyMap())); + Assertions.assertEquals( + ImmutableMap.of("b", "2"), + MaintenanceAuthConfigs.overlayUpdaterOptions( + ImmutableMap.of("b", "2"), Collections.emptyMap())); + } + + @Test + void overlayFromConfigLoadsAndMerges() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of( + MaintenanceAuthConfigs.IRC_AUTH_PREFIX + "rest.auth.type", + "basic", + MaintenanceAuthConfigs.IRC_AUTH_PREFIX + "rest.auth.basic.username", + "irc", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "authType", + "simple", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "username", + "tms"), + key -> key.startsWith("gravitino.")); + + Assertions.assertEquals( + ImmutableMap.of( + "uri", "http://policy";, "rest.auth.type", "basic", "rest.auth.basic.username", "irc"), + MaintenanceAuthConfigs.overlayJobOptions(config, ImmutableMap.of("uri", "http://policy";))); + Assertions.assertEquals( + ImmutableMap.of("metalake", "ml", "auth_type", "simple", "username", "tms"), + MaintenanceAuthConfigs.overlayUpdaterOptions(config, ImmutableMap.of("metalake", "ml"))); + } + + @Test + void emptyAuthValueRejected() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of(MaintenanceAuthConfigs.IRC_AUTH_PREFIX + "rest.auth.type", " "), + key -> key.startsWith("gravitino.")); + Assertions.assertThrows( + IllegalArgumentException.class, () -> MaintenanceAuthConfigs.loadIrcAuth(config)); + } + + @Test + void unknownGravitinoAuthSuffixRejected() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of( + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "authTyp", + "basic", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "username", + "tms"), + key -> key.startsWith("gravitino.")); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, () -> MaintenanceAuthConfigs.loadGravitinoAuth(config)); + Assertions.assertTrue(ex.getMessage().contains("Unknown Gravitino auth configuration")); + Assertions.assertTrue(ex.getMessage().contains("authTyp")); + } + + @Test + void blankGravitinoAuthSuffixRejected() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of(MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX, "none"), + key -> key.startsWith("gravitino.")); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, () -> MaintenanceAuthConfigs.loadGravitinoAuth(config)); + Assertions.assertTrue(ex.getMessage().contains("blank suffix")); + } + + @Test + void blankIrcAuthSuffixRejected() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of(MaintenanceAuthConfigs.IRC_AUTH_PREFIX, "none"), + key -> key.startsWith("gravitino.")); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, () -> MaintenanceAuthConfigs.loadIrcAuth(config)); + Assertions.assertTrue(ex.getMessage().contains("blank suffix")); + } + + @Test + void missingAuthTypeRejectedWhenOtherKeysPresent() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of(MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "username", "tms"), + key -> key.startsWith("gravitino.")); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, () -> MaintenanceAuthConfigs.loadGravitinoAuth(config)); + Assertions.assertTrue(ex.getMessage().contains("authType is required")); + } + + @Test + void unsupportedAuthTypeRejected() { + Config config = new Config(false) {}; + config.loadFromMap( + Map.of(MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "authType", "kerberos"), + key -> key.startsWith("gravitino.")); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, () -> MaintenanceAuthConfigs.loadGravitinoAuth(config)); + Assertions.assertTrue(ex.getMessage().contains("Unsupported")); + Assertions.assertTrue(ex.getMessage().contains("kerberos")); + } + + @Test + void basicAuthRequiresUsernameAndPassword() { + Config missingPassword = new Config(false) {}; + missingPassword.loadFromMap( + Map.of( + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "authType", + "basic", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "username", + "tms"), + key -> key.startsWith("gravitino.")); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> MaintenanceAuthConfigs.loadGravitinoAuth(missingPassword)); + Assertions.assertTrue(ex.getMessage().contains("password is required")); + } + + @Test + void oauthAuthRequiresTokenSettings() { + Config missingScope = new Config(false) {}; + missingScope.loadFromMap( + Map.of( + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "authType", + "oauth", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "oauthServerUri", + "https://idp.example.com";, + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "oauthPath", + "oauth2/token", + MaintenanceAuthConfigs.GRAVITINO_AUTH_PREFIX + "oauthCredential", + "id:secret"), + key -> key.startsWith("gravitino.")); + IllegalArgumentException ex = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> MaintenanceAuthConfigs.loadGravitinoAuth(missingScope)); + Assertions.assertTrue(ex.getMessage().contains("oauthScope is required")); + } +} diff --git a/docs/table-maintenance-service/optimizer-configuration.md b/docs/table-maintenance-service/optimizer-configuration.md index a7618665b0..5e8f04ca4a 100644 --- a/docs/table-maintenance-service/optimizer-configuration.md +++ b/docs/table-maintenance-service/optimizer-configuration.md @@ -9,11 +9,11 @@ license: "This software is licensed under the Apache License version 2." Three layers of configuration apply, and they are set in different places for different lifetimes. Server configuration governs how jobs run at all, CLI configuration governs how the commands reach Gravitino, and `jobConf` governs a single job submission. -| Layer | Where it lives | Lifetime | -|---------------------|----------------------------------|---------------------| -| Server | `gravitino.conf` | Until server restart | -| CLI | `conf/gravitino-optimizer.conf` | Per command | -| Job submission | `jobConf` in the request body | One job run | +| Layer | Where it lives | Lifetime | +| -------------- | ------------------------------- | -------------------- | +| Server | `gravitino.conf` | Until server restart | +| CLI | `conf/gravitino-optimizer.conf` | Per command | +| Job submission | `jobConf` in the request body | One job run | ## Server Configuration @@ -27,6 +27,122 @@ gravitino.jobExecutor.local.sparkHome=/path/to/spark `gravitino.job.statusPullIntervalInMs` defaults to five minutes. Job status is polled rather than pushed, so REST status can lag the real Spark process by a full interval, which makes a working job look hung. Lower it to `10000` for local work and restart the server. +### TMS cluster-static client auth + +For in-process Table Maintenance Service (TMS), client credentials are **cluster-static**: set them under +the prefixes below in `gravitino.conf`. Do not store passwords, OAuth secrets, or keytab paths in policy +content. Values may be plaintext or secret-manager references when the deployment supports them. + +At `runJob`, `gravitino.maintenance.gravitinoAuth.*` overlays policy `updaterOptions`, and +`gravitino.maintenance.ircAuth.*` overlays policy `jobOptions`. Auth wins on duplicate keys. The +live scheduler submit call site is wired in a follow-up; `MaintenanceAuthConfigs` exposes the load +and overlay helpers (`overlayUpdaterOptions` / `overlayJobOptions`) for that path. + +#### Gravitino client auth (`gravitino.maintenance.gravitinoAuth.`) — fixed schema + +`gravitino.conf` uses **camelCase** suffixes. At load time they are remapped to built-in +`--updater-options` / `GravitinoAuthSettings` short names (`auth_type`, `oauth_server_uri`, …) +before overlay onto policy `updaterOptions`. Only the suffixes in the table below are accepted; +unknown or blank suffixes fail fast when the server loads config. Supported `authType` values are +exactly four: **`none`**, **`simple`**, **`basic`**, **`oauth`**. When any Gravitino auth key is +set, `authType` is required, and `basic` / `oauth` also require their mode-specific fields below. + +| Conf suffix | `updaterOptions` key | Used when `authType` is | Meaning | +| ----------------- | -------------------- | ----------------------- | ------------------------------------------- | +| `authType` | `auth_type` | all modes | `none`, `simple`, `basic`, or `oauth` | +| `username` | `username` | `simple` / `basic` | User principal (`simple` optional; `basic` required) | +| `password` | `password` | `basic` | Password or secret reference (required) | +| `oauthServerUri` | `oauth_server_uri` | `oauth` | OAuth2 token server base URI (required) | +| `oauthPath` | `oauth_path` | `oauth` | Token path (for example `oauth2/token`) (required) | +| `oauthCredential` | `oauth_credential` | `oauth` | Client credential, often `client_id:secret` (required) | +| `oauthScope` | `oauth_scope` | `oauth` | Token scope (IdP-specific) (required) | + +**1. `none`** — Gravitino authorization disabled; omit the prefix or set: + +```properties +gravitino.maintenance.gravitinoAuth.authType = none +``` + +**2. `simple`**: + +```properties +gravitino.maintenance.gravitinoAuth.authType = simple +gravitino.maintenance.gravitinoAuth.username = {username} +``` + +**3. `basic`**: + +```properties +gravitino.maintenance.gravitinoAuth.authType = basic +gravitino.maintenance.gravitinoAuth.username = {username} +gravitino.maintenance.gravitinoAuth.password = {password} +``` + +**4. `oauth`**: + +```properties +gravitino.maintenance.gravitinoAuth.authType = oauth +gravitino.maintenance.gravitinoAuth.oauthServerUri = https://idp.example.com/realms/demo +gravitino.maintenance.gravitinoAuth.oauthPath = protocol/openid-connect/token +gravitino.maintenance.gravitinoAuth.oauthCredential = {client_id}:{client_secret} +gravitino.maintenance.gravitinoAuth.oauthScope = openid profile email +``` + +#### Iceberg REST / Spark auth (`gravitino.maintenance.ircAuth.`) — arbitrary passthrough + +Open prefix for Spark jobs that talk to Iceberg REST: any non-empty suffix under the prefix is +stripped and overlaid onto `jobOptions`. Prefer Iceberg catalog properties (`rest.auth.*`, and +related OAuth keys such as `credential` / `oauth2-server-uri` / `scope`). + +Gravitino Iceberg REST (IRC) shares `gravitino.authenticators` with the Gravitino server +(`simple`, `basic`, `oauth`, `kerberos`), but **IRC does not support Kerberos** (see +[How to authenticate](../security/how-to-authenticate.md#kerberos-mode)). For Spark/Flink clients, +map the IRC authenticator to Iceberg `rest.auth.type` as below. Iceberg also defines `sigv4` and +`google`; those are **not** supported when the target is Gravitino IRC. + +| Iceberg `rest.auth.type` | Gravitino IRC | Notes | +| ------------------------ | ----------------- | ----------------------------------------------------------- | +| `none` | Supported | IRC `simple` / auth disabled; omit the prefix or set `none` | +| `basic` | Supported | IRC `basic` (local users) | +| `oauth2` | Supported | IRC `oauth` | +| `sigv4` | Not supported yet | Iceberg AWS signing; not a Gravitino IRC authenticator | +| `google` | Not supported yet | Iceberg Google REST auth; not a Gravitino IRC authenticator | +| *(Kerberos)* | Not supported yet | Documented: IRC does not support Kerberos | + +**1. `none`** — IRC `simple` / no REST auth (default if the prefix is omitted): + +```properties +gravitino.maintenance.ircAuth.rest.auth.type = none +``` + +**2. `basic`** — IRC `basic`: + +```properties +gravitino.maintenance.ircAuth.rest.auth.type = basic +gravitino.maintenance.ircAuth.rest.auth.basic.username = {username} +gravitino.maintenance.ircAuth.rest.auth.basic.password = {password} +``` + +**3. `oauth2`** — IRC `oauth`, bearer token: + +```properties +gravitino.maintenance.ircAuth.rest.auth.type = oauth2 +gravitino.maintenance.ircAuth.rest.auth.oauth2.token = <your-token> +``` + +`oauth2` with client credentials (Iceberg catalog property names, not Gravitino updater short names): + +```properties +gravitino.maintenance.ircAuth.rest.auth.type = oauth2 +gravitino.maintenance.ircAuth.credential = {client_id}:{client_secret} +gravitino.maintenance.ircAuth.oauth2-server-uri = https://idp.example.com/realms/demo/protocol/openid-connect/token +gravitino.maintenance.ircAuth.scope = catalog +``` + +**4. `sigv4`** — **not supported yet** for Gravitino IRC. + +**5. `google`** — **not supported yet** for Gravitino IRC. + ## CLI Configuration The CLI needs to know where Gravitino is and which components to use. This is a minimal working file for `submit-strategy-jobs`. @@ -64,11 +180,11 @@ When the Gravitino server has authentication enabled, `builtin-iceberg-update-st credentials in `--updater-options` / `updater_options` for the Gravitino client (statistics updater and `submit-update-stats-job` `runJob` calls): -| `auth_type` | Fields | -|------------------|------------------------------------------------------------------------------------------------| -| `none` (default) | (none) | -| `simple` | `username` (optional) | -| `basic` | `username`, `password` | +| `auth_type` | Fields | +| ---------------- | -------------------------------------------------------------------------------------------- | +| `none` (default) | (none) | +| `simple` | `username` (optional) | +| `basic` | `username`, `password` | | `oauth` | client-credentials only: `oauth_server_uri`, `oauth_path`, `oauth_credential`, `oauth_scope` | Iceberg REST catalog authentication is separate: set `rest.auth.*` in `spark-conf` for any built-in @@ -180,14 +296,14 @@ Four things are worth confirming before assuming a configuration problem is a co Submit `builtin-iceberg-remove-orphan-files` with the same Spark and catalog settings described above. Its job-specific `jobConf` keys are: -| Key | Meaning | Default | -| ------------------ | ------------------------------------------------------------------------------------- | -------------------------------- | -| `catalog_name` | Iceberg catalog registered in Spark | Required | -| `table_identifier` | Table identifier within the catalog, such as `db.sample` | Required | -| `older_than` | Timestamp (explicit offset or Spark session time zone); must be at least 24 hours old | Three days ago (Iceberg default) | -| `location` | Scan only this directory within the table's storage location | Table location | -| `dry_run` | `true` logs candidate paths without deleting; `false` deletes | `false` | -| `spark_conf` | JSON string containing custom Spark settings, including the Iceberg runtime if needed | None | +| Key | Meaning | Default | +| ------------------ | --------------------------------------------------------------------------------------- | -------------------------------- | +| `catalog_name` | Iceberg catalog registered in Spark | Required | +| `table_identifier` | Table identifier within the catalog, such as `db.sample` | Required | +| `older_than` | Timestamp (explicit offset or Spark session time zone); must be at least 24 hours old | Three days ago (Iceberg default) | +| `location` | Scan only this directory within the table's storage location | Table location | +| `dry_run` | `true` logs candidate paths without deleting; `false` deletes | `false` | +| `spark_conf` | JSON string containing custom Spark settings, including the Iceberg runtime if needed | None | For direct template submission, supply `older_than: ""` and `location: ""` to use the defaults, `dry_run: "false"` (or `"true"` to preview), and @@ -270,11 +386,18 @@ Create a policy through `POST /api/metalakes/{metalake}/policies`: "enabled": true, "content": { "olderThanDays": 3, - "dryRun": true + "dryRun": true, + "timeTrigger": "0 0 3 * * ?", + "minIntervalMs": 86400000 } } ``` +Optional TMS peer fields on content (`timeTrigger`, `commitTrigger`, `minIntervalMs`, `jobOptions`, +`updaterOptions`) are shared with compaction and other maintenance policy types. `commitTrigger` is +not allowed for orphan cleanup. See [Iceberg compaction policy](../iceberg-compaction-policy.md) for +peer-field semantics. + Associate the policy with a tag and attach that tag to the target table, schema, or catalog, following the [policy setup walkthrough](./optimizer.md). `olderThanDays` defaults to 3 and must be between 1 and 36500 inclusive (approximately 100 years). This bound rejects impractical retention values before a policy is stored. `dryRun` defaults to false.
