yuqi1129 commented on code in PR #13725: URL: https://github.com/apache/gravitino/pull/13725#discussion_r4256769896
########## 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: [Important] `DriverStarted` should represent execution rather than queueing. In operator 0.9.0, `AppDriverStartObserver` records this state when `PodUtils.isDriverPodStarted` sees the main driver container ready, whereas `DriverReady` waits for the whole Pod's Ready condition. For example, an unready sidecar can keep the application at `DriverStarted` while the main program is already executing. This differs from Gravitino's `QUEUED`/`STARTED` contract and the local executor, which reports STARTED after launching the process. I reproduced two consequences on head 25bd776: - `Submitted -> DriverRequested -> DriverStarted`, with a running driver Pod supplied, still returns `QUEUED`. - The same history, cancelled before `DriverReady`, with a deletion timestamp and no driver Pod supplied, returns `CANCELLED` with a null `startedAt`, despite the recorded `DriverStarted` transition. Please move `DriverStarted` from `QUEUED_STATES` into `STARTED_STATES`, so the started-time fallback and unknown-state history handling also recognize it. Update the existing parameterized expectation and add coverage for cancellation without a Pod and an unknown state following `DriverStarted`. These were direct utility probes and source checks; I did not run a real sidecar deployment. Operator source: https://github.com/apache/spark-kubernetes-operator/blob/0.9.0/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/PodUtils.java ########## 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: [Important] A known driver termination time is discarded when the operator has not recorded an outcome yet. `driverFinishedAt` is read above, but only used in the `outcome != null` branch. A direct probe on head 25bd776 reproduces this with `Submitted -> DriverReady -> RunningHealthy`, a cancel annotation, deletion requested at t=20s, and a supplied terminated driver Pod whose `finishedAt` is t=25s. The result is `CANCELLED` with `finishedAt=t=20s`, even though the actual container termination time is available. The non-cancelled external-deletion branch shares this fallback. This contradicts the execution-info contract and this method's stated preference for actual driver timestamps. Please prefer a non-null actual driver termination time when producing terminal execution info, and use outcome/deletion/current-state timestamps only as fallbacks. Keep the existing status classification; a terminated Pod alone should not imply SUCCEEDED. Add tests for both cancellation and external deletion without an operator outcome. Validation boundary: this is a reproduced public utility-input issue, not an end-to-end executor reproduction. Part 3 currently fetches the driver only when `isStopped(app)` is true, so Pod retrieval for a deleting but still-running CR needs separate verification there. The mapper change itself can stay within this class and its tests. -- 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]
