Skip to content

Commit

Permalink
[SPARK-37497][K8S] Promote `ExecutorPods[PollingSnapshot|WatchSnapsho…
Browse files Browse the repository at this point in the history
…t]Source` to DeveloperApi

### What changes were proposed in this pull request?

This PR aims to promote `ExecutorPodsWatchSnapshotSource` and `ExecutorPodsPollingSnapshotSource` as **stable** `DeveloperApi` in order to maintain it officially in a backward compatible way at Apache Spark 3.3.0.

### Why are the changes needed?

- Since SPARK-24248 at Apache Spark 2.4.0, `ExecutorPodsWatchSnapshotSource` and `ExecutorPodsPollingSnapshotSource` have been used to monitor executor pods without any interface changes for over 3 years.

- Apache Spark 3.1.1 makes `Kubernetes` module GA and provides an extensible external cluster manager framework. New `ExternalClusterManager` for K8s environment need to depend on this to monitor pods.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Manual review.

Closes #34751 from dongjoon-hyun/SPARK-37497.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 2b04496)
Signed-off-by: Dongjoon Hyun <[email protected]>
  • Loading branch information
dongjoon-hyun committed Dec 1, 2021
1 parent b8b5f94 commit 75cac1f
Show file tree
Hide file tree
Showing 2 changed files with 25 additions and 2 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,21 @@ import io.fabric8.kubernetes.client.KubernetesClient
import scala.collection.JavaConverters._

import org.apache.spark.SparkConf
import org.apache.spark.annotation.{DeveloperApi, Since, Stable}
import org.apache.spark.deploy.k8s.Config._
import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.internal.Logging
import org.apache.spark.util.{ThreadUtils, Utils}

private[spark] class ExecutorPodsPollingSnapshotSource(
/**
* :: DeveloperApi ::
*
* A class used for polling K8s executor pods by ExternalClusterManagers.
* @since 3.1.3
*/
@Stable
@DeveloperApi
class ExecutorPodsPollingSnapshotSource(
conf: SparkConf,
kubernetesClient: KubernetesClient,
snapshotsStore: ExecutorPodsSnapshotsStore,
Expand All @@ -37,13 +46,15 @@ private[spark] class ExecutorPodsPollingSnapshotSource(

private var pollingFuture: Future[_] = _

@Since("3.1.3")
def start(applicationId: String): Unit = {
require(pollingFuture == null, "Cannot start polling more than once.")
logDebug(s"Starting to check for executor pod state every $pollingInterval ms.")
pollingFuture = pollingExecutor.scheduleWithFixedDelay(
new PollRunnable(applicationId), pollingInterval, pollingInterval, TimeUnit.MILLISECONDS)
}

@Since("3.1.3")
def stop(): Unit = {
if (pollingFuture != null) {
pollingFuture.cancel(true)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,16 +22,27 @@ import io.fabric8.kubernetes.api.model.Pod
import io.fabric8.kubernetes.client.{KubernetesClient, KubernetesClientException, Watcher}
import io.fabric8.kubernetes.client.Watcher.Action

import org.apache.spark.annotation.{DeveloperApi, Since, Stable}
import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.internal.Logging
import org.apache.spark.util.Utils

private[spark] class ExecutorPodsWatchSnapshotSource(
/**
* :: DeveloperApi ::
*
* A class used for watching K8s executor pods by ExternalClusterManagers.
*
* @since 3.1.3
*/
@Stable
@DeveloperApi
class ExecutorPodsWatchSnapshotSource(
snapshotsStore: ExecutorPodsSnapshotsStore,
kubernetesClient: KubernetesClient) extends Logging {

private var watchConnection: Closeable = _

@Since("3.1.3")
def start(applicationId: String): Unit = {
require(watchConnection == null, "Cannot start the watcher twice.")
logDebug(s"Starting watch for pods with labels $SPARK_APP_ID_LABEL=$applicationId," +
Expand All @@ -42,6 +53,7 @@ private[spark] class ExecutorPodsWatchSnapshotSource(
.watch(new ExecutorPodsWatcher())
}

@Since("3.1.3")
def stop(): Unit = {
if (watchConnection != null) {
Utils.tryLogNonFatalError {
Expand Down

0 comments on commit 75cac1f

Please sign in to comment.