This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new f2b689add956 Validate executor local directories and block ids in the
external shuffle service
f2b689add956 is described below
commit f2b689add956312f927844e47197c701e5e565c1
Author: Huaxin Gao <[email protected]>
AuthorDate: Wed Jul 1 14:51:00 2026 -0700
Validate executor local directories and block ids in the external shuffle
service
Add input validation to the external shuffle service for the
executor-supplied
values it turns into filesystem paths.
* Block ids passed to removeBlocks must be plain file names (no path
separator,
NUL, "." or ".."). A malformed id is skipped instead of being used to
build a
path, and a batched call continues with the remaining ids.
* The localDirs an executor reports at RegisterExecutor are validated once
at the
RPC boundary in ExternalBlockHandler, before they are handed to either
the block
resolver or the push-based-shuffle merge manager. The check lives in a
small
LocalDirValidator built from the host's configured local-directory roots:
each
entry must canonicalize to a path contained under one of those roots and,
when
app scoping is required, within the registering application's own
directory.
The roots are supplied per host through ExternalBlockHandler constructor
overloads:
YarnShuffleService passes the NodeManager's local directories
(yarn.nodemanager.local-dirs) with app scoping, and ExternalShuffleService
passes
the configured local directories (Utils.getConfiguredLocalDirs) without app
scoping. Block-id validation is covered by
ExternalShuffleBlockResolverSuite,
local-directory validation by LocalDirValidatorSuite, and the standalone
wiring by
ExternalShuffleServiceLocalDirsSuite; YarnShuffleServiceSuite fixtures are
updated
so registered localDirs sit under the NM local dirs and contain the appId.
Co-authored-by: Peter Toth <[email protected]>
---
.../network/shuffle/ExternalBlockHandler.java | 45 ++++++++
.../shuffle/ExternalShuffleBlockResolver.java | 27 +++++
.../spark/network/shuffle/LocalDirValidator.java | 115 +++++++++++++++++++++
.../shuffle/ExternalShuffleBlockResolverSuite.java | 102 ++++++++++++++++++
.../network/shuffle/LocalDirValidatorSuite.java | 94 +++++++++++++++++
.../spark/network/yarn/YarnShuffleService.java | 5 +-
.../spark/deploy/ExternalShuffleService.scala | 7 +-
.../ExternalShuffleServiceLocalDirsSuite.scala | 52 ++++++++++
.../network/yarn/YarnShuffleServiceSuite.scala | 79 +++++++++-----
9 files changed, 498 insertions(+), 28 deletions(-)
diff --git
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
index 45d0ff69de90..a0551254c9e8 100644
---
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
+++
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
@@ -78,6 +78,7 @@ public class ExternalBlockHandler extends RpcHandler
private final OneForOneStreamManager streamManager;
private final ShuffleMetrics metrics;
private final MergedShuffleFileManager mergeManager;
+ private final LocalDirValidator localDirValidator;
public ExternalBlockHandler(TransportConf conf, File registeredExecutorFile)
throws IOException {
@@ -86,6 +87,17 @@ public class ExternalBlockHandler extends RpcHandler
new NoOpMergedShuffleFileManager(conf, null));
}
+ public ExternalBlockHandler(
+ TransportConf conf,
+ File registeredExecutorFile,
+ String[] allowedLocalDirs,
+ boolean requireAppScopedLocalDirs) throws IOException {
+ this(new OneForOneStreamManager(),
+ new ExternalShuffleBlockResolver(conf, registeredExecutorFile),
+ new NoOpMergedShuffleFileManager(conf, null),
+ new LocalDirValidator(allowedLocalDirs, requireAppScopedLocalDirs));
+ }
+
public ExternalBlockHandler(
TransportConf conf,
File registeredExecutorFile,
@@ -94,11 +106,32 @@ public class ExternalBlockHandler extends RpcHandler
new ExternalShuffleBlockResolver(conf, registeredExecutorFile),
mergeManager);
}
+ public ExternalBlockHandler(
+ TransportConf conf,
+ File registeredExecutorFile,
+ MergedShuffleFileManager mergeManager,
+ String[] allowedLocalDirs,
+ boolean requireAppScopedLocalDirs) throws IOException {
+ this(new OneForOneStreamManager(),
+ new ExternalShuffleBlockResolver(conf, registeredExecutorFile),
+ mergeManager,
+ new LocalDirValidator(allowedLocalDirs, requireAppScopedLocalDirs));
+ }
+
@VisibleForTesting
public ExternalShuffleBlockResolver getBlockResolver() {
return blockManager;
}
+ /**
+ * Validates the local directories an executor reports against the service's
configured roots.
+ * Applied at the RPC boundary so both the block resolver and the
merged-shuffle manager only
+ * consume validated paths. Throws {@link IllegalArgumentException} if an
entry is not allowed.
+ */
+ public void validateLocalDirs(String[] localDirs, String appId) {
+ localDirValidator.validate(localDirs, appId);
+ }
+
/** Enables mocking out the StreamManager and BlockManager. */
@VisibleForTesting
public ExternalBlockHandler(
@@ -113,10 +146,19 @@ public class ExternalBlockHandler extends RpcHandler
OneForOneStreamManager streamManager,
ExternalShuffleBlockResolver blockManager,
MergedShuffleFileManager mergeManager) {
+ this(streamManager, blockManager, mergeManager, new
LocalDirValidator(null, false));
+ }
+
+ private ExternalBlockHandler(
+ OneForOneStreamManager streamManager,
+ ExternalShuffleBlockResolver blockManager,
+ MergedShuffleFileManager mergeManager,
+ LocalDirValidator localDirValidator) {
this.metrics = new ShuffleMetrics();
this.streamManager = streamManager;
this.blockManager = blockManager;
this.mergeManager = mergeManager;
+ this.localDirValidator = localDirValidator;
}
@Override
@@ -185,6 +227,9 @@ public class ExternalBlockHandler extends RpcHandler
metrics.registerExecutorRequestLatencyMillis.time();
try {
checkAuth(client, msg.appId);
+ // Validate executor-supplied localDirs once here, at the RPC
boundary, so both the block
+ // resolver and the merged-shuffle manager only ever consume validated
paths.
+ validateLocalDirs(msg.executorInfo.localDirs, msg.appId);
blockManager.registerExecutor(msg.appId, msg.execId, msg.executorInfo);
mergeManager.registerExecutor(msg.appId, msg.executorInfo);
callback.onSuccess(ByteBuffer.wrap(new byte[0]));
diff --git
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
index b3002833fce1..f79000fecd2f 100644
---
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
+++
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
@@ -376,6 +376,13 @@ public class ExternalShuffleBlockResolver {
}
int numRemovedBlocks = 0;
for (String blockId : blockIds) {
+ try {
+ validateBlockId(blockId);
+ } catch (IllegalArgumentException e) {
+ logger.warn("Skipping block with invalid id: {}",
+ MDC.of(LogKeys.BLOCK_ID, blockId));
+ continue;
+ }
File file = new File(
ExecutorDiskUtils.getFilePath(executor.localDirs,
executor.subDirsPerLocalDir, blockId));
if (file.delete()) {
@@ -422,6 +429,26 @@ public class ExternalShuffleBlockResolver {
algorithm, checksumFile, reduceId, data, checksumByReader);
}
+ /**
+ * Validates that a blockId is a plain file name before it is used to build
a file path under
+ * {@code <localDir>/<subDirHex>/}. Legitimate Spark block ids contain only
ASCII alphanumerics,
+ * underscores, dots and dashes (see {@code BlockId} in core), so a
well-formed blockId never
+ * contains a path separator and is never {@code "."} or {@code ".."}. Such
values are rejected.
+ */
+ @VisibleForTesting
+ static void validateBlockId(String blockId) {
+ if (blockId == null || blockId.isEmpty()) {
+ throw new IllegalArgumentException("blockId must not be null or empty");
+ }
+ if (blockId.indexOf('/') >= 0 || blockId.indexOf('\\') >= 0 ||
blockId.indexOf('\0') >= 0) {
+ throw new IllegalArgumentException(
+ "blockId must not contain a path separator or NUL: " + blockId);
+ }
+ if (blockId.equals("..") || blockId.equals(".")) {
+ throw new IllegalArgumentException("blockId must not be \".\" or \"..\":
" + blockId);
+ }
+ }
+
/** Simply encodes an executor's full ID, which is appId + execId. */
public static class AppExecId {
public final String appId;
diff --git
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/LocalDirValidator.java
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/LocalDirValidator.java
new file mode 100644
index 000000000000..b7e08386a222
--- /dev/null
+++
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/LocalDirValidator.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.spark.internal.SparkLogger;
+import org.apache.spark.internal.SparkLoggerFactory;
+import org.apache.spark.internal.LogKeys;
+import org.apache.spark.internal.MDC;
+
+/**
+ * Validates the local directories an executor reports at registration, before
the external shuffle
+ * service turns them into filesystem paths. {@link ExternalBlockHandler} owns
one of these (built
+ * from the host's configured local-directory roots) and applies it at the RPC
boundary, so every
+ * consumer of those directories -- the block resolver and the merged-shuffle
manager -- only ever
+ * operates on validated paths.
+ *
+ * Each entry must canonicalize to a path contained under one of the
configured roots (YARN passes
+ * {@code yarn.nodemanager.local-dirs}; standalone passes the configured local
directories) and,
+ * when app scoping is required, within the registering application's own
directory (the
+ * {@code appId} must appear as a path segment). When no roots are configured
there is nothing to
+ * contain the entries against and they are accepted. An empty array is always
allowed.
+ */
+class LocalDirValidator {
+ private static final SparkLogger logger =
SparkLoggerFactory.getLogger(LocalDirValidator.class);
+
+ private final List<Path> allowedLocalDirRoots;
+ private final boolean requireAppScopedLocalDirs;
+
+ LocalDirValidator(String[] allowedLocalDirs, boolean
requireAppScopedLocalDirs) {
+ this.allowedLocalDirRoots = canonicalizeRoots(allowedLocalDirs);
+ this.requireAppScopedLocalDirs = requireAppScopedLocalDirs;
+ }
+
+ void validate(String[] localDirs, String appId) {
+ if (allowedLocalDirRoots.isEmpty()) {
+ return;
+ }
+ if (localDirs == null) {
+ throw new IllegalArgumentException("localDirs must not be null");
+ }
+ for (String localDir : localDirs) {
+ if (localDir == null || localDir.isEmpty()) {
+ throw new IllegalArgumentException("localDirs entries must be non-null
and non-empty");
+ }
+ validateContained(localDir, appId);
+ }
+ }
+
+ private void validateContained(String localDir, String appId) {
+ Path canonical;
+ try {
+ canonical = new File(localDir).getCanonicalFile().toPath();
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "localDirs entry could not be canonicalized: " + localDir, e);
+ }
+ if (allowedLocalDirRoots.stream().noneMatch(canonical::startsWith)) {
+ throw new IllegalArgumentException(
+ "localDirs entry is not under any of the service's local directories:
" + localDir);
+ }
+ if (requireAppScopedLocalDirs &&
+ (appId == null || appId.isEmpty() || !pathContainsSegment(canonical,
appId))) {
+ throw new IllegalArgumentException(
+ "localDirs entry is not within application " + appId + "'s directory:
" + localDir);
+ }
+ }
+
+ private static boolean pathContainsSegment(Path path, String segment) {
+ for (Path part : path) {
+ if (part.toString().equals(segment)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static List<Path> canonicalizeRoots(String[] allowedLocalDirs) {
+ List<Path> roots = new ArrayList<>();
+ if (allowedLocalDirs != null) {
+ for (String root : allowedLocalDirs) {
+ if (root == null || root.isEmpty()) {
+ continue;
+ }
+ try {
+ roots.add(new File(root).getCanonicalFile().toPath());
+ } catch (IOException e) {
+ logger.warn("Ignoring local dir root that could not be
canonicalized: {}",
+ MDC.of(LogKeys.PATH, root));
+ }
+ }
+ }
+ return roots;
+ }
+}
diff --git
a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolverSuite.java
b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolverSuite.java
index 488d02d63d55..fbc2e447a2bf 100644
---
a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolverSuite.java
+++
b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolverSuite.java
@@ -17,9 +17,12 @@
package org.apache.spark.network.shuffle;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.spark.network.shuffle.protocol.ExecutorShuffleInfo;
@@ -120,4 +123,103 @@ public class ExternalShuffleBlockResolverSuite {
assertEquals(shuffleInfo, mapper.readValue(legacyShuffleJson,
ExecutorShuffleInfo.class));
}
+ /**
+ * removeBlocks only operates on plain block-id file names. A block id
containing parent-directory
+ * segments is skipped: it is not removed, and a file located outside the
executor's local
+ * directory is left untouched.
+ */
+ @Test
+ public void removeBlocksIgnoresBlockIdWithParentDirSegments() throws
IOException {
+ TestShuffleDataContext context = new TestShuffleDataContext(1, 5);
+ context.create();
+ File outsideFile = null;
+ try {
+ File localDir = new File(context.localDirs[0]);
+ File parentOfLocalDir = localDir.getParentFile();
+ outsideFile = new File(parentOfLocalDir, "outside-" + UUID.randomUUID()
+ ".txt");
+ Files.writeString(outsideFile.toPath(), "should not be deleted",
StandardCharsets.UTF_8);
+ assertTrue(outsideFile.exists(), "precondition: file was created");
+
+ ExternalShuffleBlockResolver resolver = new
ExternalShuffleBlockResolver(conf, null);
+ resolver.registerExecutor("app0", "0",
context.createExecutorInfo(SORT_MANAGER));
+
+ String blockId = ".." + File.separator + ".." + File.separator +
outsideFile.getName();
+ int removed = resolver.removeBlocks("app0", "0", new String[] { blockId
});
+
+ assertEquals(0, removed, "A block id with parent-dir segments must not
remove a block");
+ assertTrue(outsideFile.exists(),
+ "A file outside the local directory must not be removed (" +
+ outsideFile.getAbsolutePath() + ")");
+ } finally {
+ if (outsideFile != null && outsideFile.exists()) {
+ assertTrue(outsideFile.delete() || !outsideFile.exists());
+ }
+ context.cleanup();
+ }
+ }
+
+ /**
+ * A block id containing a path separator is not a plain file name and is
skipped, so removeBlocks
+ * reports zero removed blocks.
+ */
+ @Test
+ public void removeBlocksIgnoresBlockIdWithSeparator() throws IOException {
+ TestShuffleDataContext context = new TestShuffleDataContext(1, 5);
+ context.create();
+ try {
+ ExternalShuffleBlockResolver resolver = new
ExternalShuffleBlockResolver(conf, null);
+ resolver.registerExecutor("app0", "0",
context.createExecutorInfo(SORT_MANAGER));
+
+ String blockId = "sub" + File.separator + "child";
+ int removed = resolver.removeBlocks("app0", "0", new String[] { blockId
});
+
+ assertEquals(0, removed, "A block id containing a path separator must be
skipped");
+ } finally {
+ context.cleanup();
+ }
+ }
+
+ /**
+ * In a batched call, an invalid block id does not abort the whole
operation: a valid block id in
+ * the same batch is still removed while the invalid one is skipped.
+ */
+ @Test
+ public void removeBlocksContinuesAfterIgnoringInvalidBlockId() throws
IOException {
+ TestShuffleDataContext context = new TestShuffleDataContext(1, 5);
+ context.create();
+ File outsideFile = null;
+ try {
+ File localDir = new File(context.localDirs[0]);
+ File parentOfLocalDir = localDir.getParentFile();
+ outsideFile = new File(parentOfLocalDir, "outside-" + UUID.randomUUID()
+ ".txt");
+ Files.writeString(outsideFile.toPath(), "should not be deleted",
StandardCharsets.UTF_8);
+
+ // Plant a real, legitimately-named block in the localDir so the valid
removal has something
+ // to actually remove.
+ context.insertCachedRddData(7, 0, new byte[] { 1, 2, 3 });
+ String validBlockId = "rdd_7_0";
+ File validFile = new File(ExecutorDiskUtils.getFilePath(
+ context.localDirs, context.subDirsPerLocalDir, validBlockId));
+ assertTrue(validFile.exists(), "precondition: block file was created");
+
+ ExternalShuffleBlockResolver resolver = new
ExternalShuffleBlockResolver(conf, null);
+ resolver.registerExecutor("app0", "0",
context.createExecutorInfo(SORT_MANAGER));
+
+ String invalidBlockId =
+ ".." + File.separator + ".." + File.separator + outsideFile.getName();
+ int removed = resolver.removeBlocks(
+ "app0", "0", new String[] { invalidBlockId, validBlockId });
+
+ assertEquals(1, removed, "The valid block id in the batch should still
be removed");
+ assertTrue(outsideFile.exists(),
+ "The invalid entry must not remove a file outside the local
directory");
+ assertFalse(validFile.exists(), "The valid block file should have been
removed by the batch");
+ } finally {
+ if (outsideFile != null && outsideFile.exists()) {
+ assertTrue(outsideFile.delete() || !outsideFile.exists());
+ }
+ context.cleanup();
+ }
+ }
+
}
diff --git
a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/LocalDirValidatorSuite.java
b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/LocalDirValidatorSuite.java
new file mode 100644
index 000000000000..cf678bb0d633
--- /dev/null
+++
b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/LocalDirValidatorSuite.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.io.File;
+import java.nio.file.Files;
+
+import org.apache.spark.network.util.JavaUtils;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class LocalDirValidatorSuite {
+
+ /**
+ * With local-directory roots configured, validate() must reject any
localDir that does not
+ * canonicalize to a path under one of those roots.
+ */
+ private static void assertRejectsLocalDir(String[] localDirs) throws
Exception {
+ File root = Files.createTempDirectory("ess-local-root").toFile();
+ try {
+ LocalDirValidator validator = new LocalDirValidator(new String[] {
root.getPath() }, false);
+ assertThrows(IllegalArgumentException.class, () ->
validator.validate(localDirs, "app0"));
+ } finally {
+ JavaUtils.deleteRecursively(root);
+ }
+ }
+
+ @Test
+ public void rejectsRootLocalDir() throws Exception {
+ assertRejectsLocalDir(new String[] { File.separator });
+ }
+
+ @Test
+ public void rejectsParentDirSegment() throws Exception {
+ assertRejectsLocalDir(new String[] {
+ "/tmp" + File.separator + "x" + File.separator + ".." + File.separator +
".." +
+ File.separator + "etc" });
+ }
+
+ @Test
+ public void rejectsNullEntry() throws Exception {
+ assertRejectsLocalDir(new String[] { null });
+ }
+
+ @Test
+ public void rejectsDirOutsideConfiguredRoots() throws Exception {
+ assertRejectsLocalDir(new String[] { "/etc/yarn" });
+ }
+
+ @Test
+ public void rejectsAnotherApplicationsDir() throws Exception {
+ File root = Files.createTempDirectory("ess-nm-local").toFile();
+ try {
+ File otherAppDir = new File(root,
"usercache/u/appcache/app1/blockmgr-x");
+ assertTrue(otherAppDir.mkdirs(), "precondition: directory created");
+ LocalDirValidator validator = new LocalDirValidator(new String[] {
root.getPath() }, true);
+ assertThrows(IllegalArgumentException.class,
+ () -> validator.validate(new String[] { otherAppDir.getPath() },
"app0"),
+ "validate must reject a localDir scoped to a different application");
+ } finally {
+ JavaUtils.deleteRecursively(root);
+ }
+ }
+
+ @Test
+ public void acceptsContainedAppScopedDir() throws Exception {
+ File root = Files.createTempDirectory("ess-nm-local").toFile();
+ try {
+ File ownDir = new File(root, "usercache/u/appcache/app0/blockmgr-x");
+ assertTrue(ownDir.mkdirs(), "precondition: directory created");
+ LocalDirValidator validator = new LocalDirValidator(new String[] {
root.getPath() }, true);
+ // Must not throw.
+ validator.validate(new String[] { ownDir.getPath() }, "app0");
+ } finally {
+ JavaUtils.deleteRecursively(root);
+ }
+ }
+}
diff --git
a/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java
b/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java
index ab23585170b4..f04d1aa2f023 100644
---
a/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java
+++
b/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java
@@ -285,8 +285,11 @@ public class YarnShuffleService extends AuxiliaryService {
if (shuffleMergeManager == null) {
shuffleMergeManager =
newMergedShuffleFileManagerInstance(transportConf, mergeManagerFile);
}
+ // Constrain registered localDirs to the NodeManager's configured local
directories
+ // (yarn.nodemanager.local-dirs) and to the registering application's
own directory.
+ String[] nmLocalDirs =
_conf.getTrimmedStrings("yarn.nodemanager.local-dirs");
blockHandler = new ExternalBlockHandler(
- transportConf, registeredExecutorFile, shuffleMergeManager);
+ transportConf, registeredExecutorFile, shuffleMergeManager,
nmLocalDirs, true);
// If authentication is enabled, set up the shuffle server to use a
// special RPC handler that filters out unauthenticated fetch requests
diff --git
a/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala
b/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala
index fec69aed4898..1f709ee758c0 100644
--- a/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala
@@ -83,15 +83,18 @@ 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.
+ val localDirs = Utils.getConfiguredLocalDirs(sparkConf)
if (sparkConf.get(config.SHUFFLE_SERVICE_DB_ENABLED) && enabled) {
val shuffleDBName = sparkConf.get(config.SHUFFLE_SERVICE_DB_BACKEND)
logInfo(
log"Use ${MDC(SHUFFLE_DB_BACKEND_NAME, shuffleDBName.name())} as the
implementation of " +
log"${MDC(SHUFFLE_DB_BACKEND_KEY,
config.SHUFFLE_SERVICE_DB_BACKEND.key)}")
new ExternalBlockHandler(conf,
-
findRegisteredExecutorsDBFile(shuffleDBName.fileName(registeredExecutorsDB)))
+
findRegisteredExecutorsDBFile(shuffleDBName.fileName(registeredExecutorsDB)),
+ localDirs, false)
} else {
- new ExternalBlockHandler(conf, null)
+ new ExternalBlockHandler(conf, null, localDirs, false)
}
}
diff --git
a/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala
b/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala
new file mode 100644
index 000000000000..29576b5fd5ae
--- /dev/null
+++
b/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy
+
+import java.io.File
+import java.util.UUID
+
+import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite}
+import org.apache.spark.internal.config.{SHUFFLE_SERVICE_DB_ENABLED,
SHUFFLE_SERVICE_ENABLED}
+import org.apache.spark.util.Utils
+
+class ExternalShuffleServiceLocalDirsSuite extends SparkFunSuite {
+
+ test("validateLocalDirs only accepts localDirs under the configured local
directories") {
+ val sparkConf = new SparkConf()
+ .set(SHUFFLE_SERVICE_ENABLED, true)
+ .set(SHUFFLE_SERVICE_DB_ENABLED, false)
+ .set("spark.local.dir", System.getProperty("java.io.tmpdir"))
+ val service = new ExternalShuffleService(sparkConf, new
SecurityManager(sparkConf))
+ val handler = service.getBlockHandler
+
+ // A localDir under one of the configured local directories is accepted.
+ val root = new File(Utils.getConfiguredLocalDirs(sparkConf).head)
+ val contained = new File(root, s"blockmgr-${UUID.randomUUID()}")
+ assert(contained.mkdirs())
+ try {
+ handler.validateLocalDirs(Array(contained.getAbsolutePath),
"app-contained")
+
+ // A well-formed absolute localDir outside every configured root is
rejected.
+ intercept[IllegalArgumentException] {
+ handler.validateLocalDirs(Array("/etc"), "app-outside")
+ }
+ } finally {
+ Utils.deleteRecursively(contained)
+ }
+ }
+}
diff --git
a/resource-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala
b/resource-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala
index 72d684e15fc6..72e35c5684d2 100644
---
a/resource-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala
+++
b/resource-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala
@@ -76,9 +76,21 @@ abstract class YarnShuffleServiceSuite extends SparkFunSuite
with Matchers {
private var recoveryLocalDir: File = _
protected var tempDir: File = _
+ // The NM local dir set as yarn.nodemanager.local-dirs.
ExternalShuffleBlockResolver only
+ // accepts executor localDirs that canonicalize to a path contained under
this root and that
+ // contain the registering appId as a path segment (see validateLocalDirs).
Tests build such
+ // paths via appLocalDir below.
+ protected var nmLocalDir: File = _
protected def shuffleDBBackend(): DBBackend
+ // Build an executor/merge local dir under the NM local dir (nmLocalDir)
scoped to the given
+ // application, mirroring the real YARN layout
<nm-local-dir>/usercache/<user>/appcache/<appId>.
+ // This satisfies the local-dir validation done at the RPC boundary
(LocalDirValidator): the path
+ // is contained under the configured NM local dir and contains the appId as
a path segment.
+ protected def appLocalDir(appId: String, suffix: String): String =
+ new File(nmLocalDir,
s"usercache/u/appcache/$appId/$suffix").getAbsolutePath
+
override def beforeEach(): Unit = {
super.beforeEach()
// Ensure that each test uses a fresh metrics system
@@ -91,6 +103,7 @@ abstract class YarnShuffleServiceSuite extends SparkFunSuite
with Matchers {
yarnConfig.setInt(SHUFFLE_SERVICE_PORT.key, 0)
yarnConfig.setBoolean(YarnShuffleService.STOP_ON_FAILURE_KEY, true)
val localDir = Utils.createTempDir()
+ nmLocalDir = localDir
yarnConfig.set(YarnConfiguration.NM_LOCAL_DIRS, localDir.getAbsolutePath)
yarnConfig.set("spark.shuffle.push.server.mergedShuffleFileManagerImpl",
"org.apache.spark.network.shuffle.RemoteBlockPushResolver")
@@ -203,15 +216,19 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
secretsFile should not be (null)
val mergeMgrFile = s1.mergeManagerFile
mergeMgrFile should not be (null)
- val shuffleInfo1 = new ExecutorShuffleInfo(Array("/foo", "/bar"), 3,
SORT_MANAGER)
- val shuffleInfo2 = new ExecutorShuffleInfo(Array("/bippy"), 5,
SORT_MANAGER)
+ val shuffleInfo1 =
+ new ExecutorShuffleInfo(
+ Array(appLocalDir(app1Id.toString, "foo/foo"),
+ appLocalDir(app1Id.toString, "bar/bar")), 3, SORT_MANAGER)
+ val shuffleInfo2 =
+ new ExecutorShuffleInfo(Array(appLocalDir(app2Id.toString,
"bippy/bippy")), 5, SORT_MANAGER)
val mergedShuffleInfo3 =
new ExecutorShuffleInfo(
- Array(new File(tempDir, "foo/foo").getAbsolutePath,
- new File(tempDir, "bar/bar").getAbsolutePath), 3,
+ Array(appLocalDir(app3Id.toString, "foo/foo"),
+ appLocalDir(app3Id.toString, "bar/bar")), 3,
SORT_MANAGER_WITH_MERGE_SHUFFLE_META_WithAttemptID1)
val mergedShuffleInfo4 =
- new ExecutorShuffleInfo(Array(new File(tempDir,
"bippy/bippy").getAbsolutePath),
+ new ExecutorShuffleInfo(Array(appLocalDir(app4Id.toString,
"bippy/bippy")),
5, SORT_MANAGER_WITH_MERGE_SHUFFLE_META_WithAttemptID1)
val blockHandler = s1.blockHandler
@@ -237,9 +254,9 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
mergeManager.registerExecutor(app3Id.toString, mergedShuffleInfo3)
mergeManager.registerExecutor(app4Id.toString, mergedShuffleInfo4)
- val localDirs3 = Array(new File(tempDir,
"foo/merge_manager_1").getAbsolutePath,
- new File(tempDir, "bar/merge_manager_1").getAbsolutePath)
- val localDirs4 = Array(new File(tempDir,
"bippy/merge_manager_1").getAbsolutePath)
+ val localDirs3 = Array(appLocalDir(app3Id.toString, "foo/merge_manager_1"),
+ appLocalDir(app3Id.toString, "bar/merge_manager_1"))
+ val localDirs4 = Array(appLocalDir(app4Id.toString,
"bippy/merge_manager_1"))
val appPathsInfo3 = new AppPathsInfo(localDirs3, 3)
val appPathsInfo4 = new AppPathsInfo(localDirs4, 5)
@@ -354,15 +371,19 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
val execStateFile = s1.registeredExecutorFile
execStateFile should not be (null)
- val shuffleInfo1 = new ExecutorShuffleInfo(Array("/foo", "/bar"), 3,
SORT_MANAGER)
- val shuffleInfo2 = new ExecutorShuffleInfo(Array("/bippy"), 5,
SORT_MANAGER)
+ val shuffleInfo1 =
+ new ExecutorShuffleInfo(
+ Array(appLocalDir(app1Id.toString, "foo/foo"),
+ appLocalDir(app1Id.toString, "bar/bar")), 3, SORT_MANAGER)
+ val shuffleInfo2 =
+ new ExecutorShuffleInfo(Array(appLocalDir(app2Id.toString,
"bippy/bippy")), 5, SORT_MANAGER)
val mergedShuffleInfo3 =
new ExecutorShuffleInfo(
- Array(new File(tempDir, "foo/foo").getAbsolutePath,
- new File(tempDir, "bar/bar").getAbsolutePath),
+ Array(appLocalDir(app3Id.toString, "foo/foo"),
+ appLocalDir(app3Id.toString, "bar/bar")),
3, SORT_MANAGER_WITH_MERGE_SHUFFLE_META_WithAttemptID1)
val mergedShuffleInfo4 =
- new ExecutorShuffleInfo(Array(new File(tempDir,
"bippy/bippy").getAbsolutePath),
+ new ExecutorShuffleInfo(Array(appLocalDir(app4Id.toString,
"bippy/bippy")),
5, SORT_MANAGER_WITH_MERGE_SHUFFLE_META_WithAttemptID1)
val blockHandler = s1.blockHandler
@@ -407,7 +428,10 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
s1.initializeApplication(app1Data)
val execStateFile = s1.registeredExecutorFile
- val shuffleInfo1 = new ExecutorShuffleInfo(Array("/foo", "/bar"), 3,
SORT_MANAGER)
+ val shuffleInfo1 =
+ new ExecutorShuffleInfo(
+ Array(appLocalDir(app1Id.toString, "foo/foo"),
+ appLocalDir(app1Id.toString, "bar/bar")), 3, SORT_MANAGER)
val blockHandler = s1.blockHandler
val blockResolver = ShuffleTestAccessor.getBlockResolver(blockHandler)
@@ -437,7 +461,8 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
val app2Id = ApplicationId.newInstance(0, 2)
val app2Data = makeAppInfo("user", app2Id)
s2.initializeApplication(app2Data)
- val shuffleInfo2 = new ExecutorShuffleInfo(Array("/bippy"), 5,
SORT_MANAGER)
+ val shuffleInfo2 =
+ new ExecutorShuffleInfo(Array(appLocalDir(app2Id.toString,
"bippy/bippy")), 5, SORT_MANAGER)
resolver2.registerExecutor(app2Id.toString, "exec-2", shuffleInfo2)
ShuffleTestAccessor.getExecutorInfo(app2Id, "exec-2", resolver2) should be
(Some(shuffleInfo2))
s2.stop()
@@ -490,8 +515,12 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
execStateFile should not be (null)
val secretsFile = s1.secretsFile
secretsFile should not be (null)
- val shuffleInfo1 = new ExecutorShuffleInfo(Array("/foo", "/bar"), 3,
SORT_MANAGER)
- val shuffleInfo2 = new ExecutorShuffleInfo(Array("/bippy"), 5,
SORT_MANAGER)
+ val shuffleInfo1 =
+ new ExecutorShuffleInfo(
+ Array(appLocalDir(app1Id.toString, "foo/foo"),
+ appLocalDir(app1Id.toString, "bar/bar")), 3, SORT_MANAGER)
+ val shuffleInfo2 =
+ new ExecutorShuffleInfo(Array(appLocalDir(app2Id.toString,
"bippy/bippy")), 5, SORT_MANAGER)
val blockHandler = s1.blockHandler
val blockResolver = ShuffleTestAccessor.getBlockResolver(blockHandler)
@@ -1148,11 +1177,11 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
val execShuffleInfo1 =
new ExecutorShuffleInfo(
- Array(new File(tempDir, "foo/foo").getAbsolutePath,
- new File(tempDir, "bar/bar").getAbsolutePath), 3,
+ Array(appLocalDir(app1Id.toString, "foo/foo"),
+ appLocalDir(app1Id.toString, "bar/bar")), 3,
SORT_MANAGER_WITH_MERGE_SHUFFLE_META_WithAttemptID1)
val execShuffleInfo2 =
- new ExecutorShuffleInfo(Array(new File(tempDir,
"bippy/bippy").getAbsolutePath),
+ new ExecutorShuffleInfo(Array(appLocalDir(app2Id.toString,
"bippy/bippy")),
3, SORT_MANAGER_WITH_MERGE_SHUFFLE_META_WithAttemptID1)
val blockHandler = s1.blockHandler
@@ -1167,9 +1196,9 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
val mergeManager =
s1.shuffleMergeManager.asInstanceOf[RemoteBlockPushResolver]
mergeManager.registerExecutor(app1Id.toString, execShuffleInfo1)
mergeManager.registerExecutor(app2Id.toString, execShuffleInfo2)
- val localDirsApp1 = Array(new File(tempDir,
"foo/merge_manager_1").getAbsolutePath,
- new File(tempDir, "bar/merge_manager_1").getAbsolutePath)
- val localDirsApp2 = Array(new File(tempDir,
"bippy/merge_manager_1").getAbsolutePath)
+ val localDirsApp1 = Array(appLocalDir(app1Id.toString,
"foo/merge_manager_1"),
+ appLocalDir(app1Id.toString, "bar/merge_manager_1"))
+ val localDirsApp2 = Array(appLocalDir(app2Id.toString,
"bippy/merge_manager_1"))
val appPathsInfo1 = new AppPathsInfo(localDirsApp1, 3)
val appPathsInfo2 = new AppPathsInfo(localDirsApp2, 3)
@@ -1238,8 +1267,8 @@ abstract class YarnShuffleServiceSuite extends
SparkFunSuite with Matchers {
s1.initializeApplication(app1Data)
val execShuffleInfo1 =
new ExecutorShuffleInfo(
- Array(new File(tempDir, "foo/foo").getAbsolutePath,
- new File(tempDir, "bar/bar").getAbsolutePath), 3, SORT_MANAGER)
+ Array(appLocalDir(app1Id.toString, "foo/foo"),
+ appLocalDir(app1Id.toString, "bar/bar")), 3, SORT_MANAGER)
val blockHandler = s1.blockHandler
val blockResolver = ShuffleTestAccessor.getBlockResolver(blockHandler)
blockResolver.registerExecutor(app1Id.toString, "exec-1", execShuffleInfo1)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
