jerryshao commented on code in PR #13724: URL: https://github.com/apache/gravitino/pull/13724#discussion_r4247582077
########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,255 @@ +/* + * 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 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. + */ + 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: with the defaults, about 15 seconds if the API server can't be reached. + * + * @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, or the kubeconfig has no context that selects a cluster with a server + */ + 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()); Review Comment: Done in #13700, commit 29caba205, as promised above. - A namespace LIST that fails because the API server can't be reached or doesn't answer (no response at all, with an `IOException` behind it) now fails the status query, instead of being cached as an empty listing and followed by a GET of the SparkApplication. - After such a failure, `getJobExecutionInfo` fails right away for 10 seconds, for every namespace, and then tries the API server again. So only the first query of a round waits for the timeout, and a round over many jobs takes about one timeout instead of two per job. - A response from the API server, even 403 or 5xx, is not treated this way: a forbidden LIST still falls back to GETs, and the next query goes to the API server again. - Only the status queries are affected. Submitting and cancelling a job always try the API server. Tests in `TestK8sJobExecutor`: `testFailFastWhileApiServerUnreachable` (against a port nothing listens on, with and without the namespace listing), `testApiServerErrorDoesNotFailFast` and `testIsUnreachable`. _🤖 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]
