dongjoon-hyun commented on code in PR #58619:
URL: https://github.com/apache/spark/pull/58619#discussion_r3985592305
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -908,6 +908,22 @@ package object config {
.stringConf
.createWithDefault("spark_shuffle")
+ private[spark] val SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS =
+ ConfigBuilder("spark.shuffle.service.requireAppScopedLocalDirs")
+ .doc("Whether the external shuffle service requires every local
directory an executor " +
+ "reports at registration to lie inside the registering application's
own " +
+ "per-application directory: the application id must appear as a path
segment of the " +
+ "directory's canonical path, under one of the service's configured
local directory " +
+ "roots. This keeps each application's shuffle and RDD blocks within
its own directory " +
+ "scope at registration and cleanup time. Only affects standalone mode.
Enable it only " +
+ "after every Worker in the cluster creates executor local directories
under a " +
+ "per-application directory (Workers on this version do); executors
launched by older " +
+ "Workers report unscoped paths and their registrations are rejected
while this is " +
+ "enabled.")
+ .version("4.4.0")
Review Comment:
New configs are required to declare a binding policy and the exception file
is frozen. This is the cause of `SparkConfigBindingPolicySuite` failure.
```suggestion
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
```
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -908,6 +908,22 @@ package object config {
.stringConf
.createWithDefault("spark_shuffle")
+ private[spark] val SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS =
+ ConfigBuilder("spark.shuffle.service.requireAppScopedLocalDirs")
+ .doc("Whether the external shuffle service requires every local
directory an executor " +
+ "reports at registration to lie inside the registering application's
own " +
+ "per-application directory: the application id must appear as a path
segment of the " +
+ "directory's canonical path, under one of the service's configured
local directory " +
+ "roots. This keeps each application's shuffle and RDD blocks within
its own directory " +
+ "scope at registration and cleanup time. Only affects standalone mode.
Enable it only " +
Review Comment:
In standalone mode, the shuffle service authenticates all applications with
a single shared secret (`SecurityManager.getSecretKey(appId)` ignores `appId`),
and `ExternalBlockHandler.checkAuth` only compares the client id with the
`appId` of the message. So, a client with the shared secret can still register
under another application's id, and there is no app id check at all when
`spark.authenticate=false`. In addition, `LocalDirValidator` accepts the app id
at any path segment, e.g., `<root>/spark-x/<otherAppId>/<appId>`.
This config helps to prevent mis-registration (e.g., bugs or races), but it
doesn't isolate applications from a malicious one. Could you revise this
sentence (and the same one in `docs/spark-standalone.md`) not to overstate it?
##########
core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala:
##########
@@ -622,7 +622,17 @@ private[deploy] class Worker(
val localRootDirs = Utils.getOrCreateLocalRootDirs(conf)
val dirs = localRootDirs.flatMap { dir =>
try {
- val appDir = Utils.createDirectory(dir, namePrefix =
"executor")
+ // Nest executor local dirs under a per-application directory
(the app id as
+ // a path segment) so the external shuffle service can require
registered
+ // localDirs to be scoped to the registering application.
+ val appIdDir = new File(dir, appId)
+ appIdDir.mkdirs()
+ if (!appIdDir.isDirectory) {
+ throw new IOException(s"Failed to create directory
$appIdDir")
+ }
+ Utils.chmod700(appIdDir)
+ val appDir = Utils.createDirectory(appIdDir.getAbsolutePath(),
+ namePrefix = "executor")
Utils.chmod700(appDir)
Some(appDir.getAbsolutePath())
Review Comment:
This additional `<appId>` level changes the local dir layout from
`spark-*/executor-*` to `spark-*/<appId>/executor-*`.
`KubernetesLocalDiskShuffleExecutorComponents.recoverDiskStore` walks a fixed
depth from the grandparent of the local dir (`volumeRootOf`), and
`KubernetesLocalDiskShuffleDataIOSuite` runs it with `local-cluster`, so the
recovery finds no shuffle files and 3 test cases fail.
Since `appDirectories` is per application (not per executor), `executor-*`
is already a per-application directory. How about using `<appId>` itself as the
application local dir instead of adding a new level? It keeps the existing
depth, satisfies `LocalDirValidator`, and makes the parent cleanup logic in
`maybeCleanupApplication` (and its new test) unnecessary.
```scala
val appDir = new File(dir, appId)
if (!Utils.createDirectory(appDir)) {
throw new IOException(s"Failed to create directory
$appDir")
}
Utils.chmod700(appDir)
Some(appDir.getAbsolutePath())
```
One caveat: in `local-cluster` mode, multiple Workers in the same JVM share
`Utils.getOrCreateLocalRootDirs`, so this directory would be shared by those
Workers. If that is a concern, the alternative is to make `recoverDiskStore`
handle the additional level.
If we keep the current layout, please use `Utils.createDirectory(appIdDir)`
instead of `mkdirs()` + `isDirectory` (SPARK-35907).
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
