dongjoon-hyun commented on code in PR #58619:
URL: https://github.com/apache/spark/pull/58619#discussion_r4109073686
##########
core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala:
##########
@@ -619,10 +619,28 @@ private[deploy] class Worker(
// SPARK_EXECUTOR_DIRS environment variable, and deleted by the
Worker when the
// application finishes.
val appLocalDirs = appDirectories.getOrElse(appId, {
- val localRootDirs = Utils.getOrCreateLocalRootDirs(conf)
+ val localRootDirs = Utils.getConfiguredLocalDirs(conf)
val dirs = localRootDirs.flatMap { dir =>
try {
- val appDir = Utils.createDirectory(dir, namePrefix =
"executor")
+ // Scope the executor's local dir to this Worker and the
application:
+ // <root>/spark-<workerId>/<appId>. The appId segment is what
+ // spark.shuffle.service.requireAppScopedLocalDirs checks at
registration.
+ // The per-worker directory keeps Workers that share local
root dirs from
+ // sharing an application directory -- in local-cluster mode
every Worker
+ // runs in one JVM, where Utils.getOrCreateLocalRootDirs
resolves to the
+ // same spark-* dirs -- so one Worker's application cleanup
cannot delete
+ // another Worker's live directory. The workerId is folded
into the
+ // spark-* name rather than added as a level of its own because
+ // Kubernetes shuffle recovery walks a fixed depth from the
executor's
+ // local dir. It is sanitized because it can contain
characters that are
+ // significant in paths, e.g. ':' from an IPv6 host, which is
also the
+ // SPARK_EXECUTOR_DIRS separator.
+ val workerDir = new File(dir,
s"spark-${Utils.sanitizeDirName(workerId)}")
+ val appDir = new File(workerDir, appId)
Review Comment:
`appId` is used as-is as a path segment and later joined into
`SPARK_EXECUTOR_DIRS` with `File.pathSeparator`. `spark.deploy.appIdPattern` is
user-configurable and only rejects whitespace. For example, with `app:%s:%04d`,
the executor splits `/tmp/spark-w/app:20260925000000:0001` into three entries
(two of them relative to the executor's working directory). The relative ones
fail the always-on root-containment check even when `requireAppScopedLocalDirs`
is false, and the shuffle data under `/tmp/spark-w/app` is not cleaned up by
the Worker. A `/` in the pattern splits the appId across segments, so
`pathContainsSegment` never matches when the flag is enabled. The comment above
sanitizes `workerId` for `:` for exactly this reason, but `appId` is not
covered.
##########
core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala:
##########
@@ -619,10 +619,28 @@ private[deploy] class Worker(
// SPARK_EXECUTOR_DIRS environment variable, and deleted by the
Worker when the
// application finishes.
val appLocalDirs = appDirectories.getOrElse(appId, {
- val localRootDirs = Utils.getOrCreateLocalRootDirs(conf)
+ val localRootDirs = Utils.getConfiguredLocalDirs(conf)
val dirs = localRootDirs.flatMap { dir =>
try {
- val appDir = Utils.createDirectory(dir, namePrefix =
"executor")
+ // Scope the executor's local dir to this Worker and the
application:
+ // <root>/spark-<workerId>/<appId>. The appId segment is what
+ // spark.shuffle.service.requireAppScopedLocalDirs checks at
registration.
+ // The per-worker directory keeps Workers that share local
root dirs from
+ // sharing an application directory -- in local-cluster mode
every Worker
+ // runs in one JVM, where Utils.getOrCreateLocalRootDirs
resolves to the
+ // same spark-* dirs -- so one Worker's application cleanup
cannot delete
+ // another Worker's live directory. The workerId is folded
into the
+ // spark-* name rather than added as a level of its own because
+ // Kubernetes shuffle recovery walks a fixed depth from the
executor's
+ // local dir. It is sanitized because it can contain
characters that are
+ // significant in paths, e.g. ':' from an IPv6 host, which is
also the
+ // SPARK_EXECUTOR_DIRS separator.
+ val workerDir = new File(dir,
s"spark-${Utils.sanitizeDirName(workerId)}")
Review Comment:
This replaces the random `spark-<uuid>/executor-<uuid>` names with a
predictable `<root>/spark-<workerId>/<appId>`. With the default
`spark.local.dir` (`java.io.tmpdir`, world-writable), another local user can
learn the workerId (e.g., from the Worker/Master UI) and pre-create
`/tmp/spark-<workerId>` with mode 777. `Utils.createDirectory`
(`Files.createDirectories`) silently accepts the existing directory, and the
result of `Utils.chmod700(workerDir)` is ignored, so the app dir ends up inside
an attacker-owned parent. The owner of that parent can rename the app dir and
substitute its own directory, and the ESS canonical-path check still passes
because the path is unchanged. The previous random names made this path
unguessable.
##########
core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala:
##########
@@ -619,10 +619,28 @@ private[deploy] class Worker(
// SPARK_EXECUTOR_DIRS environment variable, and deleted by the
Worker when the
// application finishes.
val appLocalDirs = appDirectories.getOrElse(appId, {
- val localRootDirs = Utils.getOrCreateLocalRootDirs(conf)
+ val localRootDirs = Utils.getConfiguredLocalDirs(conf)
Review Comment:
Previously, `Utils.getOrCreateLocalRootDirs` created `<root>/spark-<uuid>`
via `Utils.createTempDir`, which registers it with `ShutdownHookManager`, so a
graceful Worker stop removed all executor local dirs. With
`getConfiguredLocalDirs`, nothing registers `spark-<workerId>` for deletion,
and `onStop` calls `cleanupThreadExecutor.shutdownNow()`. If a Worker is
stopped while an app is running,
`<root>/spark-<oldWorkerId>/<appId>/blockmgr-*` (shuffle data) is never
reclaimed: the restarted Worker gets a new timestamped workerId, and
`WorkDirCleanup` only scans `workDir`. Even with clean app exits, an empty
`spark-<workerId>` directory accumulates on every Worker restart (and in every
local-cluster test run).
##########
core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala:
##########
@@ -83,18 +83,25 @@ class ExternalShuffleService(sparkConf: SparkConf,
securityManager: SecurityMana
/** Create a new shuffle block handler. Factored out for subclasses to
override. */
protected def newShuffleBlockHandler(conf: TransportConf):
ExternalBlockHandler = {
- // Constrain registered localDirs to the configured local directories.
+ // Constrain registered localDirs to the configured local directories and,
when
+ // spark.shuffle.service.requireAppScopedLocalDirs is enabled, to the
registering
+ // application's own per-app directory (the Worker creates executor local
dirs under a
+ // path containing the app id). The check is opt-in: executors launched by
Workers that
+ // predate the per-app layout register unscoped paths and would be
rejected, so enable it
+ // only after every Worker in the cluster is upgraded.
val localDirs = Utils.getConfiguredLocalDirs(sparkConf)
+ val requireAppScopedLocalDirs =
Review Comment:
Note that `LocalDirValidator` only checks that `appId` appears as some
segment anywhere in the canonical path, not that it is the per-app directory
directly under `<root>/spark-<workerId>`. For instance,
`<root>/spark-w/app-A/blockmgr-x/app-B` is accepted for `app-B`, which is
weaker than "inside the registering application's own per-application
directory" as documented. Since the standalone layout is fixed, would it be
worth checking the position of the segment (e.g.,
`<root>/spark-*/<appId>/...`), or at least toning down the documentation?
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -908,6 +908,28 @@ 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 guards against mis-registration (e.g. a bug or race
mixing up " +
+ "application directories); it is not an isolation boundary against
malicious " +
+ "applications, because the application id is supplied by the client
and is " +
+ "authenticated only when spark.authenticate is enabled, so enable both
together. " +
Review Comment:
In standalone mode, `spark.authenticate` does not bind a registration to its
appId: `SecurityManager.getSecretKey(appId)` ignores `appId`, and all
applications share a single cluster-wide secret. Any application can still
register with another application's appId and pass both `checkAuth` and this
check. Could we drop "so enable both together" (here and in
`spark-standalone.md`), or rephrase it so that it doesn't suggest enabling
authentication closes this gap?
##########
core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala:
##########
@@ -619,10 +619,28 @@ private[deploy] class Worker(
// SPARK_EXECUTOR_DIRS environment variable, and deleted by the
Worker when the
// application finishes.
val appLocalDirs = appDirectories.getOrElse(appId, {
- val localRootDirs = Utils.getOrCreateLocalRootDirs(conf)
+ val localRootDirs = Utils.getConfiguredLocalDirs(conf)
val dirs = localRootDirs.flatMap { dir =>
try {
- val appDir = Utils.createDirectory(dir, namePrefix =
"executor")
+ // Scope the executor's local dir to this Worker and the
application:
+ // <root>/spark-<workerId>/<appId>. The appId segment is what
+ // spark.shuffle.service.requireAppScopedLocalDirs checks at
registration.
+ // The per-worker directory keeps Workers that share local
root dirs from
+ // sharing an application directory -- in local-cluster mode
every Worker
+ // runs in one JVM, where Utils.getOrCreateLocalRootDirs
resolves to the
+ // same spark-* dirs -- so one Worker's application cleanup
cannot delete
+ // another Worker's live directory. The workerId is folded
into the
+ // spark-* name rather than added as a level of its own because
+ // Kubernetes shuffle recovery walks a fixed depth from the
executor's
+ // local dir. It is sanitized because it can contain
characters that are
+ // significant in paths, e.g. ':' from an IPv6 host, which is
also the
+ // SPARK_EXECUTOR_DIRS separator.
+ val workerDir = new File(dir,
s"spark-${Utils.sanitizeDirName(workerId)}")
+ val appDir = new File(workerDir, appId)
+ if (!Utils.createDirectory(appDir)) {
Review Comment:
The per-worker `spark-<workerId>` level, the workerId sanitization, and the
collision discussion in the comment are only needed because the directory names
became deterministic. A simpler alternative might be to keep the existing
`Utils.getOrCreateLocalRootDirs(conf)` + `Utils.createDirectory(dir, namePrefix
= "executor")` and create `appId` as a leaf under it, i.e.,
`<root>/spark-<uuid>/executor-<uuid>/<appId>`. That still satisfies the
appId-segment check while keeping unpredictable names and the shutdown-hook
cleanup, with no possibility of collisions between Workers.
##########
docs/spark-standalone.md:
##########
@@ -501,6 +501,29 @@ SPARK_WORKER_OPTS supports the following system properties:
</td>
<td>3.4.0</td>
</tr>
+<tr>
+ <td><code>spark.shuffle.service.requireAppScopedLocalDirs</code></td>
+ <td>false</td>
+ <td>
+ When true, the external shuffle service only accepts executor
registrations whose local
+ directories lie inside the registering application's own per-application
directory (the
+ application id must appear as a path segment under one of the service's
configured local
+ directory roots). This guards against mis-registration (e.g. a bug or race
mixing up
+ application directories); it is not an isolation boundary against
malicious applications,
+ because the application id is supplied by the client and is authenticated
only when
+ <code>spark.authenticate</code> is enabled, so enable both together.
Workers on this
+ version always create executor local directories as
+ <code><root>/spark-<workerId>/<appId></code>, which
satisfies the check; that layout
+ is additive and harmless while the check is disabled. For a rolling
upgrade, first upgrade
Review Comment:
This layout is not really additive: it unconditionally replaces the previous
`spark-<uuid>/executor-<uuid>` layout and changes the cleanup/permission
behavior regardless of the flag. Also, the PR description says
`<root>/<appId>/executor-*`, which doesn't match the implementation
(`<root>/spark-<workerId>/<appId>`).
##########
core/src/test/scala/org/apache/spark/deploy/worker/WorkerSuite.scala:
##########
@@ -443,4 +443,32 @@ class WorkerSuite extends SparkFunSuite with Matchers with
BeforeAndAfter with P
assert(getHeartbeatTask(worker) == heartbeatTask)
assert(getWorkDirCleanupTask(worker) == workDirCleanupTask)
}
+
+ test("app cleanup removes only the worker's own per-application executor
local dir") {
+ val externalShuffleServiceSupplier = new Supplier[ExternalShuffleService] {
+ override def get: ExternalShuffleService = shuffleService
+ }
+ val worker = makeWorker(new SparkConf(), externalShuffleServiceSupplier)
+ val root = Utils.createTempDir(namePrefix = "worker-local-root")
+ try {
+ // LaunchExecutor layout: the executor local dir is
<root>/spark-<workerId>/<appId>.
+ // Workers sharing a local root (local-cluster mode) each get their own
+ // spark-<workerId> directory, so application cleanup must remove only
this worker's
+ // per-application directory and leave another worker's untouched.
+ val appId = "app-20260922000000-0001"
+ val appDir = new File(new File(root, "spark-worker-1"), appId)
+ val otherWorkerAppDir = new File(new File(root, "spark-worker-2"), appId)
+ assert(appDir.mkdirs() && otherWorkerAppDir.mkdirs())
+ worker.appDirectories(appId) = Seq(appDir.getAbsolutePath)
Review Comment:
This test fills `appDirectories` by hand and only verifies that
`deleteRecursively` on one directory leaves a sibling untouched; it never goes
through `LaunchExecutor`. A regression in the `<root>/spark-<workerId>/<appId>`
construction (e.g., dropping the workerId segment) would still pass. Could it
drive `LaunchExecutor` (or a small extracted helper) so that it exercises the
production code?
##########
core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala:
##########
@@ -49,4 +54,103 @@ class ExternalShuffleServiceLocalDirsSuite extends
SparkFunSuite {
Utils.deleteRecursively(contained)
}
}
+
+ test("requireAppScopedLocalDirs restricts localDirs to the registering app's
directory") {
+ val sparkConf = new SparkConf()
+ .set(SHUFFLE_SERVICE_ENABLED, true)
+ .set(SHUFFLE_SERVICE_DB_ENABLED, false)
+ .set(SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS, true)
+ .set("spark.local.dir", System.getProperty("java.io.tmpdir"))
+ val service = new ExternalShuffleService(sparkConf, new
SecurityManager(sparkConf))
+ val handler = service.getBlockHandler
+
+ val root = new File(Utils.getConfiguredLocalDirs(sparkConf).head)
+ val appId = s"app-${UUID.randomUUID()}"
+ val otherAppId = s"app-${UUID.randomUUID()}"
+ // The layout the Worker creates: <root>/spark-<workerId>/<appId>.
+ val workerRoot = new File(root, s"spark-worker-${UUID.randomUUID()}")
+ val appScoped = new File(workerRoot, appId)
+ val otherAppScoped = new File(workerRoot, otherAppId)
+ // The pre-upgrade Worker layout: directly under a root, no appId segment.
+ val unscoped = new File(root, s"blockmgr-${UUID.randomUUID()}")
+ assert(appScoped.mkdirs() && otherAppScoped.mkdirs() && unscoped.mkdirs())
+ try {
+ // A localDir inside the registering application's own directory is
accepted.
+ handler.validateLocalDirs(Array(appScoped.getAbsolutePath), appId)
+
+ // A directory under a different application id is rejected even though
it is contained
+ // under a configured root.
+ intercept[IllegalArgumentException] {
+ handler.validateLocalDirs(Array(otherAppScoped.getAbsolutePath), appId)
+ }
+
+ // A contained directory without the appId as a path segment (what a
Worker without
+ // the per-app layout would report) is rejected.
+ intercept[IllegalArgumentException] {
+ handler.validateLocalDirs(Array(unscoped.getAbsolutePath), appId)
+ }
+
+ // Root containment is still enforced with app scoping on.
+ intercept[IllegalArgumentException] {
+ handler.validateLocalDirs(Array("/etc"), appId)
+ }
+ } finally {
+ Utils.deleteRecursively(workerRoot)
+ Utils.deleteRecursively(unscoped)
+ }
+ }
+
+ test("local-cluster executors register app-scoped local dirs end to end") {
+ val root = Utils.createTempDir(namePrefix = "ess-e2e-root")
+ val serviceConf = new SparkConf()
+ .set(SHUFFLE_SERVICE_ENABLED, true)
+ .set(SHUFFLE_SERVICE_DB_ENABLED, false)
+ .set(SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS, true)
+ .set(SHUFFLE_SERVICE_PORT, 0)
+ .set("spark.local.dir", root.getAbsolutePath)
+ val service = new ExternalShuffleService(serviceConf, new
SecurityManager(serviceConf))
+ service.start()
Review Comment:
`service.start()` and the reflective lookup are outside the `try`, so
`service.stop()` in `finally` is skipped if anything throws before the `try`,
leaking the bound server into later suites in the same JVM. Could we move them
inside the `try` (and delete `root` in `finally` as well)?
##########
core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala:
##########
@@ -49,4 +54,103 @@ class ExternalShuffleServiceLocalDirsSuite extends
SparkFunSuite {
Utils.deleteRecursively(contained)
}
}
+
+ test("requireAppScopedLocalDirs restricts localDirs to the registering app's
directory") {
+ val sparkConf = new SparkConf()
+ .set(SHUFFLE_SERVICE_ENABLED, true)
+ .set(SHUFFLE_SERVICE_DB_ENABLED, false)
+ .set(SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS, true)
+ .set("spark.local.dir", System.getProperty("java.io.tmpdir"))
+ val service = new ExternalShuffleService(sparkConf, new
SecurityManager(sparkConf))
+ val handler = service.getBlockHandler
+
+ val root = new File(Utils.getConfiguredLocalDirs(sparkConf).head)
+ val appId = s"app-${UUID.randomUUID()}"
+ val otherAppId = s"app-${UUID.randomUUID()}"
+ // The layout the Worker creates: <root>/spark-<workerId>/<appId>.
+ val workerRoot = new File(root, s"spark-worker-${UUID.randomUUID()}")
+ val appScoped = new File(workerRoot, appId)
+ val otherAppScoped = new File(workerRoot, otherAppId)
+ // The pre-upgrade Worker layout: directly under a root, no appId segment.
+ val unscoped = new File(root, s"blockmgr-${UUID.randomUUID()}")
+ assert(appScoped.mkdirs() && otherAppScoped.mkdirs() && unscoped.mkdirs())
+ try {
+ // A localDir inside the registering application's own directory is
accepted.
+ handler.validateLocalDirs(Array(appScoped.getAbsolutePath), appId)
+
+ // A directory under a different application id is rejected even though
it is contained
+ // under a configured root.
+ intercept[IllegalArgumentException] {
+ handler.validateLocalDirs(Array(otherAppScoped.getAbsolutePath), appId)
+ }
+
+ // A contained directory without the appId as a path segment (what a
Worker without
+ // the per-app layout would report) is rejected.
+ intercept[IllegalArgumentException] {
+ handler.validateLocalDirs(Array(unscoped.getAbsolutePath), appId)
+ }
+
+ // Root containment is still enforced with app scoping on.
+ intercept[IllegalArgumentException] {
+ handler.validateLocalDirs(Array("/etc"), appId)
+ }
+ } finally {
+ Utils.deleteRecursively(workerRoot)
+ Utils.deleteRecursively(unscoped)
+ }
+ }
+
+ test("local-cluster executors register app-scoped local dirs end to end") {
+ val root = Utils.createTempDir(namePrefix = "ess-e2e-root")
+ val serviceConf = new SparkConf()
+ .set(SHUFFLE_SERVICE_ENABLED, true)
+ .set(SHUFFLE_SERVICE_DB_ENABLED, false)
+ .set(SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS, true)
+ .set(SHUFFLE_SERVICE_PORT, 0)
+ .set("spark.local.dir", root.getAbsolutePath)
+ val service = new ExternalShuffleService(serviceConf, new
SecurityManager(serviceConf))
+ service.start()
+ // ExternalShuffleService does not expose the bound port; read it back the
same way
+ // ExternalShuffleServiceMetricsSuite reads the metrics source.
+ val serverField =
classOf[ExternalShuffleService].getDeclaredField("server")
Review Comment:
Instead of reflecting on the private `server` field,
`ExternalShuffleServiceSuite` shows a pattern without reflection: create `new
ExternalBlockHandler(transportConf, null, Array(root.getAbsolutePath), true)`
with `TransportContext.createServer()` and use `server.getPort` (the config
wiring is already covered by the previous test). Alternatively, a
`private[spark]` accessor for the bound port would work. Otherwise, renaming
the field would only surface as a runtime `NoSuchFieldException`.
##########
core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala:
##########
@@ -619,10 +619,28 @@ private[deploy] class Worker(
// SPARK_EXECUTOR_DIRS environment variable, and deleted by the
Worker when the
// application finishes.
val appLocalDirs = appDirectories.getOrElse(appId, {
- val localRootDirs = Utils.getOrCreateLocalRootDirs(conf)
+ val localRootDirs = Utils.getConfiguredLocalDirs(conf)
val dirs = localRootDirs.flatMap { dir =>
try {
- val appDir = Utils.createDirectory(dir, namePrefix =
"executor")
+ // Scope the executor's local dir to this Worker and the
application:
+ // <root>/spark-<workerId>/<appId>. The appId segment is what
+ // spark.shuffle.service.requireAppScopedLocalDirs checks at
registration.
+ // The per-worker directory keeps Workers that share local
root dirs from
+ // sharing an application directory -- in local-cluster mode
every Worker
+ // runs in one JVM, where Utils.getOrCreateLocalRootDirs
resolves to the
+ // same spark-* dirs -- so one Worker's application cleanup
cannot delete
+ // another Worker's live directory. The workerId is folded
into the
+ // spark-* name rather than added as a level of its own because
+ // Kubernetes shuffle recovery walks a fixed depth from the
executor's
Review Comment:
`KubernetesLocalDiskShuffleExecutorComponents.recoverDiskStore` only walks
the local dirs of K8s executors; it never sees directories created by a
standalone Worker, so this constraint doesn't seem to apply here. Also, the
comment refers to `Utils.getOrCreateLocalRootDirs`, which this code no longer
calls. Could we trim the comment to what actually applies?
--
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]
