jerryshao commented on code in PR #13700: URL: https://github.com/apache/gravitino/pull/13700#discussion_r4280617615
########## 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()); Review Comment: Fixed in d7746f31d. A new `K8sJobResourceUtils.isJobResource` requires the `managed-by=gravitino` label and a job id label that the resource name ends with (`<prefix>-<jobId>`). A SparkApplication that fails it is treated as missing on every GET path: status throws `NoSuchJobException`, cancel throws `NoSuchJobException` without touching it, and stdout returns empty without looking up its driver pod. The cached listing is filtered the same way. `testIgnoreResourceOfAnotherJob` covers both another job's resource and an unmanaged one, and fails if the check always passes. ########## 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()); Review Comment: Changed in d7746f31d, as @yuqi1129 suggested below: a job of another cluster now fails its queries with `IllegalStateException`, so `JobManager` logs it and leaves the job as it is instead of failing it. Its jobs stay unfinished until the config is changed back. The `cluster` config doc now says it must be the same on every server and must not be renamed while jobs run. Covered by `testJobOfAnotherCluster`. ########## 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); Review Comment: Right, only the kept bytes are bounded. In d7746f31d the Javadoc of `PodLogUtils.readLastLines` now says the whole stream is read and the caller bounds it, and the comment in `getJobStdout` says the lines bound the bytes read, while `maxBytes` only bounds the bytes 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)) { + // 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: Agreed, fixed in d7746f31d: `IllegalStateException` instead of `NoSuchJobException`, so the poll only logs it and a cancel fails loudly. Documented on `cluster` that it must be the same on every server and must not be renamed while jobs run. Covered by `testJobOfAnotherCluster`, which also checks that the cancel doesn't mark the job. -- 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]
