Dynamic Allocators in Kubernetes: Scaling Issues

[foxish]

This issue will summarize the various points of discussion being carried out in email threads and the SIG.
This is with respect to the Dynamic Allocation PR (#224):

Spark tries to scale up rather fast, and in an exponential fashion. If you see this section, Spark's behavior is evident; it does not wait for the executors to register before requesting double that number.

The time spent between individual attempts is governed by two properties: spark.dynamicAllocation.schedulerBacklogTimeout and spark.dynamicAllocation.sustainedSchedulerBacklogTimeout. The former is set to 1s by default. So far, our strategy was to match Spark's asking for extra executors with creating new pending executor pods. However, that gets out of hand very quickly and we run the risk of DOSing the cluster.

cc @ssuchter @apache-spark-on-k8s/contributors

[mccheah]

My thought was to have the scheduler behave somewhat like this (syntax is pseudo-code roughly matching Scala):

class KubernetesClusterSchedulerBackend {

    var baseExecutorLabels = Map("appId" -> kubernetesAppId)
    var k8s = createK8s()
    var totalExpectedExecutors = computeInitialExecutors()
    var executorRequestRound = 0L
    var executorCountWatch: Watcher[Pod] = None
    var adjustExecutorsTask = None
    val runningExecutorPods = new HashSet[Pod]

    def start(): Unit = {
        executorCountWatch = Some(k8s.watch(baseExecutorLabels, new Watcher[Pod]) {
            override def eventReceived(action: Action, pod: Pod): Unit = {
                action match {
                    case ADDED if pod.isReady =>
                        runningExecutorPods += pod
                    case DELETED | ERROR if runningExecutorPods.contains(pod) =>
                        runningExecutorPods -= pod
                }
            }
        })
        // hand-waved asynchronous execution
        adjustExecutorsTask = Some(periodically(5, TimeUnit.SECONDS) {
            adjustTotalExecutors()
        })
    }

    def stop(): Unit = {
        executorCountWatch.foreach(_.close)
        adjustExecutorsTask.foreach(_.cancel)
    }

    def doRequestTotalExecutors(expected: Int): Future[Boolean] = Future[Boolean] {
        executorCountWatch.map {
            totalExpectedExecutors = expected
            true
        }.getOrElse(false)
    }

    def adjustTotalExecutors(): Unit = {
        if (runningExecutorPods.size < totalExpectedExecutors) {
            val resolvedExecutorLabels = baseExecutorLabels + ("requestRound" -> executorRequestRound)
            executorRequestRound += 1
            val desiredExecutorsInCurrentRound = totalExpectedExecutors - runningExecutorPods.size
            var readyExecutorsInCurrentRound = 0
            val executorsInCurrentRoundReadySignal = SettableFuture.create[Boolean]
            val podsInCurrentRoundReadyWatcher = k8s.watch(resolvedExecutorLabels, new Watch[Pod] {
                override def eventReceived(action: Action, pod: Pod): Unit = {
                    action match {
                        case ADDED | MODIFIED =>
                            readyExecutorsInCurrentRound += 1
                            if (readyExecutorsInCurrentRound == desiredExecutorsInCurrentRound) {
                                executorsInCurrentRoundReadySignal.set(true)
                            }
                        case DELETED | ERROR => 
                            executorsInCurrentRoundReadySignal.setException(new SparkException(...))
                    }
                }
            })
            for (_ <- 0 until desiredExecutorsInCurrentRound) {
                createExecutorPod()
            }
            try {
                executorsInCurrentRoundReadySignal.get(...)
            } catch {
                case t: Throwable => // Some kind of error handling here
            }
        } else if (runningExecutorPods.size > totalExpectedExecutors) {
            // Delete running executors according to the difference
        }
    }
}

Idea here is to have a global record of the registered executors so far, but on every allocation round we block until the given set of executors reaches a final state, either ERROR or RUNNING.

Some open questions:

• Error handling if executors on a given round hit the DELETED or ERROR state
• What should the delay interval between allocations be?

[foxish]

Thanks @mccheah. I added a new commit, 65bc504; PTAL. I think it's rather similar in spirit to what you wrote up above. I'll be glad to incorporate the missing parts from the above into that.

Error handling if executors on a given round hit the DELETED or ERROR state

I think the DELETED state error handling will be something @varunkatta's PR will address.
Makes sense to roll the two into this though.

What should the delay interval between allocations be?

The delay interval is 1s for now which is reasonable IMO. I'm not sure what we'd gain by changing the poll interval.
It is polling the state (derived from the watch) for all launched executors entering the "ready" state which isn't expensive or reaching the APIServer.

[mccheah]

The PR is a little different from what's proposed here - in particular, the PR's logic has the allocation thread continue without regard to whether or not the most recent request round succeeded in launching executors. The PR also uses an executor limit count as well as an expected executor count; this executor limit count implicitly prevents the allocator thread from asking for more executors if the previous set of executors hadn't been allocated to reach the expected number.

However, I think we should lean towards maintaining less global state and fewer counters, as well as make the rate-limiting explicit in the executor allocation thread. The above approach does both of these by rate-limiting explicitly by using a separate watch in the allocator thread, as well as not having the need for an executor limit vs. expected executors construct.

I think the result of taking these two tradeoffs as described makes the logic easier to follow. What do you think?

[foxish]

I agree with your reasoning. The fact that we have less global state is definitely good.

I think tracking the executors per-round may not work well in practice, because pods allocated in previous rounds may crash and be recovered independently, and we don't have any distinguishing way of finding those pods launched in previous rounds as opposed to those launched in the most recent round. So, in place of watching for just pods in the previous round, we can watch for pod readiness for all pods launched so far (or even use the registered executor count which Spark exposes as an indicator of that metric). Other than that, I agree with what you propose.

[erikerlandson]

I'm in agreement with @foxish that it is simpler and effective to just block if any executors are pending, regardless of which previous round they are from.

This may be an edge case, but it seems possible that one round (say, adding 128 executors) succeeds, and then the next round it goes to double it, and so tries to add another 128. Is it possible that trying to add the 128 thrashes the kube API? Not sure what the threshold would be. If not 128, then 1024, etc.

In some earlier experiments I addressed this problem by increasing the value for spark.dynamicAllocation.sustainedSchedulerBacklogTimeout. Inspired by that idea, maybe when pending executors are detected, it could do some kind of random exponential interval backoff. If it detects pending executors, wait 5 seconds, then 10, etc. However, would only want to do this for increasing executors, not decreasing.

[foxish]

Closing, as we've resolved this and the PR is merged.