Dynamic allocation

[foxish]

Dynamic allocation updated
Please see commits individually during review for clarity

cc @mccheah @ash211 @varunkatta @apache-spark-on-k8s/contributors

[lins05]

A corner case comes into my mind: if a shuffle pod for a node died and is being restarted, and at this moment a new executor on that node registers with the driver, it would crash the driver.

Can we improve this, e.g. let the executor die when the shuffle pod is not ready, instead of throwing SparkException to abort the driver?

[foxish]

Nice catch! I think you're right. Done. Passing back an empty string which should make the executor crash.

[mccheah]

Throwing the exception here should be fine, right? This is being processed on a separate thread in the RPC environment. Thus the exception here should only propagate to the executor that asked for this configuration.

[lins05]

I could be wrong, but IIUC SparkException is thrown whenever unrecoverable errors happen.

[mccheah]

In this case it's less important what the type of exception is. What's important is where the exception is thrown from and where it propagates to.

[lins05]

Is a daemonset watchable? If so can we watch on it directly instead of using labels?

[foxish]

It is, but we don't want to make an assumption that it's a daemonset in use. In the current way, it remains - any pod that is co-located on that node with the same labels.

[lins05]

Should we throw an exception here is dynamic allocation is enabled but shuffle.labels is empty?

[foxish]

Done

[lins05]

nit: the name NodeCacheManager is not that intuitive. Maybe sth. like ShufflePodsCatalog? I don't know..

[foxish]

Changed to ShufflePodCache

[mccheah]

Reviewed the top level PR, will follow up on changes in the subsequent individual commits.

[mccheah]

Could probably just use < for clarity.

[mccheah]

Try to structure the logic so that we don't use return.

[mccheah]

Similarly here - avoid return.

[mccheah]

Does this specifically need to be a separate class? The code could just be inlined in the scheduler backend class.

[foxish]

I like the separate class for the separation it offers for this allocator mechanism. Do you strongly prefer not having it?

[mccheah]

This ought to be in a separate file then, I think. But if this also requires a backwards reference to the scheduler backend (e.g. to access fields) then this should just be inlined.

[foxish]

It requires access to totalRegisteredExecutors which is a protected field in CoarseGrainedSchedulerBackend and a couple of other accounting fields from the KubernetesClusterSchedulerBackend.

[foxish]

Oh, did you mean that it shouldn't be a separate class?

[foxish]

Done

[mccheah]

I think we prefer foreach over for in general.

[foxish]

Done

[mccheah]

Let's avoid adding tests to V1 and solely focus on V2.

[foxish]

Done

[mccheah]

val not var.

[foxish]

Done

[mccheah]

Would be good to confirm that this test is creating multiple executors and is writing files to the shuffle service. I'm not sure if we can do this in an automated way.

[lins05]

We can achieve that by setting --conf spark.dynamicAllocation.minExecutors=2 and wait for these two executors to be ready via k8s api (or spark rest api?)

[mccheah]

The executors can spin up but not write any shuffle data to disk. We should check that shuffle data is being written to the disks.

[foxish]

I chose this test so that it would have shuffle data being written to disk. I've manually verified that it does write to disk.

[mccheah]

Just make the interval an Optional configuration and use interval.foreach, as opposed to checking if an integer is greater than zero. We should validate then that any given value is positive and throw an exception if it isn't.

[mccheah]

Can we use .get instead of .getOption here?

[foxish]

I was trying to get Option[String] because if the shuffle directory is left empty, we use the default from Utils.getConfiguredLocalDirs(conf). I'm not sure how we can get this behavior using get.

[mccheah]

Using .get here without using .key on the configuration key should give back an Option.

[foxish]

Ah.. good point. Done

[foxish]

@mccheah Addressed all comments. PTAL

[foxish]

Has the unit testing changed? I'm seeing failures in files I did not touch at all.

[ash211]

rerun unit tests please

[ash211]

@foxish I've been working on an SBT-based unit test build in jenkins and it looks like it was racing with the current maven-based unit tests. I've disabled the new test build and expect just the old one to be running now.

Sorry about that!

[foxish]

