jerryshao commented on code in PR #13724: URL: https://github.com/apache/gravitino/pull/13724#discussion_r4245369186
########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sKubeconfigUtils.java: ########## @@ -0,0 +1,107 @@ +/* + * 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.internal.KubeConfigUtils; +import java.io.File; +import javax.annotation.Nullable; +import org.apache.commons.lang3.StringUtils; + +/** + * Reads a kubeconfig file with the parser 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 { + content = KubeConfigUtils.parseConfig(kubeconfig); + } catch (RuntimeException e) { + throw new IllegalArgumentException( + String.format( + "The kubeconfig %s set by %s can't be parsed", + kubeconfig, K8sJobExecutorConfigs.KUBECONFIG), + e); + } + + String name = + contextName != null ? contextName : content == null ? null : content.getCurrentContext(); Review Comment: [Nit] The `content == null` arm cannot fire, and if it could, line 74 would throw an NPE for a configured context. `KubeConfigUtils.parseConfig` never returns null: for a document that unmarshals to null it throws an NPE from `ret.getContexts()` inside `parseConfig`, which `launderThrowable` rethrows as is, and the `catch (RuntimeException)` above turns it into the intended `IllegalArgumentException`. I ran `parseConfig` over an empty file, a comment-only file, `---` and `null` in this session and got that for all four. Dropping the null arm would keep the method honest about what can happen, since the guard only protects the branch where `contextName` is null anyway. Verified by: the probe described above, and `KubeConfigUtils.parseConfig` in the fabric8 7.2.0 sources. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,212 @@ +/* + * 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. + * + * @param configs the job executor configurations + * @return the Kubernetes client configuration + * @throws IllegalArgumentException if a configured file doesn't exist, or the kubeconfig has no + * usable context + */ + 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: [Nit] An empty token file is only noticed on the first job submission. `checkFileExists` on the line above validates that the file is there, but its content is read lazily, so an empty `tokenFile` throws the `IllegalStateException` of line 76 from inside the HTTP layer on the first request, rather than at startup like every other misconfiguration in this method. Calling `getToken()` once while the configuration is built would turn it into a startup failure next to the others, and would cost one read. Verified by: reading lines 68-95 and 153-165 in this run; neither `createClient` nor `createClientConfig` calls the provider, and fabric8 only calls it per request. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java: ########## @@ -0,0 +1,423 @@ +/* + * 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 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 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); Review Comment: [Important] A `masterUrl` without a scheme silently becomes plain HTTP, which sends the bearer token in clear text. The value is accepted as written and handed to `withMasterUrl` (K8sClientUtils.java:155). fabric8's `ensureHttps` adds a scheme only when there is none, and picks it by probing the server with a 1 s TLS connect (Config.java:303-308, `SSLUtils.isHttpsAvailable`). If the API server is unreachable at startup, or presents a certificate the probe's context does not trust, the client ends up on `http://`: the token from `tokenFile` then travels unencrypted, the configured `caCertFile` is never used, and nothing says so in the log. Ran it: `new ConfigBuilder(Config.empty()).withMasterUrl("10.255.255.1:6443").withCaCertFile("/etc/does/not/matter.crt").build().getMasterUrl()` returns `http://10.255.255.1:6443/`, while `https://10.255.255.1:6443` is kept as given. `kubectl config set-cluster --server=host:port` writes exactly such a value, so it is a plausible thing to paste into the server configuration. Suggestion: require `masterUrl` to start with `https://`, or at least with a scheme, with a test. Accepting whatever scheme the client picks is right for a kubeconfig server, where the value was not typed into Gravitino's configuration; this one is ours to validate. Verified by: a probe compiled and run against the fabric8 7.2.0 jars in this session, plus Config.java:303-308 and `SSLUtils.isHttpsAvailable` in the 7.2.0 sources. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,212 @@ +/* + * 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. + * + * @param configs the job executor configurations + * @return the Kubernetes client configuration + * @throws IllegalArgumentException if a configured file doesn't exist, or the kubeconfig has no + * usable context + */ + 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.withNamespace(configs.namespace()).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, + // whatever it does with a kubeconfig that is malformed or changed 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()); Review Comment: [Important] This branch keeps the wrong-cluster hole that 8d8882e closed for an explicit kubeconfig. With `kubeconfig` unset, the context is handed to `Config.autoConfigure`, which reads the `kubeconfig` system property or `KUBECONFIG`, else `~/.kube/config` - the lookup the javadoc at lines 141-142 describes. If the selected context exists but its cluster is missing or incomplete, `KubeConfigUtils.merge` calls `setCurrentContext` and then returns before `setMasterUrl`, because `mergedClusters.get(...)` is null. The client keeps `Config.DEFAULT_MASTER_URL`, and the check at lines 190-194 passes, since the current context is exactly the configured one. The user's token is still taken from that kubeconfig, so Gravitino would submit the jobs to the API server of the cluster it runs in, using that token - the failure `K8sKubeconfigUtils` was added to prevent, reachable through the other branch. Ran it: a kubeconfig with `current-context: dev`, a `dev` context naming cluster `dev`, and only a cluster named `other` defined. `Config.autoConfigure("dev")` returns masterUrl `https://kubernetes.default.svc/` with currentContext `dev`, so `contextCheckPasses` is true; with no context configured, the same. Suggestion: when a context was resolved from a kubeconfig in this branch, apply the same check, e.g. run `K8sKubeconfigUtils.getSelectedServer` over the first existing file of `Config.getKubeconfigFilenames()`, or at least reject a resolved context whose server is still the default. A test can set the `kubeconfig` system property, which fabric8 prefers over the environment variable. Verified by: a probe compiled and run against the fabric8 7.2.0 jars in this session (output above), plus `KubeConfigUtils.merge`, `Config.autoConfigure` and `findKubeConfigFiles` in the 7.2.0 sources. -- 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]
