[SPARK-33737][K8S] Support getting pod state using Informers + Listers #58489
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| /* | ||
| * 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.spark.scheduler.cluster.k8s | ||
|
|
||
| import io.fabric8.kubernetes.api.model.Pod | ||
| import io.fabric8.kubernetes.client.informers.ResourceEventHandler | ||
|
|
||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.util.Utils | ||
|
|
||
| /** | ||
| * Publishes executor pod updates to [[ExecutorPodsSnapshotsStore]] using the shared informer | ||
| * owned by [[InformerManager]]. Event-driven counterpart of [[ExecutorPodsListerSnapshotSource]], | ||
| * which periodically snapshots the same informer's local cache. | ||
| */ | ||
| class ExecutorPodsInformerSnapshotSource( | ||
| snapshotsStore: ExecutorPodsSnapshotsStore, | ||
| informerManager: InformerManager) | ||
| extends ExecutorPodsSnapshotSource with Logging { | ||
|
|
||
| override def start(applicationId: String): Unit = { | ||
|
Contributor
There was a problem hiding this comment. 8. start() has no double-start guard, and a misleading INFO fires on every normal startup Neither new source's |
||
| informerManager.initInformer(applicationId) | ||
| informerManager.getInformer().addEventHandler(new ExecutorPodsInformer()) | ||
| informerManager.startInformer() | ||
| } | ||
|
|
||
| override def stop(): Unit = { | ||
| Utils.tryLogNonFatalError { | ||
| informerManager.stopInformer() | ||
| } | ||
| } | ||
|
|
||
| private class ExecutorPodsInformer extends ResourceEventHandler[Pod] { | ||
| override def onAdd(pod: Pod): Unit = { | ||
| logDebug(s"Received add executor pod event for pod named ${pod.getMetadata.getName}") | ||
| snapshotsStore.updatePod(pod) | ||
| } | ||
|
|
||
| override def onUpdate(oldPod: Pod, newPod: Pod): Unit = { | ||
|
Contributor
There was a problem hiding this comment. 7. resync > 0 replays N snapshots per round (churn only; off by default) With |
||
| logDebug(s"Received update executor pod event for pod named ${newPod.getMetadata.getName}") | ||
| snapshotsStore.updatePod(newPod) | ||
| } | ||
|
|
||
| override def onDelete(pod: Pod, deletedFinalStateUnknown: Boolean): Unit = { | ||
| logDebug(s"Received delete executor pod event for pod named ${pod.getMetadata.getName}") | ||
| snapshotsStore.updatePod(pod) | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /* | ||
| * 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.spark.scheduler.cluster.k8s | ||
|
|
||
| import java.util.concurrent.{Future, ScheduledExecutorService, TimeUnit} | ||
|
|
||
| import scala.jdk.CollectionConverters._ | ||
|
|
||
| import io.fabric8.kubernetes.api.model.Pod | ||
| import io.fabric8.kubernetes.client.KubernetesClient | ||
| import io.fabric8.kubernetes.client.informers.cache.Lister | ||
|
|
||
| import org.apache.spark.SparkConf | ||
| import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL | ||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.util.{ThreadUtils, Utils} | ||
|
|
||
| /** | ||
| * Periodically snapshots the local cache of the shared [[InformerManager]] and replaces the | ||
| * contents of the [[ExecutorPodsSnapshotsStore]] with the result. Companion to | ||
| * [[ExecutorPodsInformerSnapshotSource]], which pushes updates as informer events arrive. | ||
| */ | ||
| class ExecutorPodsListerSnapshotSource( | ||
| conf: SparkConf, | ||
| kubernetesClient: KubernetesClient, | ||
| snapshotsStore: ExecutorPodsSnapshotsStore, | ||
| informerManager: InformerManager, | ||
| pollingExecutor: ScheduledExecutorService) | ||
| extends ExecutorPodsSnapshotSource with Logging { | ||
|
|
||
| private val pollingInterval = conf.get(KUBERNETES_EXECUTOR_LISTER_POLLING_INTERVAL) | ||
|
|
||
| private var pollingFuture: Future[_] = _ | ||
|
|
||
| override def start(applicationId: String): Unit = { | ||
| informerManager.initInformer(applicationId) | ||
| informerManager.startInformer() | ||
| val lister = new Lister[Pod]( | ||
| informerManager.getInformer().getIndexer, kubernetesClient.getNamespace) | ||
| pollingFuture = pollingExecutor.scheduleWithFixedDelay( | ||
| new PollRunnable(lister), pollingInterval, pollingInterval, TimeUnit.MILLISECONDS) | ||
| } | ||
|
|
||
| override def stop(): Unit = { | ||
| if (pollingFuture != null) { | ||
| pollingFuture.cancel(true) | ||
| pollingFuture = null | ||
| } | ||
| Utils.tryLogNonFatalError { | ||
| informerManager.stopInformer() | ||
| } | ||
| ThreadUtils.shutdown(pollingExecutor) | ||
| } | ||
|
|
||
| private class PollRunnable(lister: Lister[Pod]) extends Runnable { | ||
| override def run(): Unit = Utils.tryLogNonFatalError { | ||
| // The informer is already scoped server-side to app-id + role=executor + non-inactive | ||
| // pods, so we can hand its snapshot to the store as-is. | ||
| snapshotsStore.replaceSnapshot(lister.list().asScala.toSeq) | ||
|
Contributor
There was a problem hiding this comment. 3. The lister poll doesn't check hasSynced(), so an unsynced empty cache would wipe the snapshot store
Skipping the round when |
||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| /* | ||
| * 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.spark.scheduler.cluster.k8s | ||
|
|
||
| /** | ||
| * Publishes snapshots of the set of executor pods that Kubernetes reports as running for an | ||
| * application. Built-in implementations are chosen by | ||
| * [[org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_ENABLE_INFORMER]]. | ||
| */ | ||
| trait ExecutorPodsSnapshotSource { | ||
|
Contributor
There was a problem hiding this comment. 11. The four new types are public with no annotation, inconsistent with adjacent types
|
||
| def start(applicationId: String): Unit | ||
| def stop(): Unit | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| /* | ||
| * 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.spark.scheduler.cluster.k8s | ||
|
|
||
| import io.fabric8.kubernetes.api.model.Pod | ||
| import io.fabric8.kubernetes.client.KubernetesClient | ||
| import io.fabric8.kubernetes.client.informers.SharedIndexInformer | ||
|
|
||
| import org.apache.spark.SparkConf | ||
| import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL | ||
| import org.apache.spark.deploy.k8s.Constants.{SPARK_APP_ID_LABEL, SPARK_EXECUTOR_INACTIVE_LABEL, SPARK_POD_EXECUTOR_ROLE, SPARK_ROLE_LABEL} | ||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.util.Utils | ||
|
|
||
| /** | ||
| * Owns the shared [[SharedIndexInformer]] used by executor pod snapshot sources when the | ||
| * informer-based mode is enabled. The informer is scoped server-side to the current | ||
| * application's executor pods that are not marked inactive, matching the filter set used by | ||
| * [[ExecutorPodsWatchSnapshotSource]] and [[ExecutorPodsPollingSnapshotSource]]. | ||
| */ | ||
| class InformerManager(kubernetesClient: KubernetesClient, conf: SparkConf) | ||
| extends Logging { | ||
|
|
||
| private val resyncInterval = conf.get(KUBERNETES_EXECUTOR_INFORMER_RESYNC_INTERVAL) | ||
| // VisibleForTesting | ||
| private[k8s] var informer: SharedIndexInformer[Pod] = _ | ||
| private var stopped = false | ||
|
|
||
| def initInformer(applicationId: String): Unit = { | ||
|
Contributor
There was a problem hiding this comment. 6. initInformer ignores After |
||
| if (informer == null) { | ||
| logInfo(s"Initializing executor pods informer for application $applicationId") | ||
| informer = kubernetesClient.pods() | ||
| .withLabel(SPARK_APP_ID_LABEL, applicationId) | ||
| .withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE) | ||
| .withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true") | ||
| .runnableInformer(resyncInterval) | ||
|
Contributor
There was a problem hiding this comment. 2. No exceptionHandler: startup errors are not retried, and a mid-run informer death is silent
Please set |
||
| } | ||
| } | ||
|
|
||
| def getInformer(): SharedIndexInformer[Pod] = { | ||
| if (informer == null) { | ||
| throw new IllegalStateException( | ||
| "Informer has not been initialized. Call initInformer() first.") | ||
| } | ||
| informer | ||
| } | ||
|
|
||
| def startInformer(): Unit = { | ||
| if (informer == null) { | ||
| throw new IllegalStateException( | ||
| "Informer has not been initialized. Call initInformer() first.") | ||
| } | ||
| if (stopped) { | ||
| throw new IllegalStateException("Cannot run informer after stopInformer() has been called.") | ||
| } | ||
| if (!informer.isRunning) { | ||
| informer.run() | ||
|
Contributor
There was a problem hiding this comment. 1. startInformer() runs the blocking, unbounded run() on the SparkContext creation thread
Please switch to |
||
| } else { | ||
| logInfo("Informer is already running.") | ||
| } | ||
| } | ||
|
|
||
| def stopInformer(): Unit = { | ||
| if (informer != null) { | ||
| Utils.tryLogNonFatalError { | ||
| informer.close() | ||
| } | ||
| informer = null | ||
| stopped = true | ||
| } | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
4. The three new user-facing configs are undocumented
The three new user-facing configs (
spark.kubernetes.executor.enableInformer,listerPollingInterval,informerResyncInterval) have no entries anywhere under docs/, whilethe closest precedent
spark.kubernetes.executor.apiPollingIntervalis documented in the configtable in docs/running-on-kubernetes.md. The mutual exclusion with the two legacy switches
currently lives only in the config doc string, so users can't discover the switch or the
migration notes from the docs; it's also worth noting there that the informer path requires
both list and watch permissions on pods. Could you add the three entries to
running-on-kubernetes.md?