Copilot commented on code in PR #13724: URL: https://github.com/apache/gravitino/pull/13724#discussion_r4228954525
########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sClientUtils.java: ########## @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.job.k8s; + +import com.google.common.base.Preconditions; +import io.fabric8.kubernetes.client.Config; +import io.fabric8.kubernetes.client.ConfigBuilder; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientBuilder; +import io.fabric8.kubernetes.client.OAuthTokenProvider; +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.attribute.FileTime; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Creates the Kubernetes client of the job cluster from the job executor configurations. */ +public final class K8sClientUtils { + + /** + * An {@link OAuthTokenProvider} that reads the token from a file, and reads it again whenever the + * file changes, so that a rotated token is picked up without restarting the server. + */ + static final class FileTokenProvider implements OAuthTokenProvider { + + private final Path tokenFile; + + private FileTime lastModified; + + private String token; + + FileTokenProvider(Path tokenFile) { + this.tokenFile = tokenFile; + } + + @Override + public synchronized String getToken() { + try { + FileTime modified = Files.getLastModifiedTime(tokenFile); + if (token == null || !modified.equals(lastModified)) { + String content = new String(Files.readAllBytes(tokenFile), StandardCharsets.UTF_8).trim(); + if (StringUtils.isNotBlank(content)) { + token = content; + lastModified = modified; + } else if (token == null) { + throw new IllegalStateException("The token file " + tokenFile + " is empty"); + } else { + // The file is being rewritten, keep the token until the new one is there. Review Comment: When the token file becomes empty (common during rotation), the provider keeps the old token but does not update `lastModified`. This causes every subsequent `getToken()` call to re-read the same empty file and emit repeated WARN logs until the file changes again. To avoid log spam and extra I/O, update `lastModified = modified` even when keeping the previous token due to blank content (and consider reducing repeated WARNs in this “expected during rewrite” scenario). ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java: ########## @@ -0,0 +1,104 @@ +/* + * 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.collect.ImmutableMap; +import io.fabric8.kubernetes.api.model.HasMetadata; +import java.util.Map; +import java.util.regex.Pattern; + +/** The naming and labeling of the Kubernetes resources created for the jobs. */ +public final class K8sJobResourceUtils { + + /** The label that marks the resources managed by Gravitino. */ + public static final String LABEL_MANAGED_BY = "app.kubernetes.io/managed-by"; + + /** The {@link #LABEL_MANAGED_BY} value of the resources managed by Gravitino. */ + public static final String MANAGED_BY_GRAVITINO = "gravitino"; + + /** The label whose value is the Gravitino job id of the resource. */ + public static final String LABEL_JOB_ID = "gravitino.apache.org/job-id"; + + /** The annotation whose value is the metalake of the job. */ + public static final String ANNOTATION_METALAKE = "gravitino.apache.org/metalake"; + + /** The annotation set when Gravitino requests to cancel the job. */ + public static final String ANNOTATION_CANCEL_REQUESTED = "gravitino.apache.org/cancel-requested"; + + // A DNS-1123 label, which a namespace name must be. + private static final Pattern NAMESPACE_PATTERN = + Pattern.compile("[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?"); + + private K8sJobResourceUtils() {} + + /** + * Returns whether the given name is a valid name of a Kubernetes namespace. + * + * @param namespace the namespace name + * @return true if the name is a DNS-1123 label + */ + public static boolean isValidNamespace(String namespace) { + return namespace != null && NAMESPACE_PATTERN.matcher(namespace).matches(); + } + + /** + * Returns the name of the Kubernetes resource of a job. + * + * @param configs the job executor configurations + * @param jobId the Gravitino job id + * @return the resource name + */ + public static String resourceName(K8sJobExecutorConfigs configs, long jobId) { + return configs.namePrefix() + "-" + jobId; + } + + /** + * Returns the labels of the Kubernetes resource of a job. + * + * @param jobId the Gravitino job id + * @return the labels + */ + public static Map<String, String> jobLabels(long jobId) { + return ImmutableMap.of( + LABEL_MANAGED_BY, MANAGED_BY_GRAVITINO, LABEL_JOB_ID, String.valueOf(jobId)); + } + + /** + * Returns whether the Kubernetes resource was created for the given job. + * + * @param resource the resource + * @param jobId the Gravitino job id + * @return true if the resource carries the job id label of the job + */ + public static boolean isOfJob(HasMetadata resource, long jobId) { + Map<String, String> labels = resource.getMetadata().getLabels(); + return labels != null && String.valueOf(jobId).equals(labels.get(LABEL_JOB_ID)); + } + + /** + * Returns whether Gravitino requested to cancel the job of the Kubernetes resource. + * + * @param resource the resource + * @return true if the cancel annotation is set + */ + public static boolean isCancelRequested(HasMetadata resource) { + Map<String, String> annotations = resource.getMetadata().getAnnotations(); + return annotations != null && "true".equals(annotations.get(ANNOTATION_CANCEL_REQUESTED)); + } Review Comment: `resource.getMetadata()` can be null for some `HasMetadata` instances, which would cause an NPE in both `isOfJob` and `isCancelRequested`. Consider making these methods null-safe by checking `resource.getMetadata()` first (returning `false` when metadata is missing); adding a small unit test for the null-metadata case would prevent regressions. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobResourceUtils.java: ########## @@ -0,0 +1,104 @@ +/* + * 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.collect.ImmutableMap; +import io.fabric8.kubernetes.api.model.HasMetadata; +import java.util.Map; +import java.util.regex.Pattern; + +/** The naming and labeling of the Kubernetes resources created for the jobs. */ +public final class K8sJobResourceUtils { + + /** The label that marks the resources managed by Gravitino. */ + public static final String LABEL_MANAGED_BY = "app.kubernetes.io/managed-by"; + + /** The {@link #LABEL_MANAGED_BY} value of the resources managed by Gravitino. */ + public static final String MANAGED_BY_GRAVITINO = "gravitino"; + + /** The label whose value is the Gravitino job id of the resource. */ + public static final String LABEL_JOB_ID = "gravitino.apache.org/job-id"; + + /** The annotation whose value is the metalake of the job. */ + public static final String ANNOTATION_METALAKE = "gravitino.apache.org/metalake"; + + /** The annotation set when Gravitino requests to cancel the job. */ + public static final String ANNOTATION_CANCEL_REQUESTED = "gravitino.apache.org/cancel-requested"; + + // A DNS-1123 label, which a namespace name must be. + private static final Pattern NAMESPACE_PATTERN = + Pattern.compile("[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?"); + + private K8sJobResourceUtils() {} + + /** + * Returns whether the given name is a valid name of a Kubernetes namespace. + * + * @param namespace the namespace name + * @return true if the name is a DNS-1123 label + */ + public static boolean isValidNamespace(String namespace) { + return namespace != null && NAMESPACE_PATTERN.matcher(namespace).matches(); + } + + /** + * Returns the name of the Kubernetes resource of a job. + * + * @param configs the job executor configurations + * @param jobId the Gravitino job id + * @return the resource name + */ + public static String resourceName(K8sJobExecutorConfigs configs, long jobId) { + return configs.namePrefix() + "-" + jobId; + } + + /** + * Returns the labels of the Kubernetes resource of a job. + * + * @param jobId the Gravitino job id + * @return the labels + */ + public static Map<String, String> jobLabels(long jobId) { + return ImmutableMap.of( + LABEL_MANAGED_BY, MANAGED_BY_GRAVITINO, LABEL_JOB_ID, String.valueOf(jobId)); + } + + /** + * Returns whether the Kubernetes resource was created for the given job. + * + * @param resource the resource + * @param jobId the Gravitino job id + * @return true if the resource carries the job id label of the job + */ + public static boolean isOfJob(HasMetadata resource, long jobId) { + Map<String, String> labels = resource.getMetadata().getLabels(); + return labels != null && String.valueOf(jobId).equals(labels.get(LABEL_JOB_ID)); + } Review Comment: `resource.getMetadata()` can be null for some `HasMetadata` instances, which would cause an NPE in both `isOfJob` and `isCancelRequested`. Consider making these methods null-safe by checking `resource.getMetadata()` first (returning `false` when metadata is missing); adding a small unit test for the null-metadata case would prevent regressions. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutorConfigs.java: ########## @@ -0,0 +1,422 @@ +/* + * 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 default namespace of the jobs. */ + public static final String NAMESPACE = "namespace"; + + /** The default namespace. */ + public static final String DEFAULT_NAMESPACE = "default"; + + /** 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 = "statusCacheTtlMs"; + + /** 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 = "noStatusTimeoutMs"; + + /** 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.resourceRetainDurationMs"; + + /** 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.ttlAfterStopMs"; + + /** 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.driverStartTimeoutMs"; + + /** How long the operator waits for the driver to be ready. */ + public static final String SPARK_DRIVER_READY_TIMEOUT_MS = "spark.driverReadyTimeoutMs"; + + /** + * 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); + this.caCertFile = stringValue(configs, CA_CERT_FILE, null); + this.tokenFile = stringValue(configs, TOKEN_FILE, null); + Preconditions.checkArgument( + masterUrl == null || (kubeconfig == null && context == null), + "%s can't be set together with %s or %s", + MASTER_URL, + KUBECONFIG, + CONTEXT); + Preconditions.checkArgument( + masterUrl != null || (caCertFile == null && tokenFile == null), + "%s and %s can only be set together with %s", Review Comment: This validation is correct, but the error message is a bit ambiguous: it reads like `caCertFile` and `tokenFile` must both be set “together with” `masterUrl`, even though the rule being enforced is “neither may be set unless `masterUrl` is set”. Consider rewording to something like “%s and %s can only be set when %s is set” to make the failure clearer to operators. -- 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]
