yuqi1129 commented on code in PR #13700: URL: https://github.com/apache/gravitino/pull/13700#discussion_r4280416878
########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java: ########## @@ -0,0 +1,568 @@ +/* + * 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.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.DeletionPropagation; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.GenericKubernetesResourceList; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.ResourceNotFoundException; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.time.Clock; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; +import org.apache.gravitino.connector.job.JobContext; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.connector.job.JobExecutor; +import org.apache.gravitino.exceptions.NoSuchJobException; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.job.k8s.spark.SparkApplicationStatusUtils; +import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A job executor that runs jobs on Kubernetes. Spark jobs run as SparkApplication resources of <a + * href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a> + * 0.9.x, shell jobs aren't supported yet. + * + * <p>All the job state lives in Kubernetes, so every Gravitino server can query and cancel any job. + * The job execution id is {@code <cluster>/<kind>/<namespace>/<name>}, see {@link + * K8sJobExecutionId}. + */ +public class K8sJobExecutor implements JobExecutor { + + private static final Logger LOG = LoggerFactory.getLogger(K8sJobExecutor.class); + + // Not defined by HttpURLConnection. + private static final int HTTP_UNPROCESSABLE_ENTITY = 422; + + // The responses of the API server that reject a resource as it is, so creating it again can't + // succeed: bad request, forbidden, not found (the namespace or the SparkApplication CRD), and + // unprocessable entity. + private static final Set<Integer> REJECTED_REQUEST_CODES = + ImmutableSet.of( + HttpURLConnection.HTTP_BAD_REQUEST, + HttpURLConnection.HTTP_FORBIDDEN, + HttpURLConnection.HTTP_NOT_FOUND, + HTTP_UNPROCESSABLE_ENTITY); + + // How long the status queries fail right away after the API server couldn't be reached. + // Gravitino queries the jobs one after another, and each query would otherwise wait for the + // connection to time out again. + @VisibleForTesting static final long UNREACHABLE_BACKOFF_MS = 10_000L; + + private final Clock clock; + + // The failure of the last status query that couldn't reach the API server, and when it failed. + // Reset once the backoff is over. + @Nullable private volatile KubernetesClientException unreachableFailure; + + private volatile long unreachableAtMs; + + private K8sJobExecutorConfigs configs; + + private KubernetesClient client; + + private NamespacedResourceCache<GenericKubernetesResource> sparkApplicationCache; + + /** Creates the job executor, which Gravitino initializes with {@link #initialize(Map)}. */ + public K8sJobExecutor() { + this(Clock.systemUTC()); + } + + @VisibleForTesting + K8sJobExecutor(Clock clock) { + this.clock = clock; + } + + @Override + public void initialize(Map<String, String> configs) { + K8sJobExecutorConfigs parsed = new K8sJobExecutorConfigs(configs); + initialize(parsed, K8sClientUtils.createClient(parsed)); + } + + @VisibleForTesting + void initialize(K8sJobExecutorConfigs configs, KubernetesClient client) { + this.configs = configs; + this.client = client; + this.sparkApplicationCache = + new NamespacedResourceCache<>( + configs.statusCacheTtlMs(), + namespace -> + getSparkApplicationClient(namespace) + .withLabel( + K8sJobResourceUtils.LABEL_MANAGED_BY, + K8sJobResourceUtils.MANAGED_BY_GRAVITINO) + .list() + .getItems(), + // Getting the SparkApplication itself couldn't reach the API server either. + e -> + e instanceof KubernetesClientException + && isUnreachable((KubernetesClientException) e), + clock::millis); + LOG.info( + "Initialized the k8s job executor for cluster {} at {}, default namespace {}", + configs.cluster(), + client.getMasterUrl(), + configs.namespace()); + } + + @Override + public String submitJob(JobContext context, JobTemplate jobTemplate) { + if (!(jobTemplate instanceof SparkJobTemplate)) { + throw new IllegalArgumentException( + String.format( + "The k8s job executor only supports Spark jobs, but job template %s is a %s job", + jobTemplate.name(), jobTemplate.jobType())); + } + + GenericKubernetesResource app = + SparkApplicationUtils.buildSparkApplication( + context, (SparkJobTemplate) jobTemplate, configs); + String namespace = app.getMetadata().getNamespace(); + String name = app.getMetadata().getName(); + K8sJobExecutionId jobExecutionId = + K8sJobExecutionId.of( + configs.cluster(), K8sJobExecutionId.Kind.SPARK_APPLICATION, namespace, name); + try { + getSparkApplicationClient(namespace).resource(app).create(); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + // The SparkApplication may have been created although no response came back. Gravitino + // records no job for a failed submission, so delete it, otherwise it would run unknown. + deleteUnsubmittedSparkApplication(jobExecutionId, context.jobId()); + throw e; + } + if (REJECTED_REQUEST_CODES.contains(e.getCode())) { + // The cluster rejects the job as it is, for example its namespace doesn't exist or isn't + // allowed, so the caller gets the reason rather than an internal error. + throw new IllegalArgumentException( + String.format( + "Kubernetes rejected the SparkApplication %s/%s of job template %s: %s", + namespace, name, jobTemplate.name(), e.getMessage()), + e); + } + if (e.getCode() != HttpURLConnection.HTTP_CONFLICT) { + throw e; + } + // Created by an earlier attempt to submit the same job. Never adopt a resource of another + // job, for example one left by another Gravitino deployment sharing the namespace, nor one + // that is being deleted. + GenericKubernetesResource existing = + getSparkApplicationClient(namespace).withName(name).get(); + if (existing == null + || existing.getMetadata().getDeletionTimestamp() != null + || !K8sJobResourceUtils.isOfJob(existing, context.jobId())) { + throw new IllegalStateException( + String.format( + "SparkApplication %s/%s already exists and can't be used for job %s. If it belongs" + + " to another Gravitino deployment sharing the namespace, use a different %s" + + " for each of them", + namespace, name, context.jobId(), K8sJobExecutorConfigs.NAME_PREFIX), + e); + } + } + + LOG.info("Submitted job {} as {}", context.jobId(), jobExecutionId); + return jobExecutionId.toString(); + } + + @Override + public JobExecutionInfo getJobExecutionInfo(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + checkApiServerReachable(); + try { + return getSparkApplicationExecutionInfo(id, jobId); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + unreachableAtMs = clock.millis(); + unreachableFailure = e; + } + throw e; + } + } + + @Override + public void cancelJob(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + GenericKubernetesResource app = + getSparkApplicationClient(id.namespace()).withName(id.name()).get(); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + if (SparkApplicationStatusUtils.isStopped(app)) { + return; + } + + // Mark the cancellation before deleting, so that a status query on any server tells a + // cancelled job from one deleted by someone else. The job may have stopped since it was read + // above, then it isn't marked, so that a finished job isn't reported as cancelled. + AtomicBoolean stopped = new AtomicBoolean(); + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + stopped.set(SparkApplicationStatusUtils.isStopped(current)); + if (!stopped.get()) { + setCancelRequested(current, true); + } + return current; + }); + } catch (ResourceNotFoundException e) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_NOT_FOUND) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } + throw e; + } + if (stopped.get()) { + return; + } + + try { + deleteSparkApplication(id); + } catch (KubernetesClientException e) { + // The job keeps running and the caller is told the cancellation failed, so remove the mark, + // otherwise the job would be reported as cancelled once it finishes by itself. + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + setCancelRequested(current, false); + return current; + }); + } catch (RuntimeException removal) { + e.addSuppressed(removal); + } + throw e; + } + LOG.info("Requested to cancel job {}", jobId); + } + + @Override + public List<String> getJobStdout(String jobId, int maxLines, int maxBytes) { + try { + K8sJobExecutionId id = parseJobExecutionId(jobId); + // The SparkApplication tells the driver pod apart from other pods with the same labels, and + // names the driver container. It may already be gone, the driver pod is retained longer. + GenericKubernetesResource app = getSparkApplication(id).orElse(null); + Optional<Pod> driver = getDriverPod(id, app); + if (!driver.isPresent()) { + return ImmutableList.of(); + } + + // Kubernetes can only limit the bytes from the head of the log, so read the last lines and + // keep their tail. + PodResource pod = + client.pods().inNamespace(id.namespace()).withName(driver.get().getMetadata().getName()); + // The container must be named if the driver pod has others, for example a sidecar. + List<Container> containers = + driver.get().getSpec() == null ? null : driver.get().getSpec().getContainers(); + String driverContainer = + app == null + ? SparkApplicationUtils.DRIVER_CONTAINER_NAME + : SparkApplicationStatusUtils.getDriverContainerName(app); + boolean nameContainer = containers != null && containers.size() > 1; + try (InputStream log = + (nameContainer ? pod.inContainer(driverContainer) : pod) + .tailingLines(maxLines) + .getLogInputStream()) { + return PodLogUtils.readLastLines(log, maxLines, maxBytes); + } + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_BAD_REQUEST) { + // For example, the driver container hasn't started yet. + LOG.debug("No driver log of job {} to read yet: {}", jobId, e.getMessage()); + } else { + LOG.warn("Failed to read the driver log of job {}", jobId, e); + } + return ImmutableList.of(); + } catch (RuntimeException | IOException e) { + // The output is best effort and must never fail the caller. + LOG.warn("Failed to read the driver log of job {}", jobId, e); + return ImmutableList.of(); + } + } + + /** + * Returns an empty list: Kubernetes merges the standard error into the standard output of a pod, + * so it is returned by {@link #getJobStdout(String, int, int)}. + */ + @Override + public List<String> getJobStderr(String jobId, int maxLines, int maxBytes) { + return ImmutableList.of(); + } + + @Override + public void close() { + if (client != null) { + client.close(); + } + } + + /** + * Returns whether a request failed because the API server couldn't be reached or didn't answer, + * rather than because the API server refused or failed the request. + * + * @param e the failure of the request + * @return true if the API server gave no response + */ + @VisibleForTesting + static boolean isUnreachable(KubernetesClientException e) { + if (e.getStatus() != null) { + return false; + } + boolean ioFailure = false; + for (Throwable cause = e.getCause(); cause != null; cause = cause.getCause()) { + // A response that isn't JSON, for example the login page of a proxy, did come back. + if (cause instanceof JsonProcessingException) { + return false; + } + ioFailure |= cause instanceof IOException; + } + return ioFailure; + } + + /** + * Fails right away if the API server couldn't be reached a moment ago, so that querying the + * status of many jobs doesn't wait for the connection to time out once per job. + */ + private void checkApiServerReachable() { + KubernetesClientException failure = unreachableFailure; + if (failure == null) { + return; + } + long remainingMs = unreachableAtMs + UNREACHABLE_BACKOFF_MS - clock.millis(); + if (remainingMs <= 0) { + unreachableFailure = null; + return; + } + throw new KubernetesClientException( + String.format( + "The API server of cluster %s couldn't be reached a moment ago, the status of its jobs" + + " isn't queried again for %s ms", + configs.cluster(), remainingMs), + failure); + } + + private JobExecutionInfo getSparkApplicationExecutionInfo(K8sJobExecutionId id, String jobId) { + GenericKubernetesResource app = + sparkApplicationCache + .find(id.namespace(), id.name()) + .orElseGet(() -> getSparkApplicationClient(id.namespace()).withName(id.name()).get()); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + return getExecutionInfo(id, jobId, app); + } + + private JobExecutionInfo getExecutionInfo( + K8sJobExecutionId id, String jobId, GenericKubernetesResource app) { + // The operator observes the driver finishing a while after it does, so a finished job takes + // its times from the driver pod, which is retained after the job finishes. So does a job whose + // SparkApplication is being deleted, for example a cancelled one, whose driver may have + // terminated before the operator recorded any outcome. + Pod driver = null; + KubernetesClientException driverFailure = null; + if (SparkApplicationStatusUtils.isStopped(app) + || app.getMetadata().getDeletionTimestamp() != null) { + try { + driver = getDriverPod(id, app).orElse(null); + } catch (KubernetesClientException e) { + driverFailure = e; + } + } + JobExecutionInfo info = + SparkApplicationStatusUtils.executionInfoOf( + app, driver, clock.instant(), configs.noStatusTimeoutMs()); + if (info.status() == JobHandle.Status.FAILED Review Comment: **[Important] Deleting a job "without status" is decided by "no state this version can parse", not by "the operator never wrote a status".** `hasState()` is false whenever `stateHistory()` parses nothing, including when a status exists in a shape this version doesn't understand (a newer operator, or a status written without `currentState`). Then `executionInfoOf` returns FAILED once `noStatusTimeoutInMs` (10 min) has passed, and `getExecutionInfoWithoutStatus` **deletes the SparkApplication**, even for a job that is running normally. Probe: a job whose status is `{"phase": "Running"}` is reported FAILED after 601 s, and its SparkApplication is gone. This also contradicts the rule for unknown states in `SparkApplicationStatusUtils`, which keeps the job unfinished when it can't tell. Suggestion: only take the no-status timeout path when the `status` field is absent. With a status that can't be parsed, keep the job QUEUED/STARTED and warn. Please add a test for it. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java: ########## @@ -0,0 +1,568 @@ +/* + * 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.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.DeletionPropagation; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.GenericKubernetesResourceList; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.ResourceNotFoundException; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.time.Clock; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; +import org.apache.gravitino.connector.job.JobContext; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.connector.job.JobExecutor; +import org.apache.gravitino.exceptions.NoSuchJobException; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.job.k8s.spark.SparkApplicationStatusUtils; +import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A job executor that runs jobs on Kubernetes. Spark jobs run as SparkApplication resources of <a + * href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a> + * 0.9.x, shell jobs aren't supported yet. + * + * <p>All the job state lives in Kubernetes, so every Gravitino server can query and cancel any job. + * The job execution id is {@code <cluster>/<kind>/<namespace>/<name>}, see {@link + * K8sJobExecutionId}. + */ +public class K8sJobExecutor implements JobExecutor { + + private static final Logger LOG = LoggerFactory.getLogger(K8sJobExecutor.class); + + // Not defined by HttpURLConnection. + private static final int HTTP_UNPROCESSABLE_ENTITY = 422; + + // The responses of the API server that reject a resource as it is, so creating it again can't + // succeed: bad request, forbidden, not found (the namespace or the SparkApplication CRD), and + // unprocessable entity. + private static final Set<Integer> REJECTED_REQUEST_CODES = + ImmutableSet.of( + HttpURLConnection.HTTP_BAD_REQUEST, + HttpURLConnection.HTTP_FORBIDDEN, + HttpURLConnection.HTTP_NOT_FOUND, + HTTP_UNPROCESSABLE_ENTITY); + + // How long the status queries fail right away after the API server couldn't be reached. + // Gravitino queries the jobs one after another, and each query would otherwise wait for the + // connection to time out again. + @VisibleForTesting static final long UNREACHABLE_BACKOFF_MS = 10_000L; + + private final Clock clock; + + // The failure of the last status query that couldn't reach the API server, and when it failed. + // Reset once the backoff is over. + @Nullable private volatile KubernetesClientException unreachableFailure; + + private volatile long unreachableAtMs; + + private K8sJobExecutorConfigs configs; + + private KubernetesClient client; + + private NamespacedResourceCache<GenericKubernetesResource> sparkApplicationCache; + + /** Creates the job executor, which Gravitino initializes with {@link #initialize(Map)}. */ + public K8sJobExecutor() { + this(Clock.systemUTC()); + } + + @VisibleForTesting + K8sJobExecutor(Clock clock) { + this.clock = clock; + } + + @Override + public void initialize(Map<String, String> configs) { + K8sJobExecutorConfigs parsed = new K8sJobExecutorConfigs(configs); + initialize(parsed, K8sClientUtils.createClient(parsed)); + } + + @VisibleForTesting + void initialize(K8sJobExecutorConfigs configs, KubernetesClient client) { + this.configs = configs; + this.client = client; + this.sparkApplicationCache = + new NamespacedResourceCache<>( + configs.statusCacheTtlMs(), + namespace -> + getSparkApplicationClient(namespace) + .withLabel( + K8sJobResourceUtils.LABEL_MANAGED_BY, + K8sJobResourceUtils.MANAGED_BY_GRAVITINO) + .list() + .getItems(), + // Getting the SparkApplication itself couldn't reach the API server either. + e -> + e instanceof KubernetesClientException + && isUnreachable((KubernetesClientException) e), + clock::millis); + LOG.info( + "Initialized the k8s job executor for cluster {} at {}, default namespace {}", + configs.cluster(), + client.getMasterUrl(), + configs.namespace()); + } + + @Override + public String submitJob(JobContext context, JobTemplate jobTemplate) { + if (!(jobTemplate instanceof SparkJobTemplate)) { + throw new IllegalArgumentException( + String.format( + "The k8s job executor only supports Spark jobs, but job template %s is a %s job", + jobTemplate.name(), jobTemplate.jobType())); + } + + GenericKubernetesResource app = + SparkApplicationUtils.buildSparkApplication( + context, (SparkJobTemplate) jobTemplate, configs); + String namespace = app.getMetadata().getNamespace(); + String name = app.getMetadata().getName(); + K8sJobExecutionId jobExecutionId = + K8sJobExecutionId.of( + configs.cluster(), K8sJobExecutionId.Kind.SPARK_APPLICATION, namespace, name); + try { + getSparkApplicationClient(namespace).resource(app).create(); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + // The SparkApplication may have been created although no response came back. Gravitino + // records no job for a failed submission, so delete it, otherwise it would run unknown. + deleteUnsubmittedSparkApplication(jobExecutionId, context.jobId()); + throw e; + } + if (REJECTED_REQUEST_CODES.contains(e.getCode())) { + // The cluster rejects the job as it is, for example its namespace doesn't exist or isn't + // allowed, so the caller gets the reason rather than an internal error. + throw new IllegalArgumentException( + String.format( + "Kubernetes rejected the SparkApplication %s/%s of job template %s: %s", + namespace, name, jobTemplate.name(), e.getMessage()), + e); + } + if (e.getCode() != HttpURLConnection.HTTP_CONFLICT) { + throw e; + } + // Created by an earlier attempt to submit the same job. Never adopt a resource of another + // job, for example one left by another Gravitino deployment sharing the namespace, nor one + // that is being deleted. + GenericKubernetesResource existing = + getSparkApplicationClient(namespace).withName(name).get(); + if (existing == null + || existing.getMetadata().getDeletionTimestamp() != null + || !K8sJobResourceUtils.isOfJob(existing, context.jobId())) { + throw new IllegalStateException( + String.format( + "SparkApplication %s/%s already exists and can't be used for job %s. If it belongs" + + " to another Gravitino deployment sharing the namespace, use a different %s" + + " for each of them", + namespace, name, context.jobId(), K8sJobExecutorConfigs.NAME_PREFIX), + e); + } + } + + LOG.info("Submitted job {} as {}", context.jobId(), jobExecutionId); + return jobExecutionId.toString(); + } + + @Override + public JobExecutionInfo getJobExecutionInfo(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + checkApiServerReachable(); + try { + return getSparkApplicationExecutionInfo(id, jobId); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + unreachableAtMs = clock.millis(); + unreachableFailure = e; + } + throw e; + } + } + + @Override + public void cancelJob(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + GenericKubernetesResource app = + getSparkApplicationClient(id.namespace()).withName(id.name()).get(); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + if (SparkApplicationStatusUtils.isStopped(app)) { + return; + } + + // Mark the cancellation before deleting, so that a status query on any server tells a + // cancelled job from one deleted by someone else. The job may have stopped since it was read + // above, then it isn't marked, so that a finished job isn't reported as cancelled. + AtomicBoolean stopped = new AtomicBoolean(); Review Comment: **[Nit] Marking the cancellation isn't atomic with the stopped check, so a job that just succeeded can be reported as CANCELLED.** In fabric8 7.2.0, `edit()` is a GET followed by a JSON diff patch (`HasMetadataOperation.edit` → `patch(null, clone, item)`), and the patch carries no resourceVersion test. If the application reaches `Succeeded` between that GET and the PATCH, or between the PATCH and the delete below, it gets the annotation, and `SparkApplicationStatusUtils.executionInfoOf` reports CANCELLED, overriding the real outcome. The window is one or two round trips, so it's rare, but the comment above claims more than the code guarantees. Suggestion (either one): in the mapping, keep an outcome recorded before the `deletionTimestamp` (or with no deletion) rather than reporting CANCELLED, which covers both windows; or lock the patch with `lockResourceVersion`. Please also make the comment accurate. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java: ########## @@ -0,0 +1,568 @@ +/* + * 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.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.DeletionPropagation; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.GenericKubernetesResourceList; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.ResourceNotFoundException; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.time.Clock; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; +import org.apache.gravitino.connector.job.JobContext; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.connector.job.JobExecutor; +import org.apache.gravitino.exceptions.NoSuchJobException; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.job.k8s.spark.SparkApplicationStatusUtils; +import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A job executor that runs jobs on Kubernetes. Spark jobs run as SparkApplication resources of <a + * href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a> + * 0.9.x, shell jobs aren't supported yet. + * + * <p>All the job state lives in Kubernetes, so every Gravitino server can query and cancel any job. + * The job execution id is {@code <cluster>/<kind>/<namespace>/<name>}, see {@link + * K8sJobExecutionId}. + */ +public class K8sJobExecutor implements JobExecutor { + + private static final Logger LOG = LoggerFactory.getLogger(K8sJobExecutor.class); + + // Not defined by HttpURLConnection. + private static final int HTTP_UNPROCESSABLE_ENTITY = 422; + + // The responses of the API server that reject a resource as it is, so creating it again can't + // succeed: bad request, forbidden, not found (the namespace or the SparkApplication CRD), and + // unprocessable entity. + private static final Set<Integer> REJECTED_REQUEST_CODES = + ImmutableSet.of( + HttpURLConnection.HTTP_BAD_REQUEST, + HttpURLConnection.HTTP_FORBIDDEN, + HttpURLConnection.HTTP_NOT_FOUND, + HTTP_UNPROCESSABLE_ENTITY); + + // How long the status queries fail right away after the API server couldn't be reached. + // Gravitino queries the jobs one after another, and each query would otherwise wait for the + // connection to time out again. + @VisibleForTesting static final long UNREACHABLE_BACKOFF_MS = 10_000L; + + private final Clock clock; + + // The failure of the last status query that couldn't reach the API server, and when it failed. + // Reset once the backoff is over. + @Nullable private volatile KubernetesClientException unreachableFailure; + + private volatile long unreachableAtMs; + + private K8sJobExecutorConfigs configs; + + private KubernetesClient client; + + private NamespacedResourceCache<GenericKubernetesResource> sparkApplicationCache; + + /** Creates the job executor, which Gravitino initializes with {@link #initialize(Map)}. */ + public K8sJobExecutor() { + this(Clock.systemUTC()); + } + + @VisibleForTesting + K8sJobExecutor(Clock clock) { + this.clock = clock; + } + + @Override + public void initialize(Map<String, String> configs) { + K8sJobExecutorConfigs parsed = new K8sJobExecutorConfigs(configs); + initialize(parsed, K8sClientUtils.createClient(parsed)); + } + + @VisibleForTesting + void initialize(K8sJobExecutorConfigs configs, KubernetesClient client) { + this.configs = configs; + this.client = client; + this.sparkApplicationCache = + new NamespacedResourceCache<>( + configs.statusCacheTtlMs(), + namespace -> + getSparkApplicationClient(namespace) + .withLabel( + K8sJobResourceUtils.LABEL_MANAGED_BY, + K8sJobResourceUtils.MANAGED_BY_GRAVITINO) + .list() + .getItems(), + // Getting the SparkApplication itself couldn't reach the API server either. + e -> + e instanceof KubernetesClientException + && isUnreachable((KubernetesClientException) e), + clock::millis); + LOG.info( + "Initialized the k8s job executor for cluster {} at {}, default namespace {}", + configs.cluster(), + client.getMasterUrl(), + configs.namespace()); + } + + @Override + public String submitJob(JobContext context, JobTemplate jobTemplate) { + if (!(jobTemplate instanceof SparkJobTemplate)) { + throw new IllegalArgumentException( + String.format( + "The k8s job executor only supports Spark jobs, but job template %s is a %s job", + jobTemplate.name(), jobTemplate.jobType())); + } + + GenericKubernetesResource app = + SparkApplicationUtils.buildSparkApplication( + context, (SparkJobTemplate) jobTemplate, configs); + String namespace = app.getMetadata().getNamespace(); + String name = app.getMetadata().getName(); + K8sJobExecutionId jobExecutionId = + K8sJobExecutionId.of( + configs.cluster(), K8sJobExecutionId.Kind.SPARK_APPLICATION, namespace, name); + try { + getSparkApplicationClient(namespace).resource(app).create(); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { Review Comment: **[Important, can be deferred] The cleanup after an unreachable API server is best effort, so a SparkApplication can be left running with no Gravitino job.** fabric8 retries IOExceptions and 5xx responses (`requestRetries`, 2 by default), and a retry that hits 409 adopts the resource of the same job, so a lost response is mostly handled already. What's left is the case where every attempt fails: the create may have landed, and the API server being unreachable is exactly when the cleanup GET below fails too. The Spark job then runs untracked, and nothing ever deletes it. A resubmission by the user would run the same work twice, for example a compaction. Suggestion: reword the PR description ("deleted in case it was created" reads like a guarantee), and file a follow-up (or fold it into #13616) to sweep the SparkApplications labelled managed-by Gravitino whose job id isn't in the store. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java: ########## @@ -0,0 +1,568 @@ +/* + * 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.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.DeletionPropagation; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.GenericKubernetesResourceList; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.ResourceNotFoundException; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.time.Clock; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; +import org.apache.gravitino.connector.job.JobContext; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.connector.job.JobExecutor; +import org.apache.gravitino.exceptions.NoSuchJobException; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.job.k8s.spark.SparkApplicationStatusUtils; +import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A job executor that runs jobs on Kubernetes. Spark jobs run as SparkApplication resources of <a + * href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a> + * 0.9.x, shell jobs aren't supported yet. + * + * <p>All the job state lives in Kubernetes, so every Gravitino server can query and cancel any job. + * The job execution id is {@code <cluster>/<kind>/<namespace>/<name>}, see {@link + * K8sJobExecutionId}. + */ +public class K8sJobExecutor implements JobExecutor { + + private static final Logger LOG = LoggerFactory.getLogger(K8sJobExecutor.class); + + // Not defined by HttpURLConnection. + private static final int HTTP_UNPROCESSABLE_ENTITY = 422; + + // The responses of the API server that reject a resource as it is, so creating it again can't + // succeed: bad request, forbidden, not found (the namespace or the SparkApplication CRD), and + // unprocessable entity. + private static final Set<Integer> REJECTED_REQUEST_CODES = + ImmutableSet.of( + HttpURLConnection.HTTP_BAD_REQUEST, + HttpURLConnection.HTTP_FORBIDDEN, + HttpURLConnection.HTTP_NOT_FOUND, + HTTP_UNPROCESSABLE_ENTITY); + + // How long the status queries fail right away after the API server couldn't be reached. + // Gravitino queries the jobs one after another, and each query would otherwise wait for the + // connection to time out again. + @VisibleForTesting static final long UNREACHABLE_BACKOFF_MS = 10_000L; + + private final Clock clock; + + // The failure of the last status query that couldn't reach the API server, and when it failed. + // Reset once the backoff is over. + @Nullable private volatile KubernetesClientException unreachableFailure; + + private volatile long unreachableAtMs; + + private K8sJobExecutorConfigs configs; + + private KubernetesClient client; + + private NamespacedResourceCache<GenericKubernetesResource> sparkApplicationCache; + + /** Creates the job executor, which Gravitino initializes with {@link #initialize(Map)}. */ + public K8sJobExecutor() { + this(Clock.systemUTC()); + } + + @VisibleForTesting + K8sJobExecutor(Clock clock) { + this.clock = clock; + } + + @Override + public void initialize(Map<String, String> configs) { + K8sJobExecutorConfigs parsed = new K8sJobExecutorConfigs(configs); + initialize(parsed, K8sClientUtils.createClient(parsed)); + } + + @VisibleForTesting + void initialize(K8sJobExecutorConfigs configs, KubernetesClient client) { + this.configs = configs; + this.client = client; + this.sparkApplicationCache = + new NamespacedResourceCache<>( + configs.statusCacheTtlMs(), + namespace -> + getSparkApplicationClient(namespace) + .withLabel( + K8sJobResourceUtils.LABEL_MANAGED_BY, + K8sJobResourceUtils.MANAGED_BY_GRAVITINO) + .list() + .getItems(), + // Getting the SparkApplication itself couldn't reach the API server either. + e -> + e instanceof KubernetesClientException + && isUnreachable((KubernetesClientException) e), + clock::millis); + LOG.info( + "Initialized the k8s job executor for cluster {} at {}, default namespace {}", + configs.cluster(), + client.getMasterUrl(), + configs.namespace()); + } + + @Override + public String submitJob(JobContext context, JobTemplate jobTemplate) { + if (!(jobTemplate instanceof SparkJobTemplate)) { + throw new IllegalArgumentException( + String.format( + "The k8s job executor only supports Spark jobs, but job template %s is a %s job", + jobTemplate.name(), jobTemplate.jobType())); + } + + GenericKubernetesResource app = + SparkApplicationUtils.buildSparkApplication( + context, (SparkJobTemplate) jobTemplate, configs); + String namespace = app.getMetadata().getNamespace(); + String name = app.getMetadata().getName(); + K8sJobExecutionId jobExecutionId = + K8sJobExecutionId.of( + configs.cluster(), K8sJobExecutionId.Kind.SPARK_APPLICATION, namespace, name); + try { + getSparkApplicationClient(namespace).resource(app).create(); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + // The SparkApplication may have been created although no response came back. Gravitino + // records no job for a failed submission, so delete it, otherwise it would run unknown. + deleteUnsubmittedSparkApplication(jobExecutionId, context.jobId()); + throw e; + } + if (REJECTED_REQUEST_CODES.contains(e.getCode())) { + // The cluster rejects the job as it is, for example its namespace doesn't exist or isn't + // allowed, so the caller gets the reason rather than an internal error. + throw new IllegalArgumentException( + String.format( + "Kubernetes rejected the SparkApplication %s/%s of job template %s: %s", + namespace, name, jobTemplate.name(), e.getMessage()), + e); + } + if (e.getCode() != HttpURLConnection.HTTP_CONFLICT) { + throw e; + } + // Created by an earlier attempt to submit the same job. Never adopt a resource of another + // job, for example one left by another Gravitino deployment sharing the namespace, nor one + // that is being deleted. + GenericKubernetesResource existing = + getSparkApplicationClient(namespace).withName(name).get(); + if (existing == null + || existing.getMetadata().getDeletionTimestamp() != null + || !K8sJobResourceUtils.isOfJob(existing, context.jobId())) { + throw new IllegalStateException( + String.format( + "SparkApplication %s/%s already exists and can't be used for job %s. If it belongs" + + " to another Gravitino deployment sharing the namespace, use a different %s" + + " for each of them", + namespace, name, context.jobId(), K8sJobExecutorConfigs.NAME_PREFIX), + e); + } + } + + LOG.info("Submitted job {} as {}", context.jobId(), jobExecutionId); + return jobExecutionId.toString(); + } + + @Override + public JobExecutionInfo getJobExecutionInfo(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + checkApiServerReachable(); + try { + return getSparkApplicationExecutionInfo(id, jobId); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + unreachableAtMs = clock.millis(); + unreachableFailure = e; + } + throw e; + } + } + + @Override + public void cancelJob(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + GenericKubernetesResource app = + getSparkApplicationClient(id.namespace()).withName(id.name()).get(); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + if (SparkApplicationStatusUtils.isStopped(app)) { + return; + } + + // Mark the cancellation before deleting, so that a status query on any server tells a + // cancelled job from one deleted by someone else. The job may have stopped since it was read + // above, then it isn't marked, so that a finished job isn't reported as cancelled. + AtomicBoolean stopped = new AtomicBoolean(); + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + stopped.set(SparkApplicationStatusUtils.isStopped(current)); + if (!stopped.get()) { + setCancelRequested(current, true); + } + return current; + }); + } catch (ResourceNotFoundException e) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_NOT_FOUND) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } + throw e; + } + if (stopped.get()) { + return; + } + + try { + deleteSparkApplication(id); + } catch (KubernetesClientException e) { + // The job keeps running and the caller is told the cancellation failed, so remove the mark, + // otherwise the job would be reported as cancelled once it finishes by itself. + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + setCancelRequested(current, false); + return current; + }); + } catch (RuntimeException removal) { + e.addSuppressed(removal); + } + throw e; + } + LOG.info("Requested to cancel job {}", jobId); + } + + @Override + public List<String> getJobStdout(String jobId, int maxLines, int maxBytes) { + try { + K8sJobExecutionId id = parseJobExecutionId(jobId); + // The SparkApplication tells the driver pod apart from other pods with the same labels, and + // names the driver container. It may already be gone, the driver pod is retained longer. + GenericKubernetesResource app = getSparkApplication(id).orElse(null); + Optional<Pod> driver = getDriverPod(id, app); + if (!driver.isPresent()) { + return ImmutableList.of(); + } + + // Kubernetes can only limit the bytes from the head of the log, so read the last lines and + // keep their tail. + PodResource pod = + client.pods().inNamespace(id.namespace()).withName(driver.get().getMetadata().getName()); + // The container must be named if the driver pod has others, for example a sidecar. + List<Container> containers = + driver.get().getSpec() == null ? null : driver.get().getSpec().getContainers(); + String driverContainer = + app == null + ? SparkApplicationUtils.DRIVER_CONTAINER_NAME + : SparkApplicationStatusUtils.getDriverContainerName(app); + boolean nameContainer = containers != null && containers.size() > 1; + try (InputStream log = + (nameContainer ? pod.inContainer(driverContainer) : pod) + .tailingLines(maxLines) + .getLogInputStream()) { + return PodLogUtils.readLastLines(log, maxLines, maxBytes); + } + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_BAD_REQUEST) { + // For example, the driver container hasn't started yet. + LOG.debug("No driver log of job {} to read yet: {}", jobId, e.getMessage()); + } else { + LOG.warn("Failed to read the driver log of job {}", jobId, e); + } + return ImmutableList.of(); + } catch (RuntimeException | IOException e) { + // The output is best effort and must never fail the caller. + LOG.warn("Failed to read the driver log of job {}", jobId, e); + return ImmutableList.of(); + } + } + + /** + * Returns an empty list: Kubernetes merges the standard error into the standard output of a pod, + * so it is returned by {@link #getJobStdout(String, int, int)}. + */ + @Override + public List<String> getJobStderr(String jobId, int maxLines, int maxBytes) { + return ImmutableList.of(); + } + + @Override + public void close() { + if (client != null) { + client.close(); + } + } + + /** + * Returns whether a request failed because the API server couldn't be reached or didn't answer, + * rather than because the API server refused or failed the request. + * + * @param e the failure of the request + * @return true if the API server gave no response + */ + @VisibleForTesting + static boolean isUnreachable(KubernetesClientException e) { + if (e.getStatus() != null) { + return false; + } + boolean ioFailure = false; + for (Throwable cause = e.getCause(); cause != null; cause = cause.getCause()) { + // A response that isn't JSON, for example the login page of a proxy, did come back. + if (cause instanceof JsonProcessingException) { + return false; + } + ioFailure |= cause instanceof IOException; + } + return ioFailure; + } + + /** + * Fails right away if the API server couldn't be reached a moment ago, so that querying the + * status of many jobs doesn't wait for the connection to time out once per job. + */ + private void checkApiServerReachable() { + KubernetesClientException failure = unreachableFailure; + if (failure == null) { + return; + } + long remainingMs = unreachableAtMs + UNREACHABLE_BACKOFF_MS - clock.millis(); + if (remainingMs <= 0) { + unreachableFailure = null; + return; + } + throw new KubernetesClientException( + String.format( + "The API server of cluster %s couldn't be reached a moment ago, the status of its jobs" + + " isn't queried again for %s ms", + configs.cluster(), remainingMs), + failure); + } + + private JobExecutionInfo getSparkApplicationExecutionInfo(K8sJobExecutionId id, String jobId) { + GenericKubernetesResource app = + sparkApplicationCache + .find(id.namespace(), id.name()) + .orElseGet(() -> getSparkApplicationClient(id.namespace()).withName(id.name()).get()); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + return getExecutionInfo(id, jobId, app); + } + + private JobExecutionInfo getExecutionInfo( + K8sJobExecutionId id, String jobId, GenericKubernetesResource app) { + // The operator observes the driver finishing a while after it does, so a finished job takes + // its times from the driver pod, which is retained after the job finishes. So does a job whose + // SparkApplication is being deleted, for example a cancelled one, whose driver may have + // terminated before the operator recorded any outcome. + Pod driver = null; + KubernetesClientException driverFailure = null; + if (SparkApplicationStatusUtils.isStopped(app) + || app.getMetadata().getDeletionTimestamp() != null) { + try { + driver = getDriverPod(id, app).orElse(null); + } catch (KubernetesClientException e) { + driverFailure = e; + } + } + JobExecutionInfo info = + SparkApplicationStatusUtils.executionInfoOf( + app, driver, clock.instant(), configs.noStatusTimeoutMs()); + if (info.status() == JobHandle.Status.FAILED + && !SparkApplicationStatusUtils.hasState(app) + && app.getMetadata().getDeletionTimestamp() == null) { + return getExecutionInfoWithoutStatus(id, jobId, info); + } + if (info.status() == JobHandle.Status.FAILED && driverFailure != null) { + // Whether the driver still runs can't be told, so the job isn't failed yet: a later query + // checks its driver again. + throw driverFailure; + } + if (driverFailure != null) { + LOG.warn("Failed to get the driver pod of job {}", jobId, driverFailure); + } + if (info.status() == JobHandle.Status.FAILED + && driver != null + && SparkApplicationStatusUtils.isDriverRunning(app, driver)) { + // The operator retains the driver pod of a stopped application, even one that is still + // pending or running after the application timed out starting. Delete it, otherwise it + // could still run the job after it failed. A failure to delete it fails this status query, + // so that the job only fails once the pod is deleted, by a later query. + LOG.warn( + "Deleting the driver pod {} of failed job {}, as it hasn't terminated", + driver.getMetadata().getName(), + jobId); + client.pods().inNamespace(id.namespace()).withName(driver.getMetadata().getName()).delete(); + } + return info; + } + + private K8sJobExecutionId parseJobExecutionId(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = K8sJobExecutionId.parse(jobId); + if (!configs.cluster().equals(id.cluster())) { + throw new NoSuchJobException( + "Job %s runs in cluster %s, but the k8s job executor is connected to cluster %s", + jobId, id.cluster(), configs.cluster()); + } + return id; + } + + /** + * Returns the execution info of a job whose SparkApplication got no status from the operator in + * time, which is failed. The SparkApplication is deleted first, otherwise the operator would + * still run it once it comes back, with the job already failed and out of Gravitino's reach. A + * failure to delete it fails the status query, so that the job only fails once it is deleted, by + * a later query. + */ + private JobExecutionInfo getExecutionInfoWithoutStatus( + K8sJobExecutionId id, String jobId, JobExecutionInfo failed) { + // The SparkApplication may come from the cache: it may have got a status since, or be being + // deleted, for example by a cancellation on another server. Then it is mapped as usual. + GenericKubernetesResource latest = + getSparkApplicationClient(id.namespace()).withName(id.name()).get(); + if (latest == null) { Review Comment: **[Nit] This should throw `NoSuchJobException` rather than return FAILED.** `JobManager` settles a CANCELLING job as CANCELLED on `NoSuchJobException`, but as FAILED when FAILED is returned. Probe: server B caches the listing while the job has no status. Server A cancels it, and the SparkApplication is gone. After the timeout, B maps the stale cached copy to FAILED, the GET here returns null, and the user's cancelled job ends up FAILED. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java: ########## @@ -0,0 +1,568 @@ +/* + * 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.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.DeletionPropagation; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.GenericKubernetesResourceList; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.ResourceNotFoundException; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.time.Clock; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; +import org.apache.gravitino.connector.job.JobContext; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.connector.job.JobExecutor; +import org.apache.gravitino.exceptions.NoSuchJobException; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.job.k8s.spark.SparkApplicationStatusUtils; +import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A job executor that runs jobs on Kubernetes. Spark jobs run as SparkApplication resources of <a + * href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a> + * 0.9.x, shell jobs aren't supported yet. + * + * <p>All the job state lives in Kubernetes, so every Gravitino server can query and cancel any job. + * The job execution id is {@code <cluster>/<kind>/<namespace>/<name>}, see {@link + * K8sJobExecutionId}. + */ +public class K8sJobExecutor implements JobExecutor { + + private static final Logger LOG = LoggerFactory.getLogger(K8sJobExecutor.class); + + // Not defined by HttpURLConnection. + private static final int HTTP_UNPROCESSABLE_ENTITY = 422; + + // The responses of the API server that reject a resource as it is, so creating it again can't + // succeed: bad request, forbidden, not found (the namespace or the SparkApplication CRD), and + // unprocessable entity. + private static final Set<Integer> REJECTED_REQUEST_CODES = + ImmutableSet.of( + HttpURLConnection.HTTP_BAD_REQUEST, + HttpURLConnection.HTTP_FORBIDDEN, + HttpURLConnection.HTTP_NOT_FOUND, + HTTP_UNPROCESSABLE_ENTITY); + + // How long the status queries fail right away after the API server couldn't be reached. + // Gravitino queries the jobs one after another, and each query would otherwise wait for the + // connection to time out again. + @VisibleForTesting static final long UNREACHABLE_BACKOFF_MS = 10_000L; + + private final Clock clock; + + // The failure of the last status query that couldn't reach the API server, and when it failed. + // Reset once the backoff is over. + @Nullable private volatile KubernetesClientException unreachableFailure; + + private volatile long unreachableAtMs; + + private K8sJobExecutorConfigs configs; + + private KubernetesClient client; + + private NamespacedResourceCache<GenericKubernetesResource> sparkApplicationCache; + + /** Creates the job executor, which Gravitino initializes with {@link #initialize(Map)}. */ + public K8sJobExecutor() { + this(Clock.systemUTC()); + } + + @VisibleForTesting + K8sJobExecutor(Clock clock) { + this.clock = clock; + } + + @Override + public void initialize(Map<String, String> configs) { + K8sJobExecutorConfigs parsed = new K8sJobExecutorConfigs(configs); + initialize(parsed, K8sClientUtils.createClient(parsed)); + } + + @VisibleForTesting + void initialize(K8sJobExecutorConfigs configs, KubernetesClient client) { + this.configs = configs; + this.client = client; + this.sparkApplicationCache = + new NamespacedResourceCache<>( + configs.statusCacheTtlMs(), + namespace -> + getSparkApplicationClient(namespace) + .withLabel( + K8sJobResourceUtils.LABEL_MANAGED_BY, + K8sJobResourceUtils.MANAGED_BY_GRAVITINO) + .list() + .getItems(), + // Getting the SparkApplication itself couldn't reach the API server either. + e -> + e instanceof KubernetesClientException + && isUnreachable((KubernetesClientException) e), + clock::millis); + LOG.info( + "Initialized the k8s job executor for cluster {} at {}, default namespace {}", + configs.cluster(), + client.getMasterUrl(), + configs.namespace()); + } + + @Override + public String submitJob(JobContext context, JobTemplate jobTemplate) { + if (!(jobTemplate instanceof SparkJobTemplate)) { + throw new IllegalArgumentException( + String.format( + "The k8s job executor only supports Spark jobs, but job template %s is a %s job", + jobTemplate.name(), jobTemplate.jobType())); + } + + GenericKubernetesResource app = + SparkApplicationUtils.buildSparkApplication( + context, (SparkJobTemplate) jobTemplate, configs); + String namespace = app.getMetadata().getNamespace(); + String name = app.getMetadata().getName(); + K8sJobExecutionId jobExecutionId = + K8sJobExecutionId.of( + configs.cluster(), K8sJobExecutionId.Kind.SPARK_APPLICATION, namespace, name); + try { + getSparkApplicationClient(namespace).resource(app).create(); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + // The SparkApplication may have been created although no response came back. Gravitino + // records no job for a failed submission, so delete it, otherwise it would run unknown. + deleteUnsubmittedSparkApplication(jobExecutionId, context.jobId()); + throw e; + } + if (REJECTED_REQUEST_CODES.contains(e.getCode())) { + // The cluster rejects the job as it is, for example its namespace doesn't exist or isn't + // allowed, so the caller gets the reason rather than an internal error. + throw new IllegalArgumentException( + String.format( + "Kubernetes rejected the SparkApplication %s/%s of job template %s: %s", + namespace, name, jobTemplate.name(), e.getMessage()), + e); + } + if (e.getCode() != HttpURLConnection.HTTP_CONFLICT) { + throw e; + } + // Created by an earlier attempt to submit the same job. Never adopt a resource of another + // job, for example one left by another Gravitino deployment sharing the namespace, nor one + // that is being deleted. + GenericKubernetesResource existing = + getSparkApplicationClient(namespace).withName(name).get(); + if (existing == null + || existing.getMetadata().getDeletionTimestamp() != null + || !K8sJobResourceUtils.isOfJob(existing, context.jobId())) { + throw new IllegalStateException( + String.format( + "SparkApplication %s/%s already exists and can't be used for job %s. If it belongs" + + " to another Gravitino deployment sharing the namespace, use a different %s" + + " for each of them", + namespace, name, context.jobId(), K8sJobExecutorConfigs.NAME_PREFIX), + e); + } + } + + LOG.info("Submitted job {} as {}", context.jobId(), jobExecutionId); + return jobExecutionId.toString(); + } + + @Override + public JobExecutionInfo getJobExecutionInfo(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + checkApiServerReachable(); + try { + return getSparkApplicationExecutionInfo(id, jobId); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + unreachableAtMs = clock.millis(); + unreachableFailure = e; + } + throw e; + } + } + + @Override + public void cancelJob(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + GenericKubernetesResource app = + getSparkApplicationClient(id.namespace()).withName(id.name()).get(); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + if (SparkApplicationStatusUtils.isStopped(app)) { + return; + } + + // Mark the cancellation before deleting, so that a status query on any server tells a + // cancelled job from one deleted by someone else. The job may have stopped since it was read + // above, then it isn't marked, so that a finished job isn't reported as cancelled. + AtomicBoolean stopped = new AtomicBoolean(); + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + stopped.set(SparkApplicationStatusUtils.isStopped(current)); + if (!stopped.get()) { + setCancelRequested(current, true); + } + return current; + }); + } catch (ResourceNotFoundException e) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_NOT_FOUND) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } + throw e; + } + if (stopped.get()) { + return; + } + + try { + deleteSparkApplication(id); + } catch (KubernetesClientException e) { + // The job keeps running and the caller is told the cancellation failed, so remove the mark, Review Comment: **[Nit] If the delete took effect but its response was lost, removing the mark turns a cancelled job into FAILED.** In that case the SparkApplication is being deleted, the mark is removed here, and the status mapping then treats it as deleted by someone else: FAILED with no outcome. Meanwhile `JobManager.cancelJob` got the exception, so it never writes CANCELLING. Suggestion: in the removal lambda, if `current.getMetadata().getDeletionTimestamp() != null`, keep the mark and treat the cancellation as done instead of rethrowing. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java: ########## @@ -0,0 +1,568 @@ +/* + * 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.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.DeletionPropagation; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.GenericKubernetesResourceList; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.ResourceNotFoundException; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.time.Clock; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; +import org.apache.gravitino.connector.job.JobContext; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.connector.job.JobExecutor; +import org.apache.gravitino.exceptions.NoSuchJobException; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.job.k8s.spark.SparkApplicationStatusUtils; +import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A job executor that runs jobs on Kubernetes. Spark jobs run as SparkApplication resources of <a + * href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a> + * 0.9.x, shell jobs aren't supported yet. + * + * <p>All the job state lives in Kubernetes, so every Gravitino server can query and cancel any job. + * The job execution id is {@code <cluster>/<kind>/<namespace>/<name>}, see {@link + * K8sJobExecutionId}. + */ +public class K8sJobExecutor implements JobExecutor { + + private static final Logger LOG = LoggerFactory.getLogger(K8sJobExecutor.class); + + // Not defined by HttpURLConnection. + private static final int HTTP_UNPROCESSABLE_ENTITY = 422; + + // The responses of the API server that reject a resource as it is, so creating it again can't + // succeed: bad request, forbidden, not found (the namespace or the SparkApplication CRD), and + // unprocessable entity. + private static final Set<Integer> REJECTED_REQUEST_CODES = + ImmutableSet.of( + HttpURLConnection.HTTP_BAD_REQUEST, + HttpURLConnection.HTTP_FORBIDDEN, + HttpURLConnection.HTTP_NOT_FOUND, + HTTP_UNPROCESSABLE_ENTITY); + + // How long the status queries fail right away after the API server couldn't be reached. + // Gravitino queries the jobs one after another, and each query would otherwise wait for the + // connection to time out again. + @VisibleForTesting static final long UNREACHABLE_BACKOFF_MS = 10_000L; + + private final Clock clock; + + // The failure of the last status query that couldn't reach the API server, and when it failed. + // Reset once the backoff is over. + @Nullable private volatile KubernetesClientException unreachableFailure; + + private volatile long unreachableAtMs; + + private K8sJobExecutorConfigs configs; + + private KubernetesClient client; + + private NamespacedResourceCache<GenericKubernetesResource> sparkApplicationCache; + + /** Creates the job executor, which Gravitino initializes with {@link #initialize(Map)}. */ + public K8sJobExecutor() { + this(Clock.systemUTC()); + } + + @VisibleForTesting + K8sJobExecutor(Clock clock) { + this.clock = clock; + } + + @Override + public void initialize(Map<String, String> configs) { + K8sJobExecutorConfigs parsed = new K8sJobExecutorConfigs(configs); + initialize(parsed, K8sClientUtils.createClient(parsed)); + } + + @VisibleForTesting + void initialize(K8sJobExecutorConfigs configs, KubernetesClient client) { + this.configs = configs; + this.client = client; + this.sparkApplicationCache = + new NamespacedResourceCache<>( + configs.statusCacheTtlMs(), + namespace -> + getSparkApplicationClient(namespace) + .withLabel( + K8sJobResourceUtils.LABEL_MANAGED_BY, + K8sJobResourceUtils.MANAGED_BY_GRAVITINO) + .list() + .getItems(), + // Getting the SparkApplication itself couldn't reach the API server either. + e -> + e instanceof KubernetesClientException + && isUnreachable((KubernetesClientException) e), + clock::millis); + LOG.info( + "Initialized the k8s job executor for cluster {} at {}, default namespace {}", + configs.cluster(), + client.getMasterUrl(), + configs.namespace()); + } + + @Override + public String submitJob(JobContext context, JobTemplate jobTemplate) { + if (!(jobTemplate instanceof SparkJobTemplate)) { + throw new IllegalArgumentException( + String.format( + "The k8s job executor only supports Spark jobs, but job template %s is a %s job", + jobTemplate.name(), jobTemplate.jobType())); + } + + GenericKubernetesResource app = + SparkApplicationUtils.buildSparkApplication( + context, (SparkJobTemplate) jobTemplate, configs); + String namespace = app.getMetadata().getNamespace(); + String name = app.getMetadata().getName(); + K8sJobExecutionId jobExecutionId = + K8sJobExecutionId.of( + configs.cluster(), K8sJobExecutionId.Kind.SPARK_APPLICATION, namespace, name); + try { + getSparkApplicationClient(namespace).resource(app).create(); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + // The SparkApplication may have been created although no response came back. Gravitino + // records no job for a failed submission, so delete it, otherwise it would run unknown. + deleteUnsubmittedSparkApplication(jobExecutionId, context.jobId()); + throw e; + } + if (REJECTED_REQUEST_CODES.contains(e.getCode())) { + // The cluster rejects the job as it is, for example its namespace doesn't exist or isn't + // allowed, so the caller gets the reason rather than an internal error. + throw new IllegalArgumentException( + String.format( + "Kubernetes rejected the SparkApplication %s/%s of job template %s: %s", + namespace, name, jobTemplate.name(), e.getMessage()), + e); + } + if (e.getCode() != HttpURLConnection.HTTP_CONFLICT) { + throw e; + } + // Created by an earlier attempt to submit the same job. Never adopt a resource of another + // job, for example one left by another Gravitino deployment sharing the namespace, nor one + // that is being deleted. + GenericKubernetesResource existing = + getSparkApplicationClient(namespace).withName(name).get(); + if (existing == null + || existing.getMetadata().getDeletionTimestamp() != null + || !K8sJobResourceUtils.isOfJob(existing, context.jobId())) { + throw new IllegalStateException( + String.format( + "SparkApplication %s/%s already exists and can't be used for job %s. If it belongs" + + " to another Gravitino deployment sharing the namespace, use a different %s" + + " for each of them", + namespace, name, context.jobId(), K8sJobExecutorConfigs.NAME_PREFIX), + e); + } + } + + LOG.info("Submitted job {} as {}", context.jobId(), jobExecutionId); + return jobExecutionId.toString(); + } + + @Override + public JobExecutionInfo getJobExecutionInfo(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + checkApiServerReachable(); + try { + return getSparkApplicationExecutionInfo(id, jobId); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + unreachableAtMs = clock.millis(); + unreachableFailure = e; + } + throw e; + } + } + + @Override + public void cancelJob(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + GenericKubernetesResource app = + getSparkApplicationClient(id.namespace()).withName(id.name()).get(); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + if (SparkApplicationStatusUtils.isStopped(app)) { + return; + } + + // Mark the cancellation before deleting, so that a status query on any server tells a + // cancelled job from one deleted by someone else. The job may have stopped since it was read + // above, then it isn't marked, so that a finished job isn't reported as cancelled. + AtomicBoolean stopped = new AtomicBoolean(); + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + stopped.set(SparkApplicationStatusUtils.isStopped(current)); + if (!stopped.get()) { + setCancelRequested(current, true); + } + return current; + }); + } catch (ResourceNotFoundException e) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_NOT_FOUND) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } + throw e; + } + if (stopped.get()) { + return; + } + + try { + deleteSparkApplication(id); + } catch (KubernetesClientException e) { + // The job keeps running and the caller is told the cancellation failed, so remove the mark, + // otherwise the job would be reported as cancelled once it finishes by itself. + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + setCancelRequested(current, false); + return current; + }); + } catch (RuntimeException removal) { + e.addSuppressed(removal); + } + throw e; + } + LOG.info("Requested to cancel job {}", jobId); + } + + @Override + public List<String> getJobStdout(String jobId, int maxLines, int maxBytes) { + try { + K8sJobExecutionId id = parseJobExecutionId(jobId); + // The SparkApplication tells the driver pod apart from other pods with the same labels, and + // names the driver container. It may already be gone, the driver pod is retained longer. + GenericKubernetesResource app = getSparkApplication(id).orElse(null); + Optional<Pod> driver = getDriverPod(id, app); + if (!driver.isPresent()) { + return ImmutableList.of(); + } + + // Kubernetes can only limit the bytes from the head of the log, so read the last lines and + // keep their tail. + PodResource pod = + client.pods().inNamespace(id.namespace()).withName(driver.get().getMetadata().getName()); + // The container must be named if the driver pod has others, for example a sidecar. + List<Container> containers = + driver.get().getSpec() == null ? null : driver.get().getSpec().getContainers(); + String driverContainer = + app == null + ? SparkApplicationUtils.DRIVER_CONTAINER_NAME + : SparkApplicationStatusUtils.getDriverContainerName(app); + boolean nameContainer = containers != null && containers.size() > 1; + try (InputStream log = + (nameContainer ? pod.inContainer(driverContainer) : pod) + .tailingLines(maxLines) + .getLogInputStream()) { + return PodLogUtils.readLastLines(log, maxLines, maxBytes); + } + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_BAD_REQUEST) { + // For example, the driver container hasn't started yet. + LOG.debug("No driver log of job {} to read yet: {}", jobId, e.getMessage()); + } else { + LOG.warn("Failed to read the driver log of job {}", jobId, e); + } + return ImmutableList.of(); + } catch (RuntimeException | IOException e) { + // The output is best effort and must never fail the caller. + LOG.warn("Failed to read the driver log of job {}", jobId, e); + return ImmutableList.of(); + } + } + + /** + * Returns an empty list: Kubernetes merges the standard error into the standard output of a pod, + * so it is returned by {@link #getJobStdout(String, int, int)}. + */ + @Override + public List<String> getJobStderr(String jobId, int maxLines, int maxBytes) { + return ImmutableList.of(); + } + + @Override + public void close() { + if (client != null) { + client.close(); + } + } + + /** + * Returns whether a request failed because the API server couldn't be reached or didn't answer, + * rather than because the API server refused or failed the request. + * + * @param e the failure of the request + * @return true if the API server gave no response + */ + @VisibleForTesting + static boolean isUnreachable(KubernetesClientException e) { + if (e.getStatus() != null) { + return false; + } + boolean ioFailure = false; + for (Throwable cause = e.getCause(); cause != null; cause = cause.getCause()) { + // A response that isn't JSON, for example the login page of a proxy, did come back. + if (cause instanceof JsonProcessingException) { + return false; + } + ioFailure |= cause instanceof IOException; + } + return ioFailure; + } + + /** + * Fails right away if the API server couldn't be reached a moment ago, so that querying the + * status of many jobs doesn't wait for the connection to time out once per job. + */ + private void checkApiServerReachable() { + KubernetesClientException failure = unreachableFailure; + if (failure == null) { + return; + } + long remainingMs = unreachableAtMs + UNREACHABLE_BACKOFF_MS - clock.millis(); + if (remainingMs <= 0) { + unreachableFailure = null; + return; + } + throw new KubernetesClientException( + String.format( + "The API server of cluster %s couldn't be reached a moment ago, the status of its jobs" + + " isn't queried again for %s ms", + configs.cluster(), remainingMs), + failure); + } + + private JobExecutionInfo getSparkApplicationExecutionInfo(K8sJobExecutionId id, String jobId) { + GenericKubernetesResource app = + sparkApplicationCache + .find(id.namespace(), id.name()) + .orElseGet(() -> getSparkApplicationClient(id.namespace()).withName(id.name()).get()); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + return getExecutionInfo(id, jobId, app); + } + + private JobExecutionInfo getExecutionInfo( + K8sJobExecutionId id, String jobId, GenericKubernetesResource app) { + // The operator observes the driver finishing a while after it does, so a finished job takes + // its times from the driver pod, which is retained after the job finishes. So does a job whose + // SparkApplication is being deleted, for example a cancelled one, whose driver may have + // terminated before the operator recorded any outcome. + Pod driver = null; + KubernetesClientException driverFailure = null; + if (SparkApplicationStatusUtils.isStopped(app) + || app.getMetadata().getDeletionTimestamp() != null) { + try { + driver = getDriverPod(id, app).orElse(null); + } catch (KubernetesClientException e) { + driverFailure = e; + } + } + JobExecutionInfo info = + SparkApplicationStatusUtils.executionInfoOf( + app, driver, clock.instant(), configs.noStatusTimeoutMs()); + if (info.status() == JobHandle.Status.FAILED + && !SparkApplicationStatusUtils.hasState(app) + && app.getMetadata().getDeletionTimestamp() == null) { + return getExecutionInfoWithoutStatus(id, jobId, info); + } + if (info.status() == JobHandle.Status.FAILED && driverFailure != null) { + // Whether the driver still runs can't be told, so the job isn't failed yet: a later query + // checks its driver again. + throw driverFailure; + } + if (driverFailure != null) { + LOG.warn("Failed to get the driver pod of job {}", jobId, driverFailure); + } + if (info.status() == JobHandle.Status.FAILED + && driver != null + && SparkApplicationStatusUtils.isDriverRunning(app, driver)) { + // The operator retains the driver pod of a stopped application, even one that is still + // pending or running after the application timed out starting. Delete it, otherwise it + // could still run the job after it failed. A failure to delete it fails this status query, + // so that the job only fails once the pod is deleted, by a later query. + LOG.warn( + "Deleting the driver pod {} of failed job {}, as it hasn't terminated", + driver.getMetadata().getName(), + jobId); + client.pods().inNamespace(id.namespace()).withName(driver.getMetadata().getName()).delete(); + } + return info; + } + + private K8sJobExecutionId parseJobExecutionId(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = K8sJobExecutionId.parse(jobId); + if (!configs.cluster().equals(id.cluster())) { Review Comment: **[Important] A cluster mismatch is reported as `NoSuchJobException`, which `JobManager` turns into a terminal state while the job keeps running.** `JobManager.pullAndUpdateOwnedJobStatus` maps `NoSuchJobException` to FAILED, and `JobManager.cancelJob` maps it to CANCELLING, which the next poll settles as CANCELLED. Since `ownsJob()` isn't overridden, every server polls every job. So if `gravitino.jobExecutor.k8s.cluster` is renamed, or two servers briefly have different values (for example during a rolling config change), every active job of the other cluster gets marked FAILED, or falsely CANCELLED, while its SparkApplication keeps running, and nothing ever corrects it. A probe against the mock server confirms that both `getJobExecutionInfo` and `cancelJob` throw `NoSuchJobException` for `other/...` ids. Suggestion: throw a non-terminal exception here (e.g. `IllegalStateException`), so the poll only logs it and keeps the job, and a cancel fails loudly. Overriding `ownsJob()` instead wouldn't work, as a CANCELLING job is only acted on by its owner for node-local executors. Also document that `cluster` must be the same on every server and must not be renamed while jobs run. ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/K8sJobExecutor.java: ########## @@ -0,0 +1,568 @@ +/* + * 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.fasterxml.jackson.core.JsonProcessingException; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.DeletionPropagation; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.GenericKubernetesResourceList; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.ResourceNotFoundException; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.PodResource; +import io.fabric8.kubernetes.client.dsl.Resource; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.time.Clock; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; +import org.apache.gravitino.connector.job.JobContext; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.connector.job.JobExecutor; +import org.apache.gravitino.exceptions.NoSuchJobException; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.JobTemplate; +import org.apache.gravitino.job.SparkJobTemplate; +import org.apache.gravitino.job.k8s.spark.SparkApplicationStatusUtils; +import org.apache.gravitino.job.k8s.spark.SparkApplicationUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A job executor that runs jobs on Kubernetes. Spark jobs run as SparkApplication resources of <a + * href="https://github.com/apache/spark-kubernetes-operator";>apache/spark-kubernetes-operator</a> + * 0.9.x, shell jobs aren't supported yet. + * + * <p>All the job state lives in Kubernetes, so every Gravitino server can query and cancel any job. + * The job execution id is {@code <cluster>/<kind>/<namespace>/<name>}, see {@link + * K8sJobExecutionId}. + */ +public class K8sJobExecutor implements JobExecutor { + + private static final Logger LOG = LoggerFactory.getLogger(K8sJobExecutor.class); + + // Not defined by HttpURLConnection. + private static final int HTTP_UNPROCESSABLE_ENTITY = 422; + + // The responses of the API server that reject a resource as it is, so creating it again can't + // succeed: bad request, forbidden, not found (the namespace or the SparkApplication CRD), and + // unprocessable entity. + private static final Set<Integer> REJECTED_REQUEST_CODES = + ImmutableSet.of( + HttpURLConnection.HTTP_BAD_REQUEST, + HttpURLConnection.HTTP_FORBIDDEN, + HttpURLConnection.HTTP_NOT_FOUND, + HTTP_UNPROCESSABLE_ENTITY); + + // How long the status queries fail right away after the API server couldn't be reached. + // Gravitino queries the jobs one after another, and each query would otherwise wait for the + // connection to time out again. + @VisibleForTesting static final long UNREACHABLE_BACKOFF_MS = 10_000L; + + private final Clock clock; + + // The failure of the last status query that couldn't reach the API server, and when it failed. + // Reset once the backoff is over. + @Nullable private volatile KubernetesClientException unreachableFailure; + + private volatile long unreachableAtMs; + + private K8sJobExecutorConfigs configs; + + private KubernetesClient client; + + private NamespacedResourceCache<GenericKubernetesResource> sparkApplicationCache; + + /** Creates the job executor, which Gravitino initializes with {@link #initialize(Map)}. */ + public K8sJobExecutor() { + this(Clock.systemUTC()); + } + + @VisibleForTesting + K8sJobExecutor(Clock clock) { + this.clock = clock; + } + + @Override + public void initialize(Map<String, String> configs) { + K8sJobExecutorConfigs parsed = new K8sJobExecutorConfigs(configs); + initialize(parsed, K8sClientUtils.createClient(parsed)); + } + + @VisibleForTesting + void initialize(K8sJobExecutorConfigs configs, KubernetesClient client) { + this.configs = configs; + this.client = client; + this.sparkApplicationCache = + new NamespacedResourceCache<>( + configs.statusCacheTtlMs(), + namespace -> + getSparkApplicationClient(namespace) + .withLabel( + K8sJobResourceUtils.LABEL_MANAGED_BY, + K8sJobResourceUtils.MANAGED_BY_GRAVITINO) + .list() + .getItems(), + // Getting the SparkApplication itself couldn't reach the API server either. + e -> + e instanceof KubernetesClientException + && isUnreachable((KubernetesClientException) e), + clock::millis); + LOG.info( + "Initialized the k8s job executor for cluster {} at {}, default namespace {}", + configs.cluster(), + client.getMasterUrl(), + configs.namespace()); + } + + @Override + public String submitJob(JobContext context, JobTemplate jobTemplate) { + if (!(jobTemplate instanceof SparkJobTemplate)) { + throw new IllegalArgumentException( + String.format( + "The k8s job executor only supports Spark jobs, but job template %s is a %s job", + jobTemplate.name(), jobTemplate.jobType())); + } + + GenericKubernetesResource app = + SparkApplicationUtils.buildSparkApplication( + context, (SparkJobTemplate) jobTemplate, configs); + String namespace = app.getMetadata().getNamespace(); + String name = app.getMetadata().getName(); + K8sJobExecutionId jobExecutionId = + K8sJobExecutionId.of( + configs.cluster(), K8sJobExecutionId.Kind.SPARK_APPLICATION, namespace, name); + try { + getSparkApplicationClient(namespace).resource(app).create(); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + // The SparkApplication may have been created although no response came back. Gravitino + // records no job for a failed submission, so delete it, otherwise it would run unknown. + deleteUnsubmittedSparkApplication(jobExecutionId, context.jobId()); + throw e; + } + if (REJECTED_REQUEST_CODES.contains(e.getCode())) { + // The cluster rejects the job as it is, for example its namespace doesn't exist or isn't + // allowed, so the caller gets the reason rather than an internal error. + throw new IllegalArgumentException( + String.format( + "Kubernetes rejected the SparkApplication %s/%s of job template %s: %s", + namespace, name, jobTemplate.name(), e.getMessage()), + e); + } + if (e.getCode() != HttpURLConnection.HTTP_CONFLICT) { + throw e; + } + // Created by an earlier attempt to submit the same job. Never adopt a resource of another + // job, for example one left by another Gravitino deployment sharing the namespace, nor one + // that is being deleted. + GenericKubernetesResource existing = + getSparkApplicationClient(namespace).withName(name).get(); + if (existing == null + || existing.getMetadata().getDeletionTimestamp() != null + || !K8sJobResourceUtils.isOfJob(existing, context.jobId())) { + throw new IllegalStateException( + String.format( + "SparkApplication %s/%s already exists and can't be used for job %s. If it belongs" + + " to another Gravitino deployment sharing the namespace, use a different %s" + + " for each of them", + namespace, name, context.jobId(), K8sJobExecutorConfigs.NAME_PREFIX), + e); + } + } + + LOG.info("Submitted job {} as {}", context.jobId(), jobExecutionId); + return jobExecutionId.toString(); + } + + @Override + public JobExecutionInfo getJobExecutionInfo(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + checkApiServerReachable(); + try { + return getSparkApplicationExecutionInfo(id, jobId); + } catch (KubernetesClientException e) { + if (isUnreachable(e)) { + unreachableAtMs = clock.millis(); + unreachableFailure = e; + } + throw e; + } + } + + @Override + public void cancelJob(String jobId) throws NoSuchJobException { + K8sJobExecutionId id = parseJobExecutionId(jobId); + GenericKubernetesResource app = + getSparkApplicationClient(id.namespace()).withName(id.name()).get(); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + if (SparkApplicationStatusUtils.isStopped(app)) { + return; + } + + // Mark the cancellation before deleting, so that a status query on any server tells a + // cancelled job from one deleted by someone else. The job may have stopped since it was read + // above, then it isn't marked, so that a finished job isn't reported as cancelled. + AtomicBoolean stopped = new AtomicBoolean(); + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + stopped.set(SparkApplicationStatusUtils.isStopped(current)); + if (!stopped.get()) { + setCancelRequested(current, true); + } + return current; + }); + } catch (ResourceNotFoundException e) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_NOT_FOUND) { + throw new NoSuchJobException(e, "SparkApplication of job %s doesn't exist", jobId); + } + throw e; + } + if (stopped.get()) { + return; + } + + try { + deleteSparkApplication(id); + } catch (KubernetesClientException e) { + // The job keeps running and the caller is told the cancellation failed, so remove the mark, + // otherwise the job would be reported as cancelled once it finishes by itself. + try { + getSparkApplicationClient(id.namespace()) + .withName(id.name()) + .edit( + current -> { + setCancelRequested(current, false); + return current; + }); + } catch (RuntimeException removal) { + e.addSuppressed(removal); + } + throw e; + } + LOG.info("Requested to cancel job {}", jobId); + } + + @Override + public List<String> getJobStdout(String jobId, int maxLines, int maxBytes) { + try { + K8sJobExecutionId id = parseJobExecutionId(jobId); + // The SparkApplication tells the driver pod apart from other pods with the same labels, and + // names the driver container. It may already be gone, the driver pod is retained longer. + GenericKubernetesResource app = getSparkApplication(id).orElse(null); + Optional<Pod> driver = getDriverPod(id, app); + if (!driver.isPresent()) { + return ImmutableList.of(); + } + + // Kubernetes can only limit the bytes from the head of the log, so read the last lines and + // keep their tail. + PodResource pod = + client.pods().inNamespace(id.namespace()).withName(driver.get().getMetadata().getName()); + // The container must be named if the driver pod has others, for example a sidecar. + List<Container> containers = + driver.get().getSpec() == null ? null : driver.get().getSpec().getContainers(); + String driverContainer = + app == null + ? SparkApplicationUtils.DRIVER_CONTAINER_NAME + : SparkApplicationStatusUtils.getDriverContainerName(app); + boolean nameContainer = containers != null && containers.size() > 1; + try (InputStream log = + (nameContainer ? pod.inContainer(driverContainer) : pod) + .tailingLines(maxLines) + .getLogInputStream()) { + return PodLogUtils.readLastLines(log, maxLines, maxBytes); + } + } catch (KubernetesClientException e) { + if (e.getCode() == HttpURLConnection.HTTP_BAD_REQUEST) { + // For example, the driver container hasn't started yet. + LOG.debug("No driver log of job {} to read yet: {}", jobId, e.getMessage()); + } else { + LOG.warn("Failed to read the driver log of job {}", jobId, e); + } + return ImmutableList.of(); + } catch (RuntimeException | IOException e) { + // The output is best effort and must never fail the caller. + LOG.warn("Failed to read the driver log of job {}", jobId, e); + return ImmutableList.of(); + } + } + + /** + * Returns an empty list: Kubernetes merges the standard error into the standard output of a pod, + * so it is returned by {@link #getJobStdout(String, int, int)}. + */ + @Override + public List<String> getJobStderr(String jobId, int maxLines, int maxBytes) { + return ImmutableList.of(); + } + + @Override + public void close() { + if (client != null) { + client.close(); + } + } + + /** + * Returns whether a request failed because the API server couldn't be reached or didn't answer, + * rather than because the API server refused or failed the request. + * + * @param e the failure of the request + * @return true if the API server gave no response + */ + @VisibleForTesting + static boolean isUnreachable(KubernetesClientException e) { + if (e.getStatus() != null) { + return false; + } + boolean ioFailure = false; + for (Throwable cause = e.getCause(); cause != null; cause = cause.getCause()) { + // A response that isn't JSON, for example the login page of a proxy, did come back. + if (cause instanceof JsonProcessingException) { + return false; + } + ioFailure |= cause instanceof IOException; + } + return ioFailure; + } + + /** + * Fails right away if the API server couldn't be reached a moment ago, so that querying the + * status of many jobs doesn't wait for the connection to time out once per job. + */ + private void checkApiServerReachable() { + KubernetesClientException failure = unreachableFailure; + if (failure == null) { + return; + } + long remainingMs = unreachableAtMs + UNREACHABLE_BACKOFF_MS - clock.millis(); + if (remainingMs <= 0) { + unreachableFailure = null; + return; + } + throw new KubernetesClientException( + String.format( + "The API server of cluster %s couldn't be reached a moment ago, the status of its jobs" + + " isn't queried again for %s ms", + configs.cluster(), remainingMs), + failure); + } + + private JobExecutionInfo getSparkApplicationExecutionInfo(K8sJobExecutionId id, String jobId) { + GenericKubernetesResource app = + sparkApplicationCache + .find(id.namespace(), id.name()) + .orElseGet(() -> getSparkApplicationClient(id.namespace()).withName(id.name()).get()); + if (app == null) { + throw new NoSuchJobException("SparkApplication of job %s doesn't exist", jobId); + } + return getExecutionInfo(id, jobId, app); + } + + private JobExecutionInfo getExecutionInfo( + K8sJobExecutionId id, String jobId, GenericKubernetesResource app) { + // The operator observes the driver finishing a while after it does, so a finished job takes + // its times from the driver pod, which is retained after the job finishes. So does a job whose + // SparkApplication is being deleted, for example a cancelled one, whose driver may have + // terminated before the operator recorded any outcome. + Pod driver = null; + KubernetesClientException driverFailure = null; + if (SparkApplicationStatusUtils.isStopped(app) + || app.getMetadata().getDeletionTimestamp() != null) { + try { + driver = getDriverPod(id, app).orElse(null); + } catch (KubernetesClientException e) { + driverFailure = e; + } + } + JobExecutionInfo info = + SparkApplicationStatusUtils.executionInfoOf( + app, driver, clock.instant(), configs.noStatusTimeoutMs()); + if (info.status() == JobHandle.Status.FAILED + && !SparkApplicationStatusUtils.hasState(app) + && app.getMetadata().getDeletionTimestamp() == null) { + return getExecutionInfoWithoutStatus(id, jobId, info); + } + if (info.status() == JobHandle.Status.FAILED && driverFailure != null) { Review Comment: **[Question] A FAILED job whose driver pod can't be looked up is never failed if the failure is permanent.** If Gravitino's service account isn't allowed to list pods (403), every failed job throws here on every poll, so it stays STARTED forever and logs an ERROR each time. Neither this PR nor the docs list the RBAC this executor needs. Is that planned for #13616? At least: sparkapplications create/get/list/patch/delete, and pods list/delete plus pods/log get. -- 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]
