jerryshao commented on code in PR #13724: URL: https://github.com/apache/gravitino/pull/13724#discussion_r4236663709
########## 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: Fixed in efdf90323. Reproduced first: a kubeconfig without `current-context`, and one whose context names no defined cluster, both left the client on `https://kubernetes.default.svc`. When `kubeconfig` is set, the executor now requires a current context whose cluster is defined in that file and fails at startup otherwise. `testKubeconfigWithoutUsableContext` covers no current context, a context without a cluster, and a context naming an undefined cluster. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## 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: Done in efdf90323: the configured `namespace` is now set on the client `Config` in all three connection modes, so it wins over the namespace of the kubeconfig context or of the pod. Part 3 passes the namespace explicitly on every call, so this is a safety net. `testNamespace` asserts it for the kubeconfig and master URL modes. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## 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: Yes, deliberately out of scope here. Copying the shaded jar into the distribution and registering `k8s` in `JobExecutorFactory` are both part of #13616, which follows part 3. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ -- 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]
