jerryshao commented on code in PR #13725: URL: https://github.com/apache/gravitino/pull/13725#discussion_r4256800727
########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationStatusUtils.java: ########## @@ -0,0 +1,479 @@ +/* + * 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.spark; + +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.ContainerState; +import io.fabric8.kubernetes.api.model.ContainerStatus; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.fabric8.kubernetes.api.model.Pod; +import java.time.Instant; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import javax.annotation.Nullable; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.k8s.K8sJobResourceUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Maps the state of a SparkApplication of apache/spark-kubernetes-operator to the Gravitino job + * execution info. The state names are the operator's {@code ApplicationStateSummary}. + */ +public final class SparkApplicationStatusUtils { + + /** + * A state of the SparkApplication, an entry of its state transition history. It may carry the + * status of the driver pod the operator observed when entering the state. + */ + private static final class State { + + private final long id; + + private final String summary; + + @Nullable private final Instant time; + + @Nullable private final Instant driverStartedAt; + + @Nullable private final Instant driverFinishedAt; + + private State( + long id, + String summary, + @Nullable Instant time, + @Nullable Instant driverStartedAt, + @Nullable Instant driverFinishedAt) { + this.id = id; + this.summary = summary; + this.time = time; + this.driverStartedAt = driverStartedAt; + this.driverFinishedAt = driverFinishedAt; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(SparkApplicationStatusUtils.class); + + private static final Set<String> QUEUED_STATES = + ImmutableSet.of("Submitted", "ScheduledToRestart", "DriverRequested", "DriverStarted"); + Review Comment: Fixed in e5daa738d. I checked `AppDriverStartObserver` and `PodUtils.isDriverPodStarted` at tag 0.9.0 and agree: `DriverStarted` means the driver container is ready, `DriverReady` only adds the pod's Ready condition. `DriverStarted` moved from `QUEUED_STATES` to `STARTED_STATES`, so the started-time fallback and the unknown-state handling pick it up too. The parameterized expectation is updated, and `testDriverStartedIsStarted` covers the plain case, a running driver pod, a cancellation before `DriverReady` without a pod (now keeps its started time), and an unknown state after `DriverStarted`. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationStatusUtils.java: ########## @@ -0,0 +1,479 @@ +/* + * 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.spark; + +import com.google.common.collect.ImmutableSet; +import io.fabric8.kubernetes.api.model.ContainerState; +import io.fabric8.kubernetes.api.model.ContainerStatus; +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.fabric8.kubernetes.api.model.Pod; +import java.time.Instant; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import javax.annotation.Nullable; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.connector.job.JobExecutionInfo; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.job.k8s.K8sJobResourceUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Maps the state of a SparkApplication of apache/spark-kubernetes-operator to the Gravitino job + * execution info. The state names are the operator's {@code ApplicationStateSummary}. + */ +public final class SparkApplicationStatusUtils { + + /** + * A state of the SparkApplication, an entry of its state transition history. It may carry the + * status of the driver pod the operator observed when entering the state. + */ + private static final class State { + + private final long id; + + private final String summary; + + @Nullable private final Instant time; + + @Nullable private final Instant driverStartedAt; + + @Nullable private final Instant driverFinishedAt; + + private State( + long id, + String summary, + @Nullable Instant time, + @Nullable Instant driverStartedAt, + @Nullable Instant driverFinishedAt) { + this.id = id; + this.summary = summary; + this.time = time; + this.driverStartedAt = driverStartedAt; + this.driverFinishedAt = driverFinishedAt; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(SparkApplicationStatusUtils.class); + + private static final Set<String> QUEUED_STATES = + ImmutableSet.of("Submitted", "ScheduledToRestart", "DriverRequested", "DriverStarted"); + + private static final Set<String> STARTED_STATES = + ImmutableSet.of( + "DriverReady", + "InitializedBelowThresholdExecutors", + "RunningHealthy", + "RunningWithPartialCapacity", + "RunningWithBelowThresholdExecutors"); + + private static final String SUCCEEDED_STATE = "Succeeded"; + + private static final Set<String> FAILED_STATES = + ImmutableSet.of( + "Failed", + "SchedulingFailure", + "DriverEvicted", + "DriverStartTimedOut", + "ExecutorsStartTimedOut", + "DriverReadyTimedOut"); + + // The last states of every application, which don't tell success from failure. + private static final Set<String> STOPPED_STATES = + ImmutableSet.of("ResourceReleased", "TerminatedWithoutReleaseResources"); + + private SparkApplicationStatusUtils() {} + + /** + * Maps the SparkApplication to the job execution info. + * + * <ul> + * <li>A SparkApplication that Gravitino requested to cancel is cancelled once it is being + * deleted or has stopped. + * <li>A SparkApplication being deleted by someone else, for example by its TTL, keeps the + * outcome it reached, or fails if it has none. + * <li>The stopped states walk back through the state transition history to the last outcome. + * <li>A SparkApplication without any status after the timeout fails, for example if its + * namespace isn't watched by the operator. + * <li>An unknown state, of a newer operator, after an outcome is a stopped state too. Before + * any outcome it keeps the job unfinished: started if the application has reached a started + * state, queued otherwise. + * </ul> + * + * <p>The operator records a state when it observes it, which can be a while after the driver + * actually started or finished. So the times are the ones Kubernetes reports for the driver + * container, in seconds, taken from the given driver pod or from the driver pod status the + * operator keeps with some states. Without them, the job started when the application first + * reached a started state, and finished when it reached its outcome. As the driver container + * status is only known once the job finished, the started time of a running job is refined to the + * driver's when it finishes. A job that stopped without an outcome finished when it was deleted, + * or else when it entered its last state. + * + * @param app the SparkApplication + * @param driver the driver pod, or null if unknown + * @param now the current time + * @param noStatusTimeoutMs the time after which a SparkApplication without status fails + * @return the job execution info + */ + public static JobExecutionInfo executionInfoOf( + GenericKubernetesResource app, @Nullable Pod driver, Instant now, long noStatusTimeoutMs) { + List<State> history = stateHistory(app); + State current = history.isEmpty() ? null : history.get(history.size() - 1); + State outcome = lastOutcome(history); + Instant createdAt = parseTime(app.getMetadata().getCreationTimestamp()); + Instant deletedAt = parseTime(app.getMetadata().getDeletionTimestamp()); + + ContainerState driverState = + driver == null ? null : driverContainerState(driver, driverContainerName(app)); + Instant driverStartedAt = null; + Instant driverFinishedAt = null; + if (driverState != null && driverState.getTerminated() != null) { + driverStartedAt = parseTime(driverState.getTerminated().getStartedAt()); + driverFinishedAt = parseTime(driverState.getTerminated().getFinishedAt()); + } else if (driverState != null && driverState.getRunning() != null) { + driverStartedAt = parseTime(driverState.getRunning().getStartedAt()); + } + Instant startedAt = startedAt(history, driverStartedAt, createdAt); + // When the application stopped: when it reached its outcome, else when it was deleted, else + // when it entered its current state. A deletion time is in whole seconds, so it may read + // earlier than a started time of the same second, which Gravitino would then drop. + Instant stoppedAt; + if (outcome != null) { + stoppedAt = finishedAt(outcome, driverFinishedAt); + } else if (deletedAt != null) { Review Comment: Fixed in e5daa738d. The driver container's termination time is now the finished time whenever it is known and the status is terminal, ahead of the outcome, deletion and current-state times. The status classification is unchanged: `testDriverTerminationTimeWithoutOutcome` covers a cancellation and an external deletion without an operator outcome (CANCELLED and FAILED, both finished at the container's time), the fallback to the deletion time without a pod, and that a terminated pod alone doesn't finish a running job. As you note, part 3 only fetches the driver pod once `isStopped(app)` is true, so it doesn't benefit for a deleting application yet. I'll make #13700 fetch the pod for a deleting application too when it is rebased onto this. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ -- 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]
