jerryshao commented on code in PR #13724: URL: https://github.com/apache/gravitino/pull/13724#discussion_r4236615864
########## 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") +} Review Comment: [Question] Is distribution packaging deliberately left to part 3? The sibling plugin wires its jar into the server distribution with a `copyLibAndConfigs` task (plugins/idp-basic/build.gradle.kts:72-86) that the root build lists in the distribution assembly (build.gradle.kts:1173). This module has neither, so `gradlew assembleDistribution` produces a package without the shaded jar and the `k8s` executor could not be configured even after part 3 registers it. I assume that lands with part 3 and is intentionally out of scope here; flagging it only so it does not fall between the parts. Verified by: grepping the root `build.gradle.kts` distribution task list (only `:plugins:idp-basic:copyLibAndConfigs` under `plugins/`) and comparing with `plugins/idp-basic/build.gradle.kts` in this checkout. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java: ########## @@ -0,0 +1,422 @@ +/* + * 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. */ + 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. + */ + public static final String TOKEN_FILE = "tokenFile"; + + /** The default namespace of the jobs. */ + public static final String NAMESPACE = "namespace"; + + /** The default namespace. */ + public static final String DEFAULT_NAMESPACE = "default"; + + /** 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 = "statusCacheTtlMs"; Review Comment: [Nit] These duration keys read differently from the only existing executor's keys. The `local` executor spells its duration keys `jobStatusKeepTimeInMs` and `cancelForceKillDelayInMs` (core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutorConfigs.java:36 and 43), whereas this module uses the `...Ms` form (`statusCacheTtlMs`, `noStatusTimeoutMs`, `spark.resourceRetainDurationMs`, `spark.ttlAfterStopMs`, `spark.driverStartTimeoutMs`, `spark.driverReadyTimeoutMs`). These become user-facing configuration once part 3 registers the executor, and renaming them afterwards is a breaking change, so it is cheaper to settle the convention now. I slightly prefer the shorter `...Ms` used here; if that is the intended direction, it is worth saying so on the issue so the `local` keys can follow. Verified by: grepping all job executor config classes in this checkout; `LocalJobExecutorConfigs` is the only other one and uses `InMs` for both of its duration keys. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,152 @@ +/* + * 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.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 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. + */ + static final class FileTokenProvider implements OAuthTokenProvider { + + private final Path tokenFile; + + private FileTime lastModified; + + private String token; + + 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.isNotBlank(content)) { + token = content; + lastModified = modified; + } else if (token == null) { + throw new IllegalStateException("The token file " + tokenFile + " is empty"); + } else { + // The file is being rewritten, keep the token until the new one is there. Remember + // this version of the file, so that it is neither read nor reported again. + lastModified = modified; + LOG.warn("The token file {} is empty, keeping the token read before", tokenFile); + } + } + } catch (IOException e) { + if (token == null) { + throw new UncheckedIOException("Failed to read the token file " + tokenFile, e); + } + // For example, the file is being replaced, keep the token until the new one is there. + LOG.warn("Failed to read the token file {}, keeping the token read before", tokenFile, e); + } + return token; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(K8sClientUtils.class); + + 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> + * + * @param configs the job executor configurations + * @return the Kubernetes client configuration + */ + static Config createClientConfig(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); + builder.withOauthTokenProvider(new FileTokenProvider(Paths.get(configs.tokenFile()))); + } + return builder.build(); + } + + Config config; + if (configs.kubeconfig() != null) { + checkFileExists(configs.kubeconfig(), K8sJobExecutorConfigs.KUBECONFIG); + config = Config.fromKubeconfig(configs.context(), new File(configs.kubeconfig())); + } else { + config = Config.autoConfigure(configs.context()); + } + + // The client falls back to the current context if the given one doesn't exist, which would + // silently run the jobs in another cluster. + String context = + config.getCurrentContext() == null ? null : config.getCurrentContext().getName(); + Preconditions.checkArgument( + configs.context() == null || configs.context().equals(context), + "The context %s set by %s doesn't exist in the kubeconfig", Review Comment: [Important] A kubeconfig with no usable context silently falls back to the in-cluster API server, which is the failure the comment right above is guarding against. The guard only fires when `context` is set. When `kubeconfig` is set and `context` is not, and the file has no `current-context` (or its current context's cluster/user entries are incomplete, which fabric8 drops), `Config.fromKubeconfig` reaches `KubeConfigUtils.merge`, which returns early before `setMasterUrl` (KubeConfigUtils.java:179-181 in v7.2.0). The `Config` then keeps the default values set by `Config.empty()` -> `new Config(false)` -> `shouldSetDefaultValues=true`, i.e. `masterUrl = DEFAULT_MASTER_URL = "https://kubernetes.default.svc"` (Config.java:130 and 408). `configs.context()` is null, so the `Preconditions` here passes and the executor quietly submits jobs to the API server of the cluster Gravitino itself runs in, with no mention of the kubeconfig it was told to use. Suggested fix: when `configs.kubeconfig() != null` (or whenever we took this branch at all), also require a context to have been resolved, e.g. Preconditions.checkArgument( config.getCurrentContext() != null, "No usable context found in the kubeconfig %s set by %s", configs.kubeconfig(), K8sJobExecutorConfigs.KUBECONFIG); before the equality check, and add a kubeconfig fixture without `current-context` to TestK8sClientUtils. Verified by: reading fabric8 7.2.0 sources fetched in this run (KubeConfigUtils.merge early return at line 179; Config.empty/Config(boolean) at 258-325; DEFAULT_MASTER_URL at 130; default assignment at 408), and by checking that the only kubeconfig fixture in TestK8sClientUtils.java:36 always sets `current-context: dev`, so this path is untested. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,152 @@ +/* + * 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.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 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. + */ + static final class FileTokenProvider implements OAuthTokenProvider { + + private final Path tokenFile; + + private FileTime lastModified; + + private String token; + + 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.isNotBlank(content)) { + token = content; + lastModified = modified; + } else if (token == null) { + throw new IllegalStateException("The token file " + tokenFile + " is empty"); + } else { + // The file is being rewritten, keep the token until the new one is there. Remember + // this version of the file, so that it is neither read nor reported again. + lastModified = modified; + LOG.warn("The token file {} is empty, keeping the token read before", tokenFile); + } + } + } catch (IOException e) { + if (token == null) { + throw new UncheckedIOException("Failed to read the token file " + tokenFile, e); + } + // For example, the file is being replaced, keep the token until the new one is there. + LOG.warn("Failed to read the token file {}, keeping the token read before", tokenFile, e); + } + return token; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(K8sClientUtils.class); + + 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> + * + * @param configs the job executor configurations + * @return the Kubernetes client configuration + */ + static Config createClientConfig(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); + builder.withOauthTokenProvider(new FileTokenProvider(Paths.get(configs.tokenFile()))); Review Comment: [Question] Should the configured `namespace` be put on the client `Config`? Today the client's default namespace depends on how we connect. `gravitino.jobExecutor.k8s.namespace` is validated and stored but never reaches the fabric8 `Config`. The result differs per branch: in this `masterUrl` branch `Config.empty()` leaves `namespace` null with `defaultNamespace = true` (Config.java:156, 408), while the kubeconfig branch gets the context's namespace via `KubeConfigUtils.merge` (`clientConfig.setNamespace(currentContext.getContext().getNamespace())`, KubeConfigUtils.java:183) and `Config.autoConfigure` additionally picks up the in-cluster service-account namespace file and the `KUBERNETES_NAMESPACE` property (Config.java:593, 909-918). So any fabric8 call in part 2 or 3 that omits `.inNamespace(...)` will target the kubeconfig context's namespace, or the pod's own namespace, rather than the configured one, and the configured value would be silently ignored. If every call site is going to pass the namespace explicitly, this is fine as is and worth a comment saying so; otherwise `builder.withNamespace(configs.namespace())` here, and the same on the config returned from the kubeconfig branch, would make the configuration authoritative in both modes. Verified by: reading fabric8 7.2.0 `Config` and `KubeConfigUtils` sources fetched in this run, and grepping this module for any other use of `configs.namespace()` (only `K8sJobExecutorConfigs` validation and the getter; nothing passes it to the client). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