@ash211, we can fix the new one. The errors appeared to be:

Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=512m; support was removed in 8.0
Scalastyle checks failed at following occurrences:
[error] /home/jenkins/workspace/PR-spark-k8s-unit-tests-SBT-TESTING/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesTestComponents.scala: File must end with newline character
[error] /home/jenkins/workspace/PR-spark-k8s-unit-tests-SBT-TESTING/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/ProcessUtils.scala:28:0: Use Javadoc style indentation for multiline comments
[error] /home/jenkins/workspace/PR-spark-k8s-unit-tests-SBT-TESTING/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/constants.scala: File must end with newline character
[error] (kubernetes-integration-tests/test:scalastyle) errors exist
[error] Total time: 13 s, completed May 15, 2017 9:02:13 PM
[error] running /home/jenkins/workspace/PR-spark-k8s-unit-tests-SBT-TESTING/dev/lint-scala ; received return code 1

Sending a PR to fix these.

[ash211]

@foxish I've got a commit that does that already in another PR: 6c84023

[foxish]

Ah! Okay, SG. Thanks!

[foxish]

Tests passed. Are there any more comments @mccheah, @ash211, @lins05? I'd like to get this merged before Wednesday because it blocks Varun's recovery behavior PR.

[mccheah]

Would it be possible to add unit-level tests around these? It would be great if we can start hardening the features we are implementing here. Unit testing these kinds of things can be difficult; we would probably have to refactor much of the scheduler backend and the shuffle pod cache to make us able to verify the things that are important.

[mccheah]

Add what the value the user specified was.

[mccheah]

Put allocatorRunnable, 0, TimeUnit.SECONDS all on this line.

[mccheah]

I think since now podAllocationInterval is always going to be provided (we always set it to Some(...) or throw an exception) then this thread will always be running. Is this the intended behavior? If so, no need to use foreach and options here.

[mccheah]

Would be cleaner I think to use if...else if... else here:

if (...) {
  logDebug("Maximum allowed executor limit...")
} else if (...) {
  logDebug("Waiting for pending...")
} else {
  // Actual logic
}

[foxish]

Done. Thanks! Trying to add a couple of unit tests to ShufflePodCache now and a mechanism that might help us add tests easily in the future.

[mccheah]

There's probably no need to make this an option - just assign podAllocationInterval directly, and then check the variable directly + throw the SparkException immediately afterwards.

[foxish]

Done

[mccheah]

This should be KeyValueUtils? This doesn't seem related to the command line.

[foxish]

I envisioned it as a place for utility functions related to commandline options which we could have more of, in future. The parsing of key-values is necessitated by commandline strings being supplied.

[foxish]

Is it confusing?

[mccheah]

This seems to mainly be used to parse out labels and annotations from SparkConf values - the command line doesn't seem to be related to that.

[foxish]

I see, I was assuming the primary way of supplying those args was via the cmdline. Okay, how about ConfigurationUtils? KeyValueUtils.parseKeyValuePairs() just seems a bit redundant.

[mccheah]

ConfigurationUtils is fine.

[foxish]

Unit tests seem more complex than expected, because watchers and such. https://mvnrepository.com/artifact/io.fabric8/kubernetes-server-mock provided an easy beginnning but I think I'll take it separately instead of blocking experiments using dynamic allocation.

[mccheah]

We can probably test the watches separately and just ensure that if the watch receives an event then the scheduler responds accordingly.

[foxish]

The mock server can be taught to expect the watch calls and respond appropriately. I used a similar thing in the unit tests here.

[foxish]

Created #275, will follow up there

[foxish]

Updated docs, any other comments?

[mccheah]

There's a few minor style things but they can be addressed either here or at some other point. Someone else can make a final pass before merging, but if there are no objections before the end of the day then feel free to proceed with the merge.

[mccheah]

We can use require here and in other similar places.

[mccheah]

Move up to previous line.

[mccheah]

These don't have to be vals and it's preferred that they aren't since these variables are now accessible from outside the scope of this class.

[foxish]

Addressed comments. Will merge after tests pass.

[foxish]

@ash211, do you want to merge #251 before this?

[ash211]

I think this dynamic allocation PR first, then afterwards the init containers one. That way the executor recovery PR can start making progress given that it's also blocked on this PR merging

[foxish]

Okay, SG. Merging this now, as tests passed.