jerryshao commented on code in PR #13700: URL: https://github.com/apache/gravitino/pull/13700#discussion_r4280619259
########## 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: Good catch, fixed in d7746f31d. A new `SparkApplicationStatusUtils.hasStatus` checks for a non-empty `status` field. Only an application without one fails after `noStatusTimeoutInMs` and is deleted. One whose status has no readable state stays QUEUED with a warning, like an unknown state. Tests: `testUnreadableStatus` (mapping) and `testKeepJobWithUnreadableStatus` (`{"phase": "Running"}` after 601 s is QUEUED and the SparkApplication is kept). ########## 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: Agreed. I reworded the PR description so it reads as best effort, and added a follow-up to the epic (#13554): 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(); Review Comment: You're right, and the comment was wrong: I had assumed `edit` locks the resource version. Fixed in d7746f31d. The mark is now written by an update carrying the resource version that was read (`updateSparkApplication`). On 409 it reads again and decides again (up to 3 retries), so a job that stops between the check and the write isn't marked. The comment now says so. `testCancelJobThatJustStopped` serves a stale copy first, so it goes through the 409 and retry path. I also re-ran the cancel end to end on a real cluster (K8s 1.35, operator 0.9.0). The delete itself can still race with the job finishing, after the mark is written. I left that as it is: the user asked to cancel a job that was still running when it was marked. ########## 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: Fixed in d7746f31d as suggested. When the deletion fails, the removal re-reads the SparkApplication. If it is being deleted (or already gone), the mark is kept and the cancellation succeeds with a warning instead of throwing. Covered by `testCancelWhoseDeletionResponseIsLost`. ########## 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: Fixed in d7746f31d: if the re-read finds the SparkApplication gone, it throws `NoSuchJobException`, so a cancelling job settles as CANCELLED. Covered by `testJobWithoutStatusGoneMeanwhile`. ########## 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: Yes, the RBAC goes into the docs in #13618 (image, docs and IT): `sparkapplications` create/get/list/update/patch/delete, `pods` list/delete and `pods/log` get. Note that `update` is needed now that the cancel mark is written with a version-checked update. Until then, a missing pod permission keeps a failed job STARTED and logs an error on every poll, as you describe. That is the intended trade-off: without the driver lookup, the executor can't tell whether the driver still runs. -- 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]
