jerryshao commented on code in PR #13724: URL: https://github.com/apache/gravitino/pull/13724#discussion_r4245163599
########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,272 @@ +/* + * 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 io.fabric8.kubernetes.client.utils.Serialization; +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.List; +import java.util.Map; +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 { + + private final Path tokenFile; + + private FileTime lastModified; + + private String token; + + private boolean readFailureLogged; + + 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; Review Comment: [Important] Remembering the modification time of an empty token file can pin a stale token until the file changes again. The empty-read branch now advances `lastModified`, so the next call skips the read at line 68. If the rotating writer truncates the file and writes the new token inside the same modification-time tick - a file system with a coarse timestamp resolution (NFS, CIFS, ext3), or simply a fast rewrite - the provider has already recorded that tick while the file was empty, so it keeps serving the token read before until the file changes again, which may be the next rotation hours later. That token is expired by then, every request gets a 401, and nothing re-reads the file. Before this commit the provider kept re-reading and picked the new token up; the repeated WARN was the problem Copilot raised, not the token handling. The read-failure branch three lines below solves the same problem without that window: the `readFailureLogged` flag suppresses the repeated log while the file is still read every time. The same shape here, e.g. an `emptyLogged` flag cleared on a successful read and no write to `lastModified`, keeps the log quiet with no stale-token window. Re-reading a few bytes per request is cheaper than pinning an expired token. `testTokenRotation` cannot catch this: it bumps the modification time by 1-2 s between writes precisely to be robust on "file systems with a coarse resolution" (TestK8sClientUtils.java:210, 216), which is the case where this bites. Verified by: running a standalone javac replica of `getToken()` in this session over the sequence token-1 at mtime M0, empty at M1, token-2 still at M1. With this line the provider returns token-1 with token-2 on disk; without it, token-2. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,272 @@ +/* + * 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 io.fabric8.kubernetes.client.utils.Serialization; +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.List; +import java.util.Map; +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 { + + private final Path tokenFile; + + private FileTime lastModified; + + private String token; + + private boolean readFailureLogged; + + 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. It + // is checked again on every request, so only the first failure in a row is reported. + if (!readFailureLogged) { + LOG.warn("Failed to read the token file {}, keeping the token read before", tokenFile, e); + readFailureLogged = true; + } Review Comment: [Nit] A token file that stays unreadable is now reported once and then serves a stale token silently. `readFailureLogged` is cleared only by a successful read at line 94, so if the file is deleted or its permissions change for good, there is a single WARN and then nothing, while every request keeps using a token that will expire. Logging again at an interval, or raising the severity after a number of consecutive failures, would keep the condition visible without the spam the flag avoids. Verified by: reading this branch and line 94 in this run. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,272 @@ +/* + * 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 io.fabric8.kubernetes.client.utils.Serialization; +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.List; +import java.util.Map; +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 { + + private final Path tokenFile; + + private FileTime lastModified; + + private String token; + + private boolean readFailureLogged; + + 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. It + // is checked again on every request, so only the first failure in a row is reported. + if (!readFailureLogged) { + LOG.warn("Failed to read the token file {}, keeping the token read before", tokenFile, e); + readFailureLogged = true; + } + return token; + } + readFailureLogged = false; + 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> + * + * <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); + String server = getKubeconfigServer(configs); + config = Config.fromKubeconfig(configs.context(), new File(configs.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( + normalizeServer(server).equals(normalizeServer(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 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", + configs.context(), + K8sJobExecutorConfigs.CONTEXT); + config.setNamespace(configs.namespace()); Review Comment: [Question] Should an unset `namespace` keep the namespace of the pod rather than becoming `default`? `namespace` defaults to `default` (K8sJobExecutorConfigs.java:183), so this line and line 142 set `default` on the client whenever the operator configured none. In the fabric8-default mode that overrides what `Config.autoConfigure` read from the service account's namespace file (Config.java:282, `tryNamespaceFromPath`), so a Gravitino pod running in namespace `gravitino` would create its job resources in `default`. That precedence is right when the value was set explicitly, which is what I asked for last round; it is the unset case I am unsure about, where the pod's own namespace looks like the friendlier default. Overriding only when the key is present would give both. Fine as is if the intent is that the jobs' namespace is always explicit - the javadoc at lines 123-124 does say so. Verified by: reading `Config.autoConfigure` and `tryNamespaceFromPath` in the fabric8 7.2.0 sources, and `testNamespace`, which covers the kubeconfig and master URL modes but not this one. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,272 @@ +/* + * 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 io.fabric8.kubernetes.client.utils.Serialization; +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.List; +import java.util.Map; +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 { + + private final Path tokenFile; + + private FileTime lastModified; + + private String token; + + private boolean readFailureLogged; + + 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. It + // is checked again on every request, so only the first failure in a row is reported. + if (!readFailureLogged) { + LOG.warn("Failed to read the token file {}, keeping the token read before", tokenFile, e); + readFailureLogged = true; + } + return token; + } + readFailureLogged = false; + 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> + * + * <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); + String server = getKubeconfigServer(configs); + config = Config.fromKubeconfig(configs.context(), new File(configs.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( + normalizeServer(server).equals(normalizeServer(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 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", + configs.context(), + K8sJobExecutorConfigs.CONTEXT); + config.setNamespace(configs.namespace()); + return config; + } + + /** + * Returns the API server of the cluster that the kubeconfig selects, and fails if it selects + * none. Otherwise the client keeps its default API server, the one of the cluster Gravitino runs + * in, and would silently run the jobs there. The default API server itself isn't rejected, as a + * kubeconfig may well select it. + */ + private static String getKubeconfigServer(K8sJobExecutorConfigs configs) { + Map<String, Object> kubeconfig; + try { + kubeconfig = + getMap( + Serialization.unmarshal( + new String( + Files.readAllBytes(Paths.get(configs.kubeconfig())), StandardCharsets.UTF_8), + Map.class)); Review Comment: [Nit] This walk over the raw YAML could use fabric8's own parser, the one the client uses on the next line. `Config.fromKubeconfig(String, File)` parses the file with `KubeConfigUtils.parseConfig(File)` (Config.java:790-791), a public method that returns the typed `io.fabric8.kubernetes.api.model.Config`. Reusing it would replace `getNamedEntry`, `getMap` and the `Map` navigation - about 45 lines - with typed lookups over `getContexts()` and `getClusters()`, parse the file exactly as the client does, and keep the two from disagreeing on an unusual kubeconfig: with duplicate entries of one name, fabric8's merge keeps the last (KubeConfigUtils.java:240-272) while `getNamedEntry` returns the first, which would fail the server comparison rather than agree with the client. Verified by: reading KubeConfigUtils.java:68-88 and 240-272 in the fabric8 7.2.0 sources fetched in this run. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,272 @@ +/* + * 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 io.fabric8.kubernetes.client.utils.Serialization; +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.List; +import java.util.Map; +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 { + + private final Path tokenFile; + + private FileTime lastModified; + + private String token; + + private boolean readFailureLogged; + + 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. It + // is checked again on every request, so only the first failure in a row is reported. + if (!readFailureLogged) { + LOG.warn("Failed to read the token file {}, keeping the token read before", tokenFile, e); + readFailureLogged = true; + } + return token; + } + readFailureLogged = false; + 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> + * + * <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); + String server = getKubeconfigServer(configs); + config = Config.fromKubeconfig(configs.context(), new File(configs.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( + normalizeServer(server).equals(normalizeServer(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 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", + configs.context(), + K8sJobExecutorConfigs.CONTEXT); + config.setNamespace(configs.namespace()); + return config; + } + + /** + * Returns the API server of the cluster that the kubeconfig selects, and fails if it selects + * none. Otherwise the client keeps its default API server, the one of the cluster Gravitino runs + * in, and would silently run the jobs there. The default API server itself isn't rejected, as a + * kubeconfig may well select it. + */ + private static String getKubeconfigServer(K8sJobExecutorConfigs configs) { + Map<String, Object> kubeconfig; + try { + kubeconfig = + getMap( + Serialization.unmarshal( + new String( + Files.readAllBytes(Paths.get(configs.kubeconfig())), StandardCharsets.UTF_8), + Map.class)); + } catch (IOException e) { + throw new UncheckedIOException("Failed to read the kubeconfig " + configs.kubeconfig(), e); + } catch (RuntimeException e) { + throw new IllegalArgumentException( + String.format( + "The kubeconfig %s set by %s can't be parsed", + configs.kubeconfig(), K8sJobExecutorConfigs.KUBECONFIG), + e); + } + + Object contextName = + configs.context() != null + ? configs.context() + : kubeconfig == null ? null : kubeconfig.get("current-context"); + Preconditions.checkArgument( + contextName instanceof String && StringUtils.isNotBlank((String) contextName), + "The kubeconfig %s set by %s has no current context, set its current-context or %s", + configs.kubeconfig(), + K8sJobExecutorConfigs.KUBECONFIG, + K8sJobExecutorConfigs.CONTEXT); + + Map<String, Object> context = getNamedEntry(kubeconfig, "contexts", contextName, "context"); + Preconditions.checkArgument( + context != null, + "The context %s doesn't exist in the kubeconfig %s set by %s", + contextName, + configs.kubeconfig(), + K8sJobExecutorConfigs.KUBECONFIG); + + Object clusterName = context.get("cluster"); + Map<String, Object> cluster = getNamedEntry(kubeconfig, "clusters", clusterName, "cluster"); + Object server = cluster == null ? null : cluster.get("server"); + Preconditions.checkArgument( + server instanceof String && StringUtils.isNotBlank((String) server), + "The cluster %s of the context %s has no server in the kubeconfig %s set by %s", + clusterName, + contextName, + configs.kubeconfig(), + K8sJobExecutorConfigs.KUBECONFIG); + return (String) server; + } + + /** Returns the URL of an API server in the form the client keeps it, to compare two of them. */ + private static String normalizeServer(String server) { + String url = server.trim(); + if (!url.startsWith("http://";) && !url.startsWith("https://";)) { + url = "https://"; + url; + } + return url.endsWith("/") ? url : url + "/"; Review Comment: [Important] This assumes `https`, while the client picks the scheme from a live TLS probe, so a kubeconfig whose `server` has no scheme is rejected at startup by the check at line 152. For `server: 10.0.0.1:6443` - legal in a kubeconfig, and what `kubectl config set-cluster --server=host:port` writes - `KubeConfigUtils.merge` calls `clientConfig.setMasterUrl(cluster.getServer())` (KubeConfigUtils.java:191) before it sets the cluster's CA (lines 195-197). `setMasterUrl` runs `ensureHttps`, which prefixes `https://` only if `SSLUtils.isHttpsAvailable(config)` returns true (Config.java:303-308, 1141-1144), and that method opens a real HTTPS connection with a 1 s timeout using an SSL context built from a config that has no CA yet (SSLUtils.java:59-71). With a private CA, or an API server not reachable at startup, it returns false and the client ends up with `http://10.0.0.1:6443/`. `normalizeServer` returns `https://10.0.0.1:6443/`, the precondition fails, and the server refuses to start with "selects the server 10.0.0.1:6443, but the client would use http://10.0.0.1:6443/";, which reads like a contradiction. Two smaller mismatches in the same helper: the prefix test is case-sensitive while fabric8 lowercases it, so `HTTPS://host` gets a second prefix, and this trims the value while fabric8 does not. Comparing the two with the scheme stripped, or deriving the expected value through fabric8 itself (`new ConfigBuilder(Config.empty()).withMasterUrl(server).build().getMasterUrl()`), keeps the check without re-implementing the normalization. A fixture with a scheme-less server would cover it. Verified by: reading the fabric8 7.2.0 sources jar fetched in this run (Config.java:303-308 and 1141-1144, SSLUtils.java:59-71, KubeConfigUtils.java:191-197); no fixture in TestK8sClientUtils uses a scheme-less server. -- 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]
