yuqi1129 commented on code in PR #13724: URL: https://github.com/apache/gravitino/pull/13724#discussion_r4246068582
########## 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: [Question] Bounding each call works. Two things about how part 3 (#13700) uses these limits: 1. **Serial status poll.** fabric8 7.2.0 retries on any `IOException`, including `HttpTimeoutException`, and on 429/5xx (`StandardHttpClient.shouldRetry`). With the defaults, a server that accepts TCP but never finishes the handshake took 15.8 s (3 connections) in my probe, and 5.7 s with `requestRetries=0`. A server that completes TLS and then never answers should take up to about 3 x `requestTimeoutInMs` = 30 s, which the Javadoc at line 149 doesn't mention. You said JobManager polls the jobs one after another on one thread, so while the job cluster is down one round takes N x 16-30 s, which is about 27-50 minutes for 100 jobs. Does #13700 short-circuit the rest of a round after the first connection failure? 2. **Non-idempotent retries.** The retry applies to POST too. If creating the SparkApplication succeeds on the server but the response times out, the retry gets 409 AlreadyExists. Does part 3 treat that as submitted, by checking the labels or annotations, rather than reporting a failure and leaving an orphan application? fabric8's default 10 retries already did this, so it isn't new here, but `requestRetries` now makes it configurable. Not blocking for part 1. Verified by: the probe above (retries 2 and 0), and `javap -c` of `StandardHttpClient` in kubernetes-client-api 7.2.0. The 30 s hang case is derived from the code, not measured. ########## 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()); + 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()); + // The default lookup may merge several kubeconfig files, so they aren't checked one by one. + // A context was found in them if there is a current context, and a context that selects a + // cluster with a server moves the client off its default API server. If it is still there, + // the context selects none, and the client would silently run the jobs in the cluster + // Gravitino runs in. + Preconditions.checkArgument( + config.getCurrentContext() == null Review Comment: [Nit] A kubeconfig found by the default lookup that selects no context still passes startup. When fabric8 finds a kubeconfig but no context is selected, `KubeConfigUtils.merge` returns before `setCurrentContext`. That happens when the file has no `current-context` and `context` isn't set, or when `current-context` names a context that doesn't exist. `getCurrentContext()` is then null, so the first operand lets this check pass, and the client stays on `https://kubernetes.default.svc/`. The same file set through `kubeconfig` is rejected with "has no current context", so the two branches now disagree. Ran it on a2db2c7f6 with the shaded jar, pointing the `kubeconfig` system property at the file. Both shapes give `master=https://kubernetes.default.svc/ ctx=null`, with no token. It fails closed: once a kubeconfig is found, fabric8 skips `tryServiceAccount`, so there is no token and no service-account CA. But startup succeeds and the first submission fails with a TLS or 401 error that doesn't point at the kubeconfig. Suggestion: tell "no kubeconfig found" (in-cluster, fine) apart from "kubeconfig found but no context selected" (reject). For example, reject when `getCurrentContext()` is null but `config.getContexts()` isn't empty, or when any of `Config.getKubeconfigFilenames()` exists. A case without `current-context` in `testDefaultLookupWithoutUsableCluster` would pin it. Verified by: the probe above, plus `KubeConfigUtils.merge` and `Config.autoConfigure` in the fabric8 7.2.0 sources. I did not run it inside a pod. ########## 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()); + 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()); + // The default lookup may merge several kubeconfig files, so they aren't checked one by one. + // A context was found in them if there is a current context, and a context that selects a + // cluster with a server moves the client off its default API server. If it is still there, + // the context selects none, and the client would silently run the jobs in the cluster + // Gravitino runs in. + Preconditions.checkArgument( + config.getCurrentContext() == null + || !getHostAndPath(Config.DEFAULT_MASTER_URL) + .equals(getHostAndPath(config.getMasterUrl())), + "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", + config.getCurrentContext() == null ? null : config.getCurrentContext().getName(), + K8sJobExecutorConfigs.KUBECONFIG, + K8sJobExecutorConfigs.MASTER_URL, + Config.DEFAULT_MASTER_URL, + K8sJobExecutorConfigs.MASTER_URL); + } + + // 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( Review Comment: [Nit] When the configured `context` doesn't exist, the error names a different context. With `context=prod`, a kubeconfig without `prod`, and a current context `dev` whose cluster isn't defined, fabric8 falls back to `dev`. The default-lookup check above then fails first, with "The context dev of the kubeconfig found by the default lookup selects no cluster with a server". The real problem is the one this check reports, "The context prod set by context doesn't exist in the kubeconfig", but it never runs. Moving this check before the default-lookup check would report the right cause. Verified by: a probe on a2db2c7f6 with the shaded jar and that kubeconfig. -- 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]
