This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 48572b5eb7 [#13613] feat(k8s-job-executor): Add the k8s job executor
module with its configurations and Kubernetes connection (part 1/3) (#13724)
48572b5eb7 is described below
commit 48572b5eb7a9a0ae4d62ce171242a39374904d70
Author: Jerry Shao <[email protected]>
AuthorDate: Tue Oct 13 14:25:25 2026 +0800
[#13613] feat(k8s-job-executor): Add the k8s job executor module with its
configurations and Kubernetes connection (part 1/3) (#13724)
### What changes were proposed in this pull request?
This is part 1 of 3 of #13613, which adds a job executor named `k8s`
that runs Spark jobs on Kubernetes. It was split from #13700 to keep
each part reviewable.
This part adds the module `plugins/k8s-job-executor` and the classes
that don't depend on how jobs run:
- **Build**: a shaded jar with fabric8 kubernetes-client 7.2.0 and its
JDK HTTP client. Jackson and the YAML libraries are relocated to
`org.apache.gravitino.k8s.shaded`. `LICENSE.bin` lists the bundled
dependencies.
- **`K8sJobExecutorConfigs`**: the configurations of the executor, under
`gravitino.jobExecutor.k8s.`.
- **`K8sClientUtils`**: creates the Kubernetes client from a kubeconfig
and context, or from an API server URL with a CA certificate and a token
file. The token file is read again when it changes. Without either,
fabric8's default lookup applies, which covers running inside the job
cluster.
- **`K8sJobExecutionId`**: the job execution id,
`<cluster>/<kind>/<namespace>/<name>`.
- **`K8sJobResourceUtils`**: the names, labels and annotations of the
Kubernetes resources created for the jobs.
Nothing uses these classes yet. Part 2 (#13725) adds the
SparkApplication mapping, and part 3 (#13700) adds the executor itself.
### Why are the changes needed?
The only built-in job executor, `local`, runs jobs as child processes of
the Gravitino server, which isn't suitable for production. #13554 adds
an executor that runs jobs on Kubernetes and keeps all the job state
there.
Part of: #13613
### Does this PR introduce _any_ user-facing change?
No. The configurations only take effect once the executor is added in
part 3 and registered in #13616.
### How was this patch tested?
Unit tests in `plugins/k8s-job-executor`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
LICENSE.bin | 5 +
gradle/libs.versions.toml | 5 +
plugins/k8s-job-executor/build.gradle.kts | 85 ++++
.../apache/gravitino/job/k8s/K8sClientUtils.java | 288 ++++++++++++
.../gravitino/job/k8s/K8sJobExecutionId.java | 132 ++++++
.../gravitino/job/k8s/K8sJobExecutorConfigs.java | 519 +++++++++++++++++++++
.../gravitino/job/k8s/K8sJobResourceUtils.java | 106 +++++
.../gravitino/job/k8s/K8sKubeconfigUtils.java | 115 +++++
.../gravitino/job/k8s/TestK8sClientUtils.java | 451 ++++++++++++++++++
.../gravitino/job/k8s/TestK8sJobExecutionId.java | 76 +++
.../job/k8s/TestK8sJobExecutorConfigs.java | 197 ++++++++
.../gravitino/job/k8s/TestK8sJobResourceUtils.java | 85 ++++
settings.gradle.kts | 1 +
13 files changed, 2065 insertions(+)
diff --git a/LICENSE.bin b/LICENSE.bin
index a95017bd21..9da550578c 100644
--- a/LICENSE.bin
+++ b/LICENSE.bin
@@ -222,6 +222,7 @@
Jackson Core
Jackson Databind
Jackson Data Mapper
+ Jackson Dataformat YAML
Jackson Datatype Guava
Jackson Datatype JDK8
Jackson Datatype Joda
@@ -356,6 +357,7 @@
jmxutils
SQLite JDBC
SnakeYAML
+ SnakeYAML Engine
StAX API
Tomcat Jasper Compiler
Apache Xerces Java
@@ -406,6 +408,9 @@
IPAddress
Aliyun SDK OSS
QLExpress
+ Fabric8 Kubernetes Client
+ Fabric8 Kubernetes Model
+ Fabric8 zjsonpatch
This product bundles various third-party components also under the
Apache Software Foundation License 1.1
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index c491cc5134..cfe5bb7ce6 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -55,6 +55,8 @@ httpclient5 = "5.4.4"
httpcore5 = "5.3.4"
micrometer-core = "1.12.2"
mockserver = "5.15.0"
+# 7.3.x moves to Jackson 2.19, which isn't audited in
dev/release/maven/dependencies.txt yet.
+fabric8-kubernetes = "7.2.0"
commons-configuration2 = "2.10.1"
commons-csv = "1.12.0"
commons-lang3 = "3.14.0"
@@ -271,6 +273,9 @@ micrometer-core = { group = "io.micrometer", name =
"micrometer-core", version.r
mockserver-netty = { group = "org.mock-server", name = "mockserver-netty",
version.ref = "mockserver" }
mockserver-client-java = { group = "org.mock-server", name =
"mockserver-client-java", version.ref = "mockserver" }
+fabric8-kubernetes-client = { group = "io.fabric8", name =
"kubernetes-client", version.ref = "fabric8-kubernetes" }
+fabric8-kubernetes-httpclient-jdk = { group = "io.fabric8", name =
"kubernetes-httpclient-jdk", version.ref = "fabric8-kubernetes" }
+fabric8-kubernetes-server-mock = { group = "io.fabric8", name =
"kubernetes-server-mock", version.ref = "fabric8-kubernetes" }
commons-csv = { group = "org.apache.commons", name = "commons-csv",
version.ref = "commons-csv" }
commons-lang = { group = "commons-lang", name = "commons-lang", version.ref =
"commons-lang" }
commons-lang3 = { group = "org.apache.commons", name = "commons-lang3",
version.ref = "commons-lang3" }
diff --git a/plugins/k8s-job-executor/build.gradle.kts
b/plugins/k8s-job-executor/build.gradle.kts
new file mode 100644
index 0000000000..0005b4832a
--- /dev/null
+++ b/plugins/k8s-job-executor/build.gradle.kts
@@ -0,0 +1,85 @@
+/*
+ * 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 com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+
+plugins {
+ id("java")
+ id("idea")
+ alias(libs.plugins.shadow)
+}
+
+dependencies {
+ annotationProcessor(libs.lombok)
+
+ // Provided by the Gravitino server at runtime, never bundled into the
shaded jar.
+ compileOnly(project(":api"))
+ compileOnly(project(":common"))
+ compileOnly(project(":core"))
+ compileOnly(libs.commons.lang3)
+ compileOnly(libs.guava)
+ compileOnly(libs.lombok)
+ compileOnly(libs.slf4j.api)
+
+ // Bundled and relocated, as fabric8 needs a newer Jackson than the
Gravitino server ships.
+ implementation(libs.fabric8.kubernetes.client) {
+ exclude(group = "io.fabric8", module = "kubernetes-httpclient-okhttp")
+ exclude(group = "io.fabric8", module = "kubernetes-httpclient-vertx")
+ }
+ implementation(libs.fabric8.kubernetes.httpclient.jdk)
+
+ testImplementation(project(":api"))
+ testImplementation(project(":common"))
+ testImplementation(project(":core"))
+ testImplementation(libs.commons.lang3)
+ testImplementation(libs.fabric8.kubernetes.server.mock)
+ testImplementation(libs.guava)
+ testImplementation(libs.junit.jupiter.api)
+ testImplementation(libs.junit.jupiter.params)
+
+ testRuntimeOnly(libs.junit.jupiter.engine)
+}
+
+tasks.withType(ShadowJar::class.java) {
+ isZip64 = true
+ configurations = listOf(project.configurations.runtimeClasspath.get())
+ archiveClassifier.set("")
+ // fabric8 loads its HTTP client and resource handlers through ServiceLoader.
+ mergeServiceFiles()
+
+ exclude("module-info.class")
+ exclude("META-INF/versions/**")
+ // Leftovers of the fabric8 build, of no use at runtime.
+ exclude("META-INF/jandex.idx")
+ exclude("manifest.vm")
+
+ dependencies {
+ exclude(dependency("org.slf4j:.*"))
+ }
+
+ // Use a prefix of its own: the built-in jobs jar already relocates Jackson
under
+ // org.apache.gravitino.shaded.
+ relocate("com.fasterxml.jackson",
"org.apache.gravitino.k8s.shaded.com.fasterxml.jackson")
+ relocate("org.snakeyaml.engine",
"org.apache.gravitino.k8s.shaded.org.snakeyaml.engine")
+ relocate("org.yaml.snakeyaml",
"org.apache.gravitino.k8s.shaded.org.yaml.snakeyaml")
+}
+
+tasks.jar {
+ dependsOn(tasks.named("shadowJar"))
+ archiveClassifier.set("empty")
+}
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java
new file mode 100644
index 0000000000..2e765c8a10
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java
@@ -0,0 +1,288 @@
+/*
+ * 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.job.k8s;
+
+import com.google.common.base.Preconditions;
+import io.fabric8.kubernetes.api.model.NamedContext;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.ConfigBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import io.fabric8.kubernetes.client.OAuthTokenProvider;
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.attribute.FileTime;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Creates the Kubernetes client of the job cluster from the job executor
configurations. */
+public final class K8sClientUtils {
+
+ /**
+ * An {@link OAuthTokenProvider} that reads the token from a file, and reads
it again whenever the
+ * file changes, so that a rotated token is picked up without restarting the
server.
+ *
+ * <p>It doesn't refresh the token itself, whoever provides the file does.
For a file mounted from
+ * a Secret, that is whatever updates the Secret, and the kubelet then
updates the mounted file,
+ * unless it is mounted with a {@code subPath}. For a projected service
account token, the kubelet
+ * rotates it by itself.
+ */
+ static final class FileTokenProvider implements OAuthTokenProvider {
+
+ // How often a token file that stays empty or unreadable is reported.
+ private static final long FAILURE_LOG_INTERVAL_NANOS =
TimeUnit.MINUTES.toNanos(5);
+
+ private final Path tokenFile;
+
+ private FileTime lastModified;
+
+ private String token;
+
+ private boolean failing;
+
+ private long failureLoggedAtNanos;
+
+ FileTokenProvider(Path tokenFile) {
+ this.tokenFile = tokenFile;
+ }
+
+ @Override
+ public synchronized String getToken() {
+ try {
+ FileTime modified = Files.getLastModifiedTime(tokenFile);
+ if (token == null || !modified.equals(lastModified)) {
+ String content = new String(Files.readAllBytes(tokenFile),
StandardCharsets.UTF_8).trim();
+ if (StringUtils.isBlank(content)) {
+ if (token == null) {
+ throw new IllegalStateException("The token file " + tokenFile +
" is empty");
+ }
+ // The file is being rewritten. Its modification time isn't
remembered: the new token
+ // may be written within the same tick of the file system's clock,
and would then
+ // never be read.
+ return keepToken("is empty", null);
+ }
+ token = content;
+ lastModified = modified;
+ }
+ } catch (IOException e) {
+ if (token == null) {
+ throw new UncheckedIOException("Failed to read the token file " +
tokenFile, e);
+ }
+ // For example, the file is being replaced.
+ return keepToken("can't be read", e);
+ }
+ failing = false;
+ return token;
+ }
+
+ /**
+ * Keeps the token read before while the file has no token, checking the
file again on the next
+ * request. The problem is reported when it starts, and then at an
interval rather than on every
+ * request, so that a file that stays broken remains visible without
flooding the log.
+ */
+ private String keepToken(String problem, @Nullable IOException cause) {
+ long now = System.nanoTime();
+ if (!failing || now - failureLoggedAtNanos >=
FAILURE_LOG_INTERVAL_NANOS) {
+ LOG.warn(
+ "The token file {} {}, keeping the token read before. Requests
will fail once that"
+ + " token expires",
+ tokenFile,
+ problem,
+ cause);
+ failing = true;
+ failureLoggedAtNanos = now;
+ }
+ return token;
+ }
+ }
+
+ private static final Logger LOG =
LoggerFactory.getLogger(K8sClientUtils.class);
+
+ private static final Pattern SERVER_SCHEME =
+ Pattern.compile("^https?://", Pattern.CASE_INSENSITIVE);
+
+ private K8sClientUtils() {}
+
+ /**
+ * Creates the Kubernetes client of the job cluster.
+ *
+ * @param configs the job executor configurations
+ * @return the Kubernetes client
+ */
+ public static KubernetesClient createClient(K8sJobExecutorConfigs configs) {
+ return new
KubernetesClientBuilder().withConfig(createClientConfig(configs)).build();
+ }
+
+ /**
+ * Creates the Kubernetes client configuration of the job cluster. It is, by
precedence:
+ *
+ * <ul>
+ * <li>the API server URL, with its CA certificate and token file, if the
master URL is set;
+ * <li>the given kubeconfig file and context, if the kubeconfig is set;
+ * <li>fabric8's default lookup otherwise, that is the {@code KUBECONFIG}
environment variable
+ * or {@code ~/.kube/config} with the given context, then the
in-cluster service account.
+ * </ul>
+ *
+ * <p>In every case the default namespace of the client is the namespace of
the job executor
+ * configurations, rather than the one of the kubeconfig context or of the
pod Gravitino runs in.
+ * The connection timeout, the request timeout and the number of request
retries are the ones of
+ * the job executor configurations too. They bound how long a call to the
client waits while the
+ * job cluster is down: each of the 1 + retries attempts waits for the
connection timeout if the
+ * API server can't be reached, and for the request timeout if it accepts
the request but never
+ * answers. With the defaults that is about 15 and 30 seconds.
+ *
+ * @param configs the job executor configurations
+ * @return the Kubernetes client configuration
+ * @throws IllegalArgumentException if a configured file doesn't exist, the
token file has no
+ * token, the kubeconfig has no context that selects a cluster with a
server, or the default
+ * lookup finds no cluster at all
+ */
+ static Config createClientConfig(K8sJobExecutorConfigs configs) {
+ Config config = createConnectionConfig(configs);
+ config.setNamespace(configs.namespace());
+ // The client would otherwise try a request again 10 times, which holds a
status query or a
+ // job submission for about two minutes while the job cluster can't be
reached.
+ config.setConnectionTimeout(configs.connectionTimeoutMs());
+ config.setRequestTimeout(configs.requestTimeoutMs());
+ config.setRequestRetryBackoffLimit(configs.requestRetries());
+ return config;
+ }
+
+ private static Config createConnectionConfig(K8sJobExecutorConfigs configs) {
+ if (configs.masterUrl() != null) {
+ ConfigBuilder builder = new
ConfigBuilder(Config.empty()).withMasterUrl(configs.masterUrl());
+ if (configs.caCertFile() != null) {
+ checkFileExists(configs.caCertFile(),
K8sJobExecutorConfigs.CA_CERT_FILE);
+ builder.withCaCertFile(configs.caCertFile());
+ }
+ if (configs.tokenFile() != null) {
+ checkFileExists(configs.tokenFile(), K8sJobExecutorConfigs.TOKEN_FILE);
+ FileTokenProvider tokenProvider = new
FileTokenProvider(Paths.get(configs.tokenFile()));
+ // Read the token once now, so that a file without a token fails the
startup rather than
+ // the first request.
+ try {
+ tokenProvider.getToken();
+ } catch (IllegalStateException | UncheckedIOException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "No token can be read from the file %s set by %s",
+ configs.tokenFile(), K8sJobExecutorConfigs.TOKEN_FILE),
+ e);
+ }
+ builder.withOauthTokenProvider(tokenProvider);
+ }
+ return builder.build();
+ }
+
+ Config config;
+ if (configs.kubeconfig() != null) {
+ checkFileExists(configs.kubeconfig(), K8sJobExecutorConfigs.KUBECONFIG);
+ File kubeconfig = new File(configs.kubeconfig());
+ String server = K8sKubeconfigUtils.getSelectedServer(kubeconfig,
configs.context());
+ config = Config.fromKubeconfig(configs.context(), kubeconfig);
+ // The client reads the file itself, so make sure it ended up with the
server found above.
+ // This is a safety net with no known input that triggers it: it only
fails if the client
+ // reads a kubeconfig differently from the check above, or the file
changes in between.
+ Preconditions.checkArgument(
+ getHostAndPath(server).equals(getHostAndPath(config.getMasterUrl())),
+ "The kubeconfig %s set by %s selects the server %s, but the client
would use %s",
+ configs.kubeconfig(),
+ K8sJobExecutorConfigs.KUBECONFIG,
+ server,
+ config.getMasterUrl());
+ } else {
+ config = Config.autoConfigure(configs.context());
+ checkDefaultLookup(config, configs);
+ }
+ return config;
+ }
+
+ /**
+ * Checks the client configuration that the default lookup found. The lookup
may merge several
+ * kubeconfig files, so they aren't checked one by one like a configured
kubeconfig. Instead, the
+ * result must not be one of those where the client silently stays on its
default API server, the
+ * one of the cluster Gravitino runs in, although a kubeconfig was found.
+ */
+ private static void checkDefaultLookup(Config config, K8sJobExecutorConfigs
configs) {
+ NamedContext current = config.getCurrentContext();
+ String currentName = current == null ? null : current.getName();
+ // The client falls back to the current context if the given one doesn't
exist. Checked
+ // first, so that the error is about the context that was asked for.
+ Preconditions.checkArgument(
+ configs.context() == null || configs.context().equals(currentName),
+ "The context %s set by %s doesn't exist in the kubeconfig found by the
default lookup",
+ configs.context(),
+ K8sJobExecutorConfigs.CONTEXT);
+
+ // A kubeconfig with contexts was found, but none of them is selected.
Without any
+ // kubeconfig, as in a pod, there are no contexts and the service account
is used.
+ Preconditions.checkArgument(
+ current != null || config.getContexts() == null ||
config.getContexts().isEmpty(),
+ "The kubeconfig found by the default lookup has no current context,
set its"
+ + " current-context or %s",
+ K8sJobExecutorConfigs.CONTEXT);
+
+ // Both a context that selects a cluster with a server, and the
environment of a pod, move
+ // the client off its default API server. If it is still there, the lookup
found neither.
+ boolean foundServer =
+
!getHostAndPath(Config.DEFAULT_MASTER_URL).equals(getHostAndPath(config.getMasterUrl()));
+ Preconditions.checkArgument(
+ foundServer || current == null,
+ "The context %s of the kubeconfig found by the default lookup selects
no cluster with a"
+ + " server. Fix that kubeconfig, or set %s or %s. To use the API
server %s itself,"
+ + " set it by %s",
+ currentName,
+ K8sJobExecutorConfigs.KUBECONFIG,
+ K8sJobExecutorConfigs.MASTER_URL,
+ Config.DEFAULT_MASTER_URL,
+ K8sJobExecutorConfigs.MASTER_URL);
+ Preconditions.checkArgument(
+ foundServer,
+ "No Kubernetes cluster is configured: the default lookup found no
kubeconfig, and"
+ + " Gravitino doesn't run in a pod. Set %s or %s. To use the API
server %s itself,"
+ + " set it by %s",
+ K8sJobExecutorConfigs.KUBECONFIG,
+ K8sJobExecutorConfigs.MASTER_URL,
+ Config.DEFAULT_MASTER_URL,
+ K8sJobExecutorConfigs.MASTER_URL);
+ }
+
+ /**
+ * Returns the URL of an API server without its scheme and trailing slashes,
to compare the server
+ * of a kubeconfig with the one the client uses. The scheme is left out
because the client picks
+ * it for a server without one, by probing whether the server speaks HTTPS.
+ */
+ private static String getHostAndPath(String server) {
+ return
SERVER_SCHEME.matcher(server.trim()).replaceFirst("").replaceAll("/+$", "");
+ }
+
+ private static void checkFileExists(String path, String key) {
+ Preconditions.checkArgument(
+ new File(path).isFile(), "The file %s set by %s doesn't exist", path,
key);
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutionId.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutionId.java
new file mode 100644
index 0000000000..b42cd89756
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutionId.java
@@ -0,0 +1,132 @@
+/*
+ * 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.job.k8s;
+
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.experimental.Accessors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.exceptions.NoSuchJobException;
+
+/**
+ * The job execution id of the {@code k8s} job executor, {@code
+ * <cluster>/<kind>/<namespace>/<name>}. It locates the Kubernetes resource of
a job: the cluster it
+ * runs in, the kind of the resource, and the resource's namespace and name.
+ */
+@Getter
+@Accessors(fluent = true)
+@EqualsAndHashCode
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public final class K8sJobExecutionId {
+
+ /** The kind of the Kubernetes resource a job runs as. */
+ @Getter
+ @Accessors(fluent = true)
+ @AllArgsConstructor(access = AccessLevel.PRIVATE)
+ public enum Kind {
+ /** A Spark job, run as a SparkApplication of the Spark operator. */
+ SPARK_APPLICATION("sparkapp");
+
+ /** The kind in the job execution id. */
+ private final String id;
+
+ /**
+ * Returns the kind of the given id in a job execution id.
+ *
+ * @param id the kind in the job execution id
+ * @return the kind
+ * @throws IllegalArgumentException if the kind is unknown
+ */
+ public static Kind fromId(String id) {
+ return Arrays.stream(values())
+ .filter(kind -> kind.id.equals(id))
+ .findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("Unknown kind of a
job: " + id));
+ }
+ }
+
+ private static final String SEPARATOR = "/";
+
+ /** The cluster the resource is in. */
+ private final String cluster;
+
+ /** The kind of the resource. */
+ private final Kind kind;
+
+ /** The namespace of the resource. */
+ private final String namespace;
+
+ /** The name of the resource. */
+ private final String name;
+
+ /**
+ * Creates the job execution id of a Kubernetes resource.
+ *
+ * @param cluster the cluster the resource is in
+ * @param kind the kind of the resource
+ * @param namespace the namespace of the resource
+ * @param name the name of the resource
+ * @return the job execution id
+ */
+ public static K8sJobExecutionId of(String cluster, Kind kind, String
namespace, String name) {
+ Preconditions.checkArgument(kind != null, "kind must not be null");
+ for (String part : new String[] {cluster, namespace, name}) {
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(part) && !part.contains(SEPARATOR),
+ "Invalid part of a job execution id: %s",
+ part);
+ }
+ return new K8sJobExecutionId(cluster, kind, namespace, name);
+ }
+
+ /**
+ * Parses a job execution id.
+ *
+ * @param jobExecutionId the job execution id
+ * @return the parsed job execution id
+ * @throws NoSuchJobException if the job execution id is malformed or of an
unknown kind, so no
+ * such job exists
+ */
+ public static K8sJobExecutionId parse(String jobExecutionId) throws
NoSuchJobException {
+ String[] parts =
StringUtils.defaultString(jobExecutionId).split(SEPARATOR, -1);
+ if (parts.length != 4 || StringUtils.isAnyBlank(parts)) {
+ throw new NoSuchJobException(
+ "Invalid job execution id %s of the k8s job executor, expected"
+ + " <cluster>/<kind>/<namespace>/<name>",
+ jobExecutionId);
+ }
+
+ Kind kind;
+ try {
+ kind = Kind.fromId(parts[1]);
+ } catch (IllegalArgumentException e) {
+ throw new NoSuchJobException(e, "Unknown kind %s of job %s", parts[1],
jobExecutionId);
+ }
+ return new K8sJobExecutionId(parts[0], kind, parts[2], parts[3]);
+ }
+
+ @Override
+ public String toString() {
+ return String.join(SEPARATOR, cluster, kind.id(), namespace, name);
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
new file mode 100644
index 0000000000..306322a771
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java
@@ -0,0 +1,519 @@
+/*
+ * 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.job.k8s;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * The configurations of the {@code k8s} job executor, set with the {@code
+ * gravitino.jobExecutor.k8s.} prefix in the Gravitino server configuration.
+ */
+public class K8sJobExecutorConfigs {
+
+ /** The name of the job executor. */
+ public static final String K8S_JOB_EXECUTOR_NAME = "k8s";
+
+ /** The name of the Kubernetes cluster the jobs run in, which is part of the
job execution id. */
+ public static final String CLUSTER = "cluster";
+
+ /** The default cluster name. */
+ public static final String DEFAULT_CLUSTER = "default";
+
+ /** The kubeconfig file of the job cluster. */
+ public static final String KUBECONFIG = "kubeconfig";
+
+ /** The kubeconfig context to use, the current context if not set. */
+ public static final String CONTEXT = "context";
+
+ /**
+ * The API server URL of the job cluster, an alternative to the kubeconfig.
It must start with
+ * {@code https://}, or with {@code http://} if no token file is set.
+ */
+ public static final String MASTER_URL = "masterUrl";
+
+ /** The CA certificate file of the API server, used with {@link
#MASTER_URL}. */
+ public static final String CA_CERT_FILE = "caCertFile";
+
+ /**
+ * The bearer token file to access the API server, used with {@link
#MASTER_URL}. It is read again
+ * whenever it changes, so the token can be rotated by whoever provides the
file, for example by
+ * updating the Secret it is mounted from. Gravitino doesn't refresh the
token itself.
+ */
+ public static final String TOKEN_FILE = "tokenFile";
+
+ /** How long to wait to connect to the API server. */
+ public static final String CONNECTION_TIMEOUT_MS = "connectionTimeoutInMs";
+
+ /** The default connection timeout, 5 seconds. */
+ public static final long DEFAULT_CONNECTION_TIMEOUT_MS = 5_000L;
+
+ /** How long to wait for the response of a request to the API server. */
+ public static final String REQUEST_TIMEOUT_MS = "requestTimeoutInMs";
+
+ /** The default request timeout, 10 seconds. */
+ public static final long DEFAULT_REQUEST_TIMEOUT_MS = 10_000L;
+
+ /**
+ * How many times a failed request to the API server is tried again, for
example if the server
+ * can't be reached, doesn't answer in time or answers with a server error.
Each attempt waits for
+ * the timeouts again, so while the job cluster is down a submission, a
status query or a
+ * cancellation waits up to 1 + retries times the connection timeout if the
API server can't be
+ * reached, or the request timeout if it never answers.
+ */
+ public static final String REQUEST_RETRIES = "requestRetries";
+
+ /** The default number of request retries. */
+ public static final long DEFAULT_REQUEST_RETRIES = 2L;
+
+ /**
+ * The namespace the jobs run in, unless a job sets its own. It must be set:
the namespace needs
+ * the permissions and the Spark operator set up for the jobs, so none is
assumed.
+ */
+ public static final String NAMESPACE = "namespace";
+
+ /** The prefix of the names of the Kubernetes resources created for the
jobs. */
+ public static final String NAME_PREFIX = "namePrefix";
+
+ /** The default name prefix. */
+ public static final String DEFAULT_NAME_PREFIX = "gravitino-job";
+
+ /** How long the resources listed in a namespace are cached for the job
status queries. */
+ public static final String STATUS_CACHE_TTL_MS = "statusCacheTtlInMs";
+
+ /** The default status cache TTL, 5 seconds. */
+ public static final long DEFAULT_STATUS_CACHE_TTL_MS = 5_000L;
+
+ /** A job whose resource still has no status after this time is considered
failed. */
+ public static final String NO_STATUS_TIMEOUT_MS = "noStatusTimeoutInMs";
+
+ /** The default no status timeout, 10 minutes. */
+ public static final long DEFAULT_NO_STATUS_TIMEOUT_MS = 10 * 60 * 1000L;
+
+ /** The Spark image of the driver and executors, the default of a job's
container image. */
+ public static final String SPARK_IMAGE = "spark.image";
+
+ /** The Spark version of the image, set as the SparkApplication runtime
version. */
+ public static final String SPARK_VERSION = "spark.sparkVersion";
+
+ /** The service account of the Spark driver, the default of a job's driver
service account. */
+ public static final String SPARK_SERVICE_ACCOUNT = "spark.serviceAccount";
+
+ /** The default Spark driver service account. */
+ public static final String DEFAULT_SPARK_SERVICE_ACCOUNT = "spark";
+
+ /** How long the pods of a finished Spark job are retained, so that its
output can be read. */
+ public static final String SPARK_RESOURCE_RETAIN_DURATION_MS =
"spark.resourceRetainDurationInMs";
+
+ /** The default pod retain duration, 1 day. */
+ public static final long DEFAULT_SPARK_RESOURCE_RETAIN_DURATION_MS = 24 * 60
* 60 * 1000L;
+
+ /** The time to live of a SparkApplication after it stops. */
+ public static final String SPARK_TTL_AFTER_STOP_MS =
"spark.ttlAfterStopInMs";
+
+ /** The default SparkApplication time to live after it stops, 7 days. */
+ public static final long DEFAULT_SPARK_TTL_AFTER_STOP_MS = 7 * 24 * 60 * 60
* 1000L;
+
+ /** How long the operator waits for the driver pod to start. */
+ public static final String SPARK_DRIVER_START_TIMEOUT_MS =
"spark.driverStartTimeoutInMs";
+
+ /** How long the operator waits for the driver to be ready. */
+ public static final String SPARK_DRIVER_READY_TIMEOUT_MS =
"spark.driverReadyTimeoutInMs";
+
+ /**
+ * The default of the operator's start timeouts, 1 hour, so that a job waits
for resources instead
+ * of failing after the operator's default of 5 minutes.
+ */
+ public static final long DEFAULT_SPARK_START_TIMEOUT_MS = 60 * 60 * 1000L;
+
+ /**
+ * The prefix of the default Spark configurations of every Spark job. The
job's own Spark
+ * configurations take precedence.
+ */
+ public static final String SPARK_CONF_PREFIX = "spark.conf.";
+
+ // The resource name is the prefix, a dash and a job id of up to 19 digits.
Keep it within 44
+ // characters, so that the operator doesn't hash-truncate the Spark app id
derived from it.
+ private static final int MAX_NAME_PREFIX_LENGTH = 24;
+
+ // A DNS-1123 label, which the resource name must be.
+ private static final Pattern NAME_PREFIX_PATTERN =
Pattern.compile("[a-z]([-a-z0-9]*[a-z0-9])?");
+
+ private final String cluster;
+ @Nullable private final String kubeconfig;
+ @Nullable private final String context;
+ @Nullable private final String masterUrl;
+ @Nullable private final String caCertFile;
+ @Nullable private final String tokenFile;
+ private final String namespace;
+ private final int connectionTimeoutMs;
+ private final int requestTimeoutMs;
+ private final int requestRetries;
+ private final String namePrefix;
+ private final long statusCacheTtlMs;
+ private final long noStatusTimeoutMs;
+ private final String sparkImage;
+ private final String sparkVersion;
+ private final String sparkServiceAccount;
+ private final long sparkResourceRetainDurationMs;
+ private final long sparkTtlAfterStopMs;
+ private final long sparkDriverStartTimeoutMs;
+ private final long sparkDriverReadyTimeoutMs;
+ private final Map<String, String> sparkConf;
+
+ /**
+ * Parses and validates the job executor configurations.
+ *
+ * @param configs the configurations without the {@code
gravitino.jobExecutor.k8s.} prefix
+ * @throws IllegalArgumentException if a configuration is invalid
+ */
+ public K8sJobExecutorConfigs(Map<String, String> configs) {
+ this.cluster = stringValue(configs, CLUSTER, DEFAULT_CLUSTER);
+ Preconditions.checkArgument(
+ !cluster.contains("/"), "%s must not contain '/': %s", CLUSTER,
cluster);
+
+ this.kubeconfig = stringValue(configs, KUBECONFIG, null);
+ this.context = stringValue(configs, CONTEXT, null);
+ this.masterUrl = stringValue(configs, MASTER_URL, null);
+ this.caCertFile = stringValue(configs, CA_CERT_FILE, null);
+ this.tokenFile = stringValue(configs, TOKEN_FILE, null);
+ Preconditions.checkArgument(
+ masterUrl == null || (kubeconfig == null && context == null),
+ "%s can't be set together with %s or %s",
+ MASTER_URL,
+ KUBECONFIG,
+ CONTEXT);
+ Preconditions.checkArgument(
+ masterUrl != null || (caCertFile == null && tokenFile == null),
+ "%s and %s can only be set when %s is set",
+ CA_CERT_FILE,
+ TOKEN_FILE,
+ MASTER_URL);
+ // Without a scheme the client would pick one by probing the server, and
fall back to plain
+ // HTTP if the probe fails, for example because the server isn't reachable
at startup.
+ boolean https = StringUtils.startsWithIgnoreCase(masterUrl, "https://";);
+ Preconditions.checkArgument(
+ masterUrl == null || https ||
StringUtils.startsWithIgnoreCase(masterUrl, "http://";),
+ "%s must start with https:// or http://: %s",
+ MASTER_URL,
+ masterUrl);
+ Preconditions.checkArgument(
+ masterUrl == null || https || tokenFile == null,
+ "%s must start with https:// when %s is set, otherwise the token would
be sent in clear"
+ + " text: %s",
+ MASTER_URL,
+ TOKEN_FILE,
+ masterUrl);
+ Preconditions.checkArgument(
+ masterUrl == null || https || caCertFile == null,
+ "%s has no effect on an http:// %s, which isn't verified nor
encrypted: %s",
+ CA_CERT_FILE,
+ MASTER_URL,
+ masterUrl);
+
+ this.namespace = stringValue(configs, NAMESPACE, null);
+ Preconditions.checkArgument(namespace != null, "%s must be set",
NAMESPACE);
+ Preconditions.checkArgument(
+ K8sJobResourceUtils.isValidNamespace(namespace),
+ "%s isn't a valid namespace name: %s",
+ NAMESPACE,
+ namespace);
+ this.connectionTimeoutMs =
+ intValue(
+ CONNECTION_TIMEOUT_MS,
+ positiveLongValue(configs, CONNECTION_TIMEOUT_MS,
DEFAULT_CONNECTION_TIMEOUT_MS));
+ this.requestTimeoutMs =
+ intValue(
+ REQUEST_TIMEOUT_MS,
+ positiveLongValue(configs, REQUEST_TIMEOUT_MS,
DEFAULT_REQUEST_TIMEOUT_MS));
+ long retries = longValue(configs, REQUEST_RETRIES,
DEFAULT_REQUEST_RETRIES);
+ Preconditions.checkArgument(
+ retries >= 0, "%s must not be negative: %s", REQUEST_RETRIES, retries);
+ this.requestRetries = intValue(REQUEST_RETRIES, retries);
+
+ this.namePrefix = stringValue(configs, NAME_PREFIX, DEFAULT_NAME_PREFIX);
+ Preconditions.checkArgument(
+ namePrefix.length() <= MAX_NAME_PREFIX_LENGTH
+ && NAME_PREFIX_PATTERN.matcher(namePrefix).matches(),
+ "%s must be at most %s characters of lowercase letters, digits and
'-', starting with a"
+ + " letter and ending with a letter or digit: %s",
+ NAME_PREFIX,
+ MAX_NAME_PREFIX_LENGTH,
+ namePrefix);
+
+ this.statusCacheTtlMs = longValue(configs, STATUS_CACHE_TTL_MS,
DEFAULT_STATUS_CACHE_TTL_MS);
+ this.noStatusTimeoutMs =
+ positiveLongValue(configs, NO_STATUS_TIMEOUT_MS,
DEFAULT_NO_STATUS_TIMEOUT_MS);
+
+ this.sparkImage = stringValue(configs, SPARK_IMAGE, null);
+ Preconditions.checkArgument(sparkImage != null, "%s must be set",
SPARK_IMAGE);
+ this.sparkVersion = stringValue(configs, SPARK_VERSION, null);
+ Preconditions.checkArgument(sparkVersion != null, "%s must be set",
SPARK_VERSION);
+ this.sparkServiceAccount =
+ stringValue(configs, SPARK_SERVICE_ACCOUNT,
DEFAULT_SPARK_SERVICE_ACCOUNT);
+ this.sparkResourceRetainDurationMs =
+ positiveLongValue(
+ configs, SPARK_RESOURCE_RETAIN_DURATION_MS,
DEFAULT_SPARK_RESOURCE_RETAIN_DURATION_MS);
+ this.sparkTtlAfterStopMs =
+ positiveLongValue(configs, SPARK_TTL_AFTER_STOP_MS,
DEFAULT_SPARK_TTL_AFTER_STOP_MS);
+ this.sparkDriverStartTimeoutMs =
+ positiveLongValue(configs, SPARK_DRIVER_START_TIMEOUT_MS,
DEFAULT_SPARK_START_TIMEOUT_MS);
+ this.sparkDriverReadyTimeoutMs =
+ positiveLongValue(configs, SPARK_DRIVER_READY_TIMEOUT_MS,
DEFAULT_SPARK_START_TIMEOUT_MS);
+
+ Map<String, String> conf = new TreeMap<>();
+ configs.forEach(
+ (key, value) -> {
+ if (key.startsWith(SPARK_CONF_PREFIX) && key.length() >
SPARK_CONF_PREFIX.length()) {
+ conf.put(key.substring(SPARK_CONF_PREFIX.length()), value);
+ }
+ });
+ this.sparkConf = ImmutableMap.copyOf(conf);
+ }
+
+ /**
+ * Returns the name of the Kubernetes cluster the jobs run in.
+ *
+ * @return the cluster name
+ */
+ public String cluster() {
+ return cluster;
+ }
+
+ /**
+ * Returns the kubeconfig file of the job cluster.
+ *
+ * @return the kubeconfig file, or null if not set
+ */
+ @Nullable
+ public String kubeconfig() {
+ return kubeconfig;
+ }
+
+ /**
+ * Returns the kubeconfig context to use.
+ *
+ * @return the kubeconfig context, or null for the current context
+ */
+ @Nullable
+ public String context() {
+ return context;
+ }
+
+ /**
+ * Returns the API server URL of the job cluster.
+ *
+ * @return the API server URL, or null if not set
+ */
+ @Nullable
+ public String masterUrl() {
+ return masterUrl;
+ }
+
+ /**
+ * Returns the CA certificate file of the API server.
+ *
+ * @return the CA certificate file, or null if not set
+ */
+ @Nullable
+ public String caCertFile() {
+ return caCertFile;
+ }
+
+ /**
+ * Returns the bearer token file to access the API server.
+ *
+ * @return the token file, or null if not set
+ */
+ @Nullable
+ public String tokenFile() {
+ return tokenFile;
+ }
+
+ /**
+ * Returns the namespace the jobs run in, unless a job sets its own.
+ *
+ * @return the namespace of the jobs
+ */
+ public String namespace() {
+ return namespace;
+ }
+
+ /**
+ * Returns how long to wait to connect to the API server.
+ *
+ * @return the connection timeout in milliseconds
+ */
+ public int connectionTimeoutMs() {
+ return connectionTimeoutMs;
+ }
+
+ /**
+ * Returns how long to wait for the response of a request to the API server.
+ *
+ * @return the request timeout in milliseconds
+ */
+ public int requestTimeoutMs() {
+ return requestTimeoutMs;
+ }
+
+ /**
+ * Returns how many times a failed request to the API server is tried again.
+ *
+ * @return the number of request retries
+ */
+ public int requestRetries() {
+ return requestRetries;
+ }
+
+ /**
+ * Returns the prefix of the resource names.
+ *
+ * @return the name prefix
+ */
+ public String namePrefix() {
+ return namePrefix;
+ }
+
+ /**
+ * Returns how long the resources listed in a namespace are cached.
+ *
+ * @return the status cache TTL in milliseconds, caching is disabled if not
positive
+ */
+ public long statusCacheTtlMs() {
+ return statusCacheTtlMs;
+ }
+
+ /**
+ * Returns the time after which a job whose resource has no status is
considered failed.
+ *
+ * @return the no status timeout in milliseconds
+ */
+ public long noStatusTimeoutMs() {
+ return noStatusTimeoutMs;
+ }
+
+ /**
+ * Returns the default Spark image.
+ *
+ * @return the Spark image
+ */
+ public String sparkImage() {
+ return sparkImage;
+ }
+
+ /**
+ * Returns the Spark version of the image.
+ *
+ * @return the Spark version
+ */
+ public String sparkVersion() {
+ return sparkVersion;
+ }
+
+ /**
+ * Returns the default service account of the Spark driver.
+ *
+ * @return the driver service account
+ */
+ public String sparkServiceAccount() {
+ return sparkServiceAccount;
+ }
+
+ /**
+ * Returns how long the pods of a finished Spark job are retained.
+ *
+ * @return the pod retain duration in milliseconds
+ */
+ public long sparkResourceRetainDurationMs() {
+ return sparkResourceRetainDurationMs;
+ }
+
+ /**
+ * Returns the time to live of a SparkApplication after it stops.
+ *
+ * @return the time to live in milliseconds
+ */
+ public long sparkTtlAfterStopMs() {
+ return sparkTtlAfterStopMs;
+ }
+
+ /**
+ * Returns how long the operator waits for the driver pod to start.
+ *
+ * @return the timeout in milliseconds
+ */
+ public long sparkDriverStartTimeoutMs() {
+ return sparkDriverStartTimeoutMs;
+ }
+
+ /**
+ * Returns how long the operator waits for the driver to be ready.
+ *
+ * @return the timeout in milliseconds
+ */
+ public long sparkDriverReadyTimeoutMs() {
+ return sparkDriverReadyTimeoutMs;
+ }
+
+ /**
+ * Returns the default Spark configurations of every Spark job.
+ *
+ * @return the default Spark configurations
+ */
+ public Map<String, String> sparkConf() {
+ return sparkConf;
+ }
+
+ @Nullable
+ private static String stringValue(
+ Map<String, String> configs, String key, @Nullable String defaultValue) {
+ String value = configs.get(key);
+ return StringUtils.isNotBlank(value) ? value.trim() : defaultValue;
+ }
+
+ private static long longValue(Map<String, String> configs, String key, long
defaultValue) {
+ String value = stringValue(configs, key, null);
+ if (value == null) {
+ return defaultValue;
+ }
+ try {
+ return Long.parseLong(value);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(String.format("%s must be a number:
%s", key, value), e);
+ }
+ }
+
+ private static int intValue(String key, long value) {
+ Preconditions.checkArgument(value <= Integer.MAX_VALUE, "%s is too large:
%s", key, value);
+ return (int) value;
+ }
+
+ private static long positiveLongValue(
+ Map<String, String> configs, String key, long defaultValue) {
+ long value = longValue(configs, key, defaultValue);
+ Preconditions.checkArgument(value > 0, "%s must be positive: %s", key,
value);
+ return value;
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
new file mode 100644
index 0000000000..a0ffb490e8
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java
@@ -0,0 +1,106 @@
+/*
+ * 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.job.k8s;
+
+import com.google.common.collect.ImmutableMap;
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+/** The naming and labeling of the Kubernetes resources created for the jobs.
*/
+public final class K8sJobResourceUtils {
+
+ /** The label that marks the resources managed by Gravitino. */
+ public static final String LABEL_MANAGED_BY = "app.kubernetes.io/managed-by";
+
+ /** The {@link #LABEL_MANAGED_BY} value of the resources managed by
Gravitino. */
+ public static final String MANAGED_BY_GRAVITINO = "gravitino";
+
+ /** The label whose value is the Gravitino job id of the resource. */
+ public static final String LABEL_JOB_ID = "gravitino.apache.org/job-id";
+
+ /** The annotation whose value is the metalake of the job. */
+ public static final String ANNOTATION_METALAKE =
"gravitino.apache.org/metalake";
+
+ /** The annotation set when Gravitino requests to cancel the job. */
+ public static final String ANNOTATION_CANCEL_REQUESTED =
"gravitino.apache.org/cancel-requested";
+
+ // A DNS-1123 label, which a namespace name must be.
+ private static final Pattern NAMESPACE_PATTERN =
+ Pattern.compile("[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?");
+
+ private K8sJobResourceUtils() {}
+
+ /**
+ * Returns whether the given name is a valid name of a Kubernetes namespace.
+ *
+ * @param namespace the namespace name
+ * @return true if the name is a DNS-1123 label
+ */
+ public static boolean isValidNamespace(String namespace) {
+ return namespace != null && NAMESPACE_PATTERN.matcher(namespace).matches();
+ }
+
+ /**
+ * Returns the name of the Kubernetes resource of a job.
+ *
+ * @param configs the job executor configurations
+ * @param jobId the Gravitino job id
+ * @return the resource name
+ */
+ public static String resourceName(K8sJobExecutorConfigs configs, long jobId)
{
+ return configs.namePrefix() + "-" + jobId;
+ }
+
+ /**
+ * Returns the labels of the Kubernetes resource of a job.
+ *
+ * @param jobId the Gravitino job id
+ * @return the labels
+ */
+ public static Map<String, String> jobLabels(long jobId) {
+ return ImmutableMap.of(
+ LABEL_MANAGED_BY, MANAGED_BY_GRAVITINO, LABEL_JOB_ID,
String.valueOf(jobId));
+ }
+
+ /**
+ * Returns whether the Kubernetes resource was created for the given job.
+ *
+ * @param resource the resource
+ * @param jobId the Gravitino job id
+ * @return true if the resource carries the job id label of the job
+ */
+ public static boolean isOfJob(HasMetadata resource, long jobId) {
+ Map<String, String> labels =
+ resource.getMetadata() == null ? null :
resource.getMetadata().getLabels();
+ return labels != null &&
String.valueOf(jobId).equals(labels.get(LABEL_JOB_ID));
+ }
+
+ /**
+ * Returns whether Gravitino requested to cancel the job of the Kubernetes
resource.
+ *
+ * @param resource the resource
+ * @return true if the cancel annotation is set
+ */
+ public static boolean isCancelRequested(HasMetadata resource) {
+ Map<String, String> annotations =
+ resource.getMetadata() == null ? null :
resource.getMetadata().getAnnotations();
+ return annotations != null &&
"true".equals(annotations.get(ANNOTATION_CANCEL_REQUESTED));
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sKubeconfigUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sKubeconfigUtils.java
new file mode 100644
index 0000000000..1b7888c11e
--- /dev/null
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sKubeconfigUtils.java
@@ -0,0 +1,115 @@
+/*
+ * 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.job.k8s;
+
+import com.google.common.base.Preconditions;
+import io.fabric8.kubernetes.api.model.Cluster;
+import io.fabric8.kubernetes.api.model.Config;
+import io.fabric8.kubernetes.api.model.Context;
+import io.fabric8.kubernetes.api.model.NamedCluster;
+import io.fabric8.kubernetes.api.model.NamedContext;
+import io.fabric8.kubernetes.client.utils.Serialization;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * Reads a kubeconfig file into the model of the Kubernetes client, to check
what the client will
+ * make of it.
+ */
+final class K8sKubeconfigUtils {
+
+ private K8sKubeconfigUtils() {}
+
+ /**
+ * Returns the API server of the cluster that the kubeconfig selects, and
fails if it selects
+ * none. Without one the client keeps its default API server, the one of the
cluster Gravitino
+ * runs in, and would silently run the jobs there.
+ *
+ * @param kubeconfig the kubeconfig file, set by {@link
K8sJobExecutorConfigs#KUBECONFIG}
+ * @param contextName the context to use, or null for the current context of
the kubeconfig
+ * @return the server of the cluster of the context
+ * @throws IllegalArgumentException if the kubeconfig can't be parsed, or
its context doesn't
+ * select a cluster with a server
+ */
+ static String getSelectedServer(File kubeconfig, @Nullable String
contextName) {
+ Config content;
+ try (InputStream in = Files.newInputStream(kubeconfig.toPath())) {
+ content = Serialization.unmarshal(in, Config.class);
+ } catch (IOException | RuntimeException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "The kubeconfig %s set by %s can't be parsed",
+ kubeconfig, K8sJobExecutorConfigs.KUBECONFIG),
+ e);
+ }
+ // An empty file has no content at all.
+ Preconditions.checkArgument(
+ content != null,
+ "The kubeconfig %s set by %s is empty",
+ kubeconfig,
+ K8sJobExecutorConfigs.KUBECONFIG);
+
+ String name = contextName != null ? contextName :
content.getCurrentContext();
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(name),
+ "The kubeconfig %s set by %s has no current context, set its
current-context or %s",
+ kubeconfig,
+ K8sJobExecutorConfigs.KUBECONFIG,
+ K8sJobExecutorConfigs.CONTEXT);
+
+ // Like the client, the last entry with a name wins, and an entry without
content is skipped.
+ Context context = null;
+ if (content.getContexts() != null) {
+ for (NamedContext entry : content.getContexts()) {
+ if (entry != null && name.equals(entry.getName()) &&
entry.getContext() != null) {
+ context = entry.getContext();
+ }
+ }
+ }
+ Preconditions.checkArgument(
+ context != null,
+ "The context %s doesn't exist in the kubeconfig %s set by %s",
+ name,
+ kubeconfig,
+ K8sJobExecutorConfigs.KUBECONFIG);
+
+ Cluster cluster = null;
+ if (content.getClusters() != null && context.getCluster() != null) {
+ for (NamedCluster entry : content.getClusters()) {
+ if (entry != null
+ && context.getCluster().equals(entry.getName())
+ && entry.getCluster() != null) {
+ cluster = entry.getCluster();
+ }
+ }
+ }
+ Preconditions.checkArgument(
+ cluster != null && StringUtils.isNotBlank(cluster.getServer()),
+ "The cluster %s of the context %s has no server in the kubeconfig %s
set by %s",
+ context.getCluster(),
+ name,
+ kubeconfig,
+ K8sJobExecutorConfigs.KUBECONFIG);
+ return cluster.getServer();
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sClientUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sClientUtils.java
new file mode 100644
index 0000000000..cfa362f32c
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sClientUtils.java
@@ -0,0 +1,451 @@
+/*
+ * 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.job.k8s;
+
+import com.google.common.collect.ImmutableList;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class TestK8sClientUtils {
+
+ private static final String KUBECONFIG =
+ "apiVersion: v1\n"
+ + "kind: Config\n"
+ + "current-context: dev\n"
+ + "clusters:\n"
+ + "- name: dev\n"
+ + " cluster: {server: 'https://dev:6443'}\n"
+ + "- name: prod\n"
+ + " cluster: {server: 'https://prod:6443'}\n"
+ + "users:\n"
+ + "- name: admin\n"
+ + " user: {token: admin-token}\n"
+ + "contexts:\n"
+ + "- name: dev\n"
+ + " context: {cluster: dev, user: admin}\n"
+ + "- name: prod\n"
+ + " context: {cluster: prod, user: admin}\n";
+
+ @TempDir Path dir;
+
+ @Test
+ public void testMasterUrl() throws IOException {
+ Path caCert = Files.write(dir.resolve("ca.crt"),
"cert".getBytes(StandardCharsets.UTF_8));
+ Path token = Files.write(dir.resolve("token"),
"token-1\n".getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.MASTER_URL, "https://jobs:6443";);
+ map.put(K8sJobExecutorConfigs.CA_CERT_FILE, caCert.toString());
+ map.put(K8sJobExecutorConfigs.TOKEN_FILE, token.toString());
+
+ Config config = K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map));
+
+ Assertions.assertEquals("https://jobs:6443/";, config.getMasterUrl());
+ Assertions.assertEquals(caCert.toString(), config.getCaCertFile());
+ Assertions.assertEquals("token-1",
config.getOauthTokenProvider().getToken());
+ }
+
+ @Test
+ public void testMissingFiles() throws IOException {
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.MASTER_URL, "https://jobs:6443";);
+ map.put(K8sJobExecutorConfigs.TOKEN_FILE,
dir.resolve("missing").toString());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)));
+
+ // A token file without a token fails the startup too, rather than the
first request.
+ map.put(
+ K8sJobExecutorConfigs.TOKEN_FILE,
+ Files.write(dir.resolve("empty-token"), new byte[0]).toString());
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)));
+ Assertions.assertTrue(e.getMessage().contains("No token can be read"),
e.getMessage());
+
+ Map<String, String> kubeconfig = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ kubeconfig.put(K8sJobExecutorConfigs.KUBECONFIG,
dir.resolve("missing").toString());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(kubeconfig)));
+ }
+
+ @Test
+ public void testKubeconfig() throws IOException {
+ Path kubeconfig =
+ Files.write(dir.resolve("config"),
KUBECONFIG.getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.KUBECONFIG, kubeconfig.toString());
+
+ Config current = K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map));
+ Assertions.assertEquals("https://dev:6443/";, current.getMasterUrl());
+
+ map.put(K8sJobExecutorConfigs.CONTEXT, "prod");
+ Config prod = K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map));
+ Assertions.assertEquals("https://prod:6443/";, prod.getMasterUrl());
+ Assertions.assertEquals("admin-token", prod.getAutoOAuthToken());
+
+ // A context that doesn't exist must not fall back to the current one.
+ map.put(K8sJobExecutorConfigs.CONTEXT, "staging");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)));
+ }
+
+ @Test
+ public void testKubeconfigWithoutUsableCluster() throws IOException {
+ // Without a context that selects a cluster with a server, the client
would silently fall
+ // back to the API server of the cluster Gravitino itself runs in.
+ String devContext = "context: {cluster: dev, user: admin}";
+ String devCluster = "- name: dev\n cluster: {server:
'https://dev:6443'}\n";
+ String[] unusable = {
+ // No current context.
+ KUBECONFIG.replace("current-context: dev\n", ""),
+ // The current context doesn't exist.
+ KUBECONFIG.replace("current-context: dev\n", "current-context:
staging\n"),
+ // The context has no content, or no cluster.
+ KUBECONFIG.replace(" " + devContext + "\n", ""),
+ KUBECONFIG.replace(devContext, "context: {user: admin}"),
+ // The cluster of the context isn't defined.
+ KUBECONFIG.replace(devContext, "context: {cluster: nope}"),
+ // The cluster is defined without content, with null or empty content,
or a blank server.
+ KUBECONFIG.replace(devCluster, "- name: dev\n"),
+ KUBECONFIG.replace(devCluster, "- name: dev\n cluster: null\n"),
+ KUBECONFIG.replace(devCluster, "- name: dev\n cluster: {}\n"),
+ KUBECONFIG.replace(devCluster, "- name: dev\n cluster: {server: '
'}\n"),
+ // Not a kubeconfig at all.
+ "- just\n- a list\n",
+ };
+ for (String content : unusable) {
+ Assertions.assertNotEquals(KUBECONFIG, content);
+ Path kubeconfig =
+ Files.write(dir.resolve("config"),
content.getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.KUBECONFIG, kubeconfig.toString());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)),
+ content);
+ }
+
+ // Naming a usable context works, whatever the current context is.
+ Path kubeconfig =
+ Files.write(
+ dir.resolve("config"),
+ KUBECONFIG.replace(devCluster, "- name:
dev\n").getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.KUBECONFIG, kubeconfig.toString());
+ map.put(K8sJobExecutorConfigs.CONTEXT, "prod");
+ Assertions.assertEquals(
+ "https://prod:6443/";,
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getMasterUrl());
+ }
+
+ @Test
+ public void testKubeconfigSelectingDefaultServer() throws IOException {
+ // A kubeconfig may well select the default API server of the client.
+ Path kubeconfig =
+ Files.write(
+ dir.resolve("config"),
+ KUBECONFIG
+ .replace("https://dev:6443";, "https://kubernetes.default.svc";)
+ .getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.KUBECONFIG, kubeconfig.toString());
+ Assertions.assertEquals(
+ "https://kubernetes.default.svc/";,
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getMasterUrl());
+ }
+
+ @Test
+ public void testKubeconfigServerWithoutScheme() throws IOException {
+ // The client picks the scheme of such a server by probing it, whatever it
picks is accepted.
+ // Nothing listens on the port, so the probe fails right away.
+ Path kubeconfig =
+ Files.write(
+ dir.resolve("config"),
+ KUBECONFIG.replace("https://dev:6443";,
"localhost:1").getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.KUBECONFIG, kubeconfig.toString());
+ String masterUrl =
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getMasterUrl();
+ Assertions.assertTrue(masterUrl.endsWith("://localhost:1/"), masterUrl);
+
+ Files.write(
+ kubeconfig,
+ KUBECONFIG
+ .replace("https://dev:6443";, "HTTPS://dev:6443/")
+ .getBytes(StandardCharsets.UTF_8));
+ // The scheme is compared ignoring its case, and the client keeps the
server as written.
+ Assertions.assertEquals(
+ "HTTPS://dev:6443/",
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getMasterUrl());
+ }
+
+ @Test
+ public void testKubeconfigWithDuplicateClusters() throws IOException {
+ // Like the client, the last cluster with a name wins.
+ Path kubeconfig =
+ Files.write(
+ dir.resolve("config"),
+ KUBECONFIG
+ .replace(
+ "- name: prod\n",
+ "- name: dev\n cluster: {server: 'https://dev2:6443'}\n-
name: prod\n")
+ .getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.KUBECONFIG, kubeconfig.toString());
+ Assertions.assertEquals(
+ "https://dev2:6443/";,
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getMasterUrl());
+ }
+
+ @Test
+ public void testEmptyKubeconfig() throws IOException {
+ for (String content : new String[] {"", "# nothing\n", "---\n"}) {
+ Path kubeconfig =
+ Files.write(dir.resolve("config"),
content.getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.KUBECONFIG, kubeconfig.toString());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)),
+ content);
+ }
+ }
+
+ @Test
+ public void testDefaultLookupWithoutUsableCluster() throws IOException {
+ // The client prefers this system property over the KUBECONFIG environment
variable.
+ String property = "kubeconfig";
+ String before = System.getProperty(property);
+ Path kubeconfig = dir.resolve("config");
+ System.setProperty(property, kubeconfig.toString());
+ try {
+ Files.write(kubeconfig, KUBECONFIG.getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ Assertions.assertEquals(
+ "https://dev:6443/";,
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getMasterUrl());
+
+ // The cluster of the context isn't defined, so the client would stay on
the API server of
+ // the cluster Gravitino itself runs in.
+ Files.write(
+ kubeconfig,
+ KUBECONFIG
+ .replace("context: {cluster: dev, user: admin}", "context:
{cluster: nope}")
+ .getBytes(StandardCharsets.UTF_8));
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)));
+ Assertions.assertTrue(e.getMessage().contains("selects no cluster"),
e.getMessage());
+
+ map.put(K8sJobExecutorConfigs.CONTEXT, "dev");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)));
+
+ map.put(K8sJobExecutorConfigs.CONTEXT, "prod");
+ Assertions.assertEquals(
+ "https://prod:6443/";,
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getMasterUrl());
+
+ // A context that doesn't exist is reported as such, even if the current
context that the
+ // client falls back to is broken too.
+ map.put(K8sJobExecutorConfigs.CONTEXT, "staging");
+ e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)));
+ Assertions.assertTrue(e.getMessage().contains("The context staging"),
e.getMessage());
+
+ // No context is selected: no current context, or one that doesn't exist.
+ map.remove(K8sJobExecutorConfigs.CONTEXT);
+ for (String content :
+ new String[] {
+ KUBECONFIG.replace("current-context: dev\n", ""),
+ KUBECONFIG.replace("current-context: dev\n", "current-context:
staging\n")
+ }) {
+ Files.write(kubeconfig, content.getBytes(StandardCharsets.UTF_8));
+ e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)),
+ content);
+ Assertions.assertTrue(e.getMessage().contains("has no current
context"), e.getMessage());
+ }
+
+ // Without any kubeconfig, and outside a pod, no cluster is configured
at all.
+ System.setProperty(property, dir.resolve("missing").toString());
+ e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)));
+ Assertions.assertTrue(e.getMessage().contains("No Kubernetes cluster"),
e.getMessage());
+
+ // In a pod, the kubelet sets the API server through these, which the
client reads from
+ // the system properties first.
+ System.setProperty("KUBERNETES_SERVICE_HOST", "10.0.0.1");
+ System.setProperty("KUBERNETES_SERVICE_PORT", "443");
+ try {
+ Assertions.assertEquals(
+ "https://10.0.0.1:443/";,
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getMasterUrl());
+ } finally {
+ System.clearProperty("KUBERNETES_SERVICE_HOST");
+ System.clearProperty("KUBERNETES_SERVICE_PORT");
+ }
+ } finally {
+ if (before == null) {
+ System.clearProperty(property);
+ } else {
+ System.setProperty(property, before);
+ }
+ }
+ }
+
+ @Test
+ public void testRequestLimits() throws IOException {
+ // The limits are set in every connection mode.
+ Path kubeconfig =
+ Files.write(dir.resolve("config"),
KUBECONFIG.getBytes(StandardCharsets.UTF_8));
+ Map<String, String> withKubeconfig = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ withKubeconfig.put(K8sJobExecutorConfigs.KUBECONFIG,
kubeconfig.toString());
+ Map<String, String> withMasterUrl = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ withMasterUrl.put(K8sJobExecutorConfigs.MASTER_URL, "https://jobs:6443";);
+ for (Map<String, String> map : ImmutableList.of(withKubeconfig,
withMasterUrl)) {
+ Config defaults = K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map));
+ Assertions.assertEquals(5_000, defaults.getConnectionTimeout());
+ Assertions.assertEquals(10_000, defaults.getRequestTimeout());
+ Assertions.assertEquals(2, defaults.getRequestRetryBackoffLimit());
+
+ map.put(K8sJobExecutorConfigs.CONNECTION_TIMEOUT_MS, "1000");
+ map.put(K8sJobExecutorConfigs.REQUEST_TIMEOUT_MS, "2000");
+ map.put(K8sJobExecutorConfigs.REQUEST_RETRIES, "0");
+ Config custom = K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map));
+ Assertions.assertEquals(1_000, custom.getConnectionTimeout());
+ Assertions.assertEquals(2_000, custom.getRequestTimeout());
+ Assertions.assertEquals(0, custom.getRequestRetryBackoffLimit());
+ }
+ }
+
+ @Test
+ public void testRequestFailsFastWithoutApiServer() {
+ // Nothing listens on the port. With the client's own defaults this takes
about 20 seconds.
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.MASTER_URL, "https://127.0.0.1:1";);
+ try (KubernetesClient client = K8sClientUtils.createClient(new
K8sJobExecutorConfigs(map))) {
+ Assertions.assertTimeout(
+ Duration.ofSeconds(10),
+ () ->
+ Assertions.assertThrows(
+ KubernetesClientException.class,
+ () -> client.pods().inNamespace("default").list()));
+ }
+ }
+
+ @Test
+ public void testNamespace() throws IOException {
+ // The configured namespace wins over the one of the kubeconfig context.
+ Path kubeconfig =
+ Files.write(
+ dir.resolve("config"),
+ KUBECONFIG
+ .replace(
+ "{cluster: dev, user: admin}", "{cluster: dev, user:
admin, namespace: ctx}")
+ .getBytes(StandardCharsets.UTF_8));
+ Map<String, String> map = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ map.put(K8sJobExecutorConfigs.KUBECONFIG, kubeconfig.toString());
+ map.put(K8sJobExecutorConfigs.NAMESPACE, "jobs");
+ Assertions.assertEquals(
+ "jobs", K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(map)).getNamespace());
+
+ Map<String, String> masterUrl = new
HashMap<>(TestK8sJobExecutorConfigs.requiredConfigs());
+ masterUrl.put(K8sJobExecutorConfigs.MASTER_URL, "https://jobs:6443";);
+ masterUrl.put(K8sJobExecutorConfigs.NAMESPACE, "jobs");
+ Assertions.assertEquals(
+ "jobs",
+ K8sClientUtils.createClientConfig(new
K8sJobExecutorConfigs(masterUrl)).getNamespace());
+ }
+
+ @Test
+ public void testTokenRotation() throws IOException {
+ Path token = Files.write(dir.resolve("token"),
"token-1".getBytes(StandardCharsets.UTF_8));
+ K8sClientUtils.FileTokenProvider provider = new
K8sClientUtils.FileTokenProvider(token);
+ Assertions.assertEquals("token-1", provider.getToken());
+
+ Files.write(token, "token-2".getBytes(StandardCharsets.UTF_8));
+ // Make sure the modification time changes, even on file systems with a
coarse resolution.
+ Files.setLastModifiedTime(
+ token, FileTime.fromMillis(Files.getLastModifiedTime(token).toMillis()
+ 1000));
+ Assertions.assertEquals("token-2", provider.getToken());
+
+ // The token read before is kept while the file is empty or missing, as it
is being rewritten.
+ Files.write(token, new byte[0]);
+ Files.setLastModifiedTime(
+ token, FileTime.fromMillis(Files.getLastModifiedTime(token).toMillis()
+ 2000));
+ Assertions.assertEquals("token-2", provider.getToken());
+ Files.delete(token);
+ Assertions.assertEquals("token-2", provider.getToken());
+
+ Files.write(token, "token-3".getBytes(StandardCharsets.UTF_8));
+ Assertions.assertEquals("token-3", provider.getToken());
+ }
+
+ @Test
+ public void testTokenRotationWithinOneModificationTime() throws IOException {
+ // The file is emptied and rewritten within one tick of a coarse file
system clock.
+ Path token = Files.write(dir.resolve("token"),
"token-1".getBytes(StandardCharsets.UTF_8));
+ K8sClientUtils.FileTokenProvider provider = new
K8sClientUtils.FileTokenProvider(token);
+ Assertions.assertEquals("token-1", provider.getToken());
+
+ FileTime rotatedAt =
FileTime.fromMillis(Files.getLastModifiedTime(token).toMillis() + 1000);
+ Files.write(token, new byte[0]);
+ Files.setLastModifiedTime(token, rotatedAt);
+ Assertions.assertEquals("token-1", provider.getToken());
+
+ Files.write(token, "token-2".getBytes(StandardCharsets.UTF_8));
+ Files.setLastModifiedTime(token, rotatedAt);
+ Assertions.assertEquals("token-2", provider.getToken());
+ }
+
+ @Test
+ public void testUnreadableTokenFile() throws IOException {
+ Path empty = Files.write(dir.resolve("empty"), new byte[0]);
+ Assertions.assertThrows(
+ IllegalStateException.class, new
K8sClientUtils.FileTokenProvider(empty)::getToken);
+ Assertions.assertThrows(
+ UncheckedIOException.class,
+ new
K8sClientUtils.FileTokenProvider(dir.resolve("missing"))::getToken);
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutionId.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutionId.java
new file mode 100644
index 0000000000..3103a051cc
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutionId.java
@@ -0,0 +1,76 @@
+/*
+ * 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.job.k8s;
+
+import org.apache.gravitino.exceptions.NoSuchJobException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestK8sJobExecutionId {
+
+ @Test
+ public void testFormatAndParse() {
+ K8sJobExecutionId id =
+ K8sJobExecutionId.of(
+ "default", K8sJobExecutionId.Kind.SPARK_APPLICATION, "jobs",
"gravitino-job-1");
+ Assertions.assertEquals("default/sparkapp/jobs/gravitino-job-1",
id.toString());
+
+ K8sJobExecutionId parsed = K8sJobExecutionId.parse(id.toString());
+ Assertions.assertEquals(id, parsed);
+ Assertions.assertEquals(id.hashCode(), parsed.hashCode());
+ Assertions.assertEquals("default", parsed.cluster());
+ Assertions.assertEquals(K8sJobExecutionId.Kind.SPARK_APPLICATION,
parsed.kind());
+ Assertions.assertEquals("jobs", parsed.namespace());
+ Assertions.assertEquals("gravitino-job-1", parsed.name());
+ }
+
+ @Test
+ public void testKind() {
+ Assertions.assertEquals("sparkapp",
K8sJobExecutionId.Kind.SPARK_APPLICATION.id());
+ Assertions.assertEquals(
+ K8sJobExecutionId.Kind.SPARK_APPLICATION,
K8sJobExecutionId.Kind.fromId("sparkapp"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
K8sJobExecutionId.Kind.fromId("job"));
+ }
+
+ @Test
+ public void testInvalid() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sJobExecutionId.of("a/b",
K8sJobExecutionId.Kind.SPARK_APPLICATION, "ns", "n"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> K8sJobExecutionId.of("default",
K8sJobExecutionId.Kind.SPARK_APPLICATION, "", "n"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> K8sJobExecutionId.of("default",
null, "ns", "n"));
+
+ for (String invalid :
+ new String[] {
+ null,
+ "",
+ "ns/name",
+ "sparkapp/ns/name",
+ "default/sparkapp/ns/",
+ "a/b/c/d/e",
+ "default/job/ns/name"
+ }) {
+ Assertions.assertThrows(NoSuchJobException.class, () ->
K8sJobExecutionId.parse(invalid));
+ }
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorConfigs.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorConfigs.java
new file mode 100644
index 0000000000..bf18938f7c
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorConfigs.java
@@ -0,0 +1,197 @@
+/*
+ * 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.job.k8s;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestK8sJobExecutorConfigs {
+
+ public static Map<String, String> requiredConfigs() {
+ return ImmutableMap.of(
+ K8sJobExecutorConfigs.NAMESPACE, "default",
+ K8sJobExecutorConfigs.SPARK_IMAGE, "gravitino-spark:3.5.9",
+ K8sJobExecutorConfigs.SPARK_VERSION, "3.5.9");
+ }
+
+ @Test
+ public void testDefaults() {
+ K8sJobExecutorConfigs configs = new
K8sJobExecutorConfigs(requiredConfigs());
+
+ Assertions.assertEquals("default", configs.cluster());
+ Assertions.assertNull(configs.kubeconfig());
+ Assertions.assertNull(configs.context());
+ Assertions.assertNull(configs.masterUrl());
+ Assertions.assertEquals("default", configs.namespace());
+ Assertions.assertEquals("gravitino-job", configs.namePrefix());
+ Assertions.assertEquals(5_000, configs.connectionTimeoutMs());
+ Assertions.assertEquals(10_000, configs.requestTimeoutMs());
+ Assertions.assertEquals(2, configs.requestRetries());
+ Assertions.assertEquals(5_000L, configs.statusCacheTtlMs());
+ Assertions.assertEquals(600_000L, configs.noStatusTimeoutMs());
+ Assertions.assertEquals("gravitino-spark:3.5.9", configs.sparkImage());
+ Assertions.assertEquals("3.5.9", configs.sparkVersion());
+ Assertions.assertEquals("spark", configs.sparkServiceAccount());
+ Assertions.assertEquals(86_400_000L,
configs.sparkResourceRetainDurationMs());
+ Assertions.assertEquals(604_800_000L, configs.sparkTtlAfterStopMs());
+ Assertions.assertEquals(3_600_000L, configs.sparkDriverStartTimeoutMs());
+ Assertions.assertEquals(3_600_000L, configs.sparkDriverReadyTimeoutMs());
+ Assertions.assertTrue(configs.sparkConf().isEmpty());
+ }
+
+ @Test
+ public void testCustomValues() {
+ Map<String, String> map = new HashMap<>(requiredConfigs());
+ map.put(K8sJobExecutorConfigs.CLUSTER, "prod");
+ map.put(K8sJobExecutorConfigs.NAMESPACE, " jobs ");
+ map.put(K8sJobExecutorConfigs.NAME_PREFIX, "gvt-a");
+ map.put(K8sJobExecutorConfigs.STATUS_CACHE_TTL_MS, "0");
+ map.put(K8sJobExecutorConfigs.SPARK_TTL_AFTER_STOP_MS, "1000");
+ map.put("spark.conf.spark.executor.instances", "2");
+ map.put("spark.conf.", "ignored");
+
+ K8sJobExecutorConfigs configs = new K8sJobExecutorConfigs(map);
+
+ Assertions.assertEquals("prod", configs.cluster());
+ Assertions.assertEquals("jobs", configs.namespace());
+ Assertions.assertEquals("gvt-a", configs.namePrefix());
+ Assertions.assertEquals(0L, configs.statusCacheTtlMs());
+ Assertions.assertEquals(1000L, configs.sparkTtlAfterStopMs());
+ Assertions.assertEquals(ImmutableMap.of("spark.executor.instances", "2"),
configs.sparkConf());
+ }
+
+ @Test
+ public void testDurationKeys() {
+ // The duration keys end with InMs, like the other job configurations of
Gravitino.
+ Map<String, String> map = new HashMap<>(requiredConfigs());
+ map.put("connectionTimeoutInMs", "7");
+ map.put("requestTimeoutInMs", "8");
+ map.put("requestRetries", "0");
+ map.put("statusCacheTtlInMs", "1");
+ map.put("noStatusTimeoutInMs", "2");
+ map.put("spark.resourceRetainDurationInMs", "3");
+ map.put("spark.ttlAfterStopInMs", "4");
+ map.put("spark.driverStartTimeoutInMs", "5");
+ map.put("spark.driverReadyTimeoutInMs", "6");
+
+ K8sJobExecutorConfigs configs = new K8sJobExecutorConfigs(map);
+
+ Assertions.assertEquals(7, configs.connectionTimeoutMs());
+ Assertions.assertEquals(8, configs.requestTimeoutMs());
+ Assertions.assertEquals(0, configs.requestRetries());
+ Assertions.assertEquals(1L, configs.statusCacheTtlMs());
+ Assertions.assertEquals(2L, configs.noStatusTimeoutMs());
+ Assertions.assertEquals(3L, configs.sparkResourceRetainDurationMs());
+ Assertions.assertEquals(4L, configs.sparkTtlAfterStopMs());
+ Assertions.assertEquals(5L, configs.sparkDriverStartTimeoutMs());
+ Assertions.assertEquals(6L, configs.sparkDriverReadyTimeoutMs());
+ }
+
+ @Test
+ public void testRequiredConfigs() {
+ for (String required :
+ new String[] {
+ K8sJobExecutorConfigs.NAMESPACE,
+ K8sJobExecutorConfigs.SPARK_IMAGE,
+ K8sJobExecutorConfigs.SPARK_VERSION
+ }) {
+ Map<String, String> map = new HashMap<>(requiredConfigs());
+ map.remove(required);
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> new
K8sJobExecutorConfigs(map), required);
+ Assertions.assertEquals(required + " must be set", e.getMessage());
+
+ map.put(required, " ");
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> new
K8sJobExecutorConfigs(map), required);
+ }
+ }
+
+ @Test
+ public void testConnectionConfigs() {
+ assertInvalid(
+ K8sJobExecutorConfigs.MASTER_URL,
+ "https://k8s:6443";,
+ K8sJobExecutorConfigs.KUBECONFIG,
+ "/kube/config");
+ assertInvalid(
+ K8sJobExecutorConfigs.MASTER_URL,
+ "https://k8s:6443";,
+ K8sJobExecutorConfigs.CONTEXT,
+ "prod");
+ assertInvalid(K8sJobExecutorConfigs.TOKEN_FILE, "/token");
+ assertInvalid(K8sJobExecutorConfigs.CA_CERT_FILE, "/ca.crt");
+
+ // A master URL without a scheme would let the client fall back to plain
HTTP, and a token
+ // must never be sent over it.
+ assertInvalid(K8sJobExecutorConfigs.MASTER_URL, "k8s:6443");
+ assertInvalid(K8sJobExecutorConfigs.MASTER_URL, "ftp://k8s:6443";);
+ assertInvalid(
+ K8sJobExecutorConfigs.MASTER_URL,
+ "http://k8s:6443";,
+ K8sJobExecutorConfigs.TOKEN_FILE,
+ "/token");
+ // A CA certificate would never be used with plain HTTP.
+ assertInvalid(
+ K8sJobExecutorConfigs.MASTER_URL,
+ "http://k8s:6443";,
+ K8sJobExecutorConfigs.CA_CERT_FILE,
+ "/ca.crt");
+ Map<String, String> http = new HashMap<>(requiredConfigs());
+ http.put(K8sJobExecutorConfigs.MASTER_URL, "http://localhost:8001";);
+ Assertions.assertEquals("http://localhost:8001";, new
K8sJobExecutorConfigs(http).masterUrl());
+
+ Map<String, String> map = new HashMap<>(requiredConfigs());
+ map.put(K8sJobExecutorConfigs.MASTER_URL, "https://k8s:6443";);
+ map.put(K8sJobExecutorConfigs.CA_CERT_FILE, "/ca.crt");
+ map.put(K8sJobExecutorConfigs.TOKEN_FILE, "/token");
+ K8sJobExecutorConfigs configs = new K8sJobExecutorConfigs(map);
+ Assertions.assertEquals("https://k8s:6443";, configs.masterUrl());
+ Assertions.assertEquals("/ca.crt", configs.caCertFile());
+ Assertions.assertEquals("/token", configs.tokenFile());
+ }
+
+ @Test
+ public void testInvalidValues() {
+ assertInvalid(K8sJobExecutorConfigs.CLUSTER, "a/b");
+ assertInvalid(K8sJobExecutorConfigs.NAMESPACE, "Gravitino_Jobs");
+ assertInvalid(K8sJobExecutorConfigs.NAME_PREFIX, "Gravitino");
+ assertInvalid(K8sJobExecutorConfigs.NAME_PREFIX, "gravitino-");
+ assertInvalid(K8sJobExecutorConfigs.NAME_PREFIX,
"a-name-prefix-that-is-too-long");
+ assertInvalid(K8sJobExecutorConfigs.NO_STATUS_TIMEOUT_MS, "0");
+ assertInvalid(K8sJobExecutorConfigs.CONNECTION_TIMEOUT_MS, "0");
+ assertInvalid(K8sJobExecutorConfigs.REQUEST_TIMEOUT_MS, "-1");
+ assertInvalid(K8sJobExecutorConfigs.REQUEST_TIMEOUT_MS, "3000000000");
+ assertInvalid(K8sJobExecutorConfigs.REQUEST_RETRIES, "-1");
+ assertInvalid(K8sJobExecutorConfigs.REQUEST_RETRIES, "many");
+ assertInvalid(K8sJobExecutorConfigs.SPARK_DRIVER_START_TIMEOUT_MS, "1h");
+ }
+
+ private static void assertInvalid(String... keyValues) {
+ Map<String, String> map = new HashMap<>(requiredConfigs());
+ for (int i = 0; i < keyValues.length; i += 2) {
+ map.put(keyValues[i], keyValues[i + 1]);
+ }
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new
K8sJobExecutorConfigs(map));
+ }
+}
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
new file mode 100644
index 0000000000..76c7661738
--- /dev/null
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/TestK8sJobResourceUtils.java
@@ -0,0 +1,85 @@
+/*
+ * 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.job.k8s;
+
+import com.google.common.collect.ImmutableMap;
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestK8sJobResourceUtils {
+
+ @Test
+ public void testResourceName() {
+ K8sJobExecutorConfigs configs =
+ new K8sJobExecutorConfigs(TestK8sJobExecutorConfigs.requiredConfigs());
+ Assertions.assertEquals("gravitino-job-42",
K8sJobResourceUtils.resourceName(configs, 42L));
+ }
+
+ @Test
+ public void testJobLabels() {
+ Assertions.assertEquals(
+ ImmutableMap.of(
+ "app.kubernetes.io/managed-by", "gravitino",
"gravitino.apache.org/job-id", "42"),
+ K8sJobResourceUtils.jobLabels(42L));
+
+ ConfigMap resource =
+ new ConfigMapBuilder()
+ .withNewMetadata()
+ .withLabels(K8sJobResourceUtils.jobLabels(42L))
+ .endMetadata()
+ .build();
+ Assertions.assertTrue(K8sJobResourceUtils.isOfJob(resource, 42L));
+ Assertions.assertFalse(K8sJobResourceUtils.isOfJob(resource, 43L));
+ Assertions.assertFalse(
+ K8sJobResourceUtils.isOfJob(
+ new ConfigMapBuilder().withNewMetadata().endMetadata().build(),
42L));
+ Assertions.assertFalse(K8sJobResourceUtils.isOfJob(new ConfigMap(), 42L));
+ }
+
+ @Test
+ public void testIsCancelRequested() {
+ Assertions.assertFalse(
+ K8sJobResourceUtils.isCancelRequested(
+ new ConfigMapBuilder().withNewMetadata().endMetadata().build()));
+ Assertions.assertFalse(K8sJobResourceUtils.isCancelRequested(new
ConfigMap()));
+
Assertions.assertFalse(K8sJobResourceUtils.isCancelRequested(annotated("false")));
+
Assertions.assertTrue(K8sJobResourceUtils.isCancelRequested(annotated("true")));
+ }
+
+ @Test
+ public void testIsValidNamespace() {
+ for (String valid : new String[] {"default", "team-a", "a", "ns1", "1ns"})
{
+ Assertions.assertTrue(K8sJobResourceUtils.isValidNamespace(valid),
valid);
+ }
+ for (String invalid :
+ new String[] {null, "", "Team", "team_a", "-team", "team-", "a/b",
"a".repeat(64)}) {
+ Assertions.assertFalse(K8sJobResourceUtils.isValidNamespace(invalid),
invalid);
+ }
+ }
+
+ private static ConfigMap annotated(String cancelRequested) {
+ return new ConfigMapBuilder()
+ .withNewMetadata()
+ .addToAnnotations(K8sJobResourceUtils.ANNOTATION_CANCEL_REQUESTED,
cancelRequested)
+ .endMetadata()
+ .build();
+ }
+}
diff --git a/settings.gradle.kts b/settings.gradle.kts
index ced9feab65..edc3da371e 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -127,6 +127,7 @@ include(":catalogs:hadoop-auth")
include(":lineage")
include(":mcp-server")
include(":plugins:idp-basic")
+include(":plugins:k8s-job-executor")
include(":plugins:tms")
include(
":maintenance:optimizer-api",
