[SPARK-37618][CORE] Remove shuffle blocks using the shuffle service for released executors

[Kimahriman]

What changes were proposed in this pull request?

Add support for removing shuffle files on released executors via the external shuffle service. The shuffle service already supports removing shuffle service cached RDD blocks, so I reused this mechanism to remove shuffle blocks as well, so as not to require updating the shuffle service itself.

To support this change functioning in a secure Yarn environment, I updated permissions on some of the block manager folders and files. Specifically:

• Block manager sub directories have the group write posix permission added to them. This gives the shuffle service permission to delete files from within these folders.
• Shuffle files have the world readable posix permission added to them. This is because when the sub directories are marked group writable, they lose the setgid bit that gets set in a secure Yarn environment. Without this, the permissions on the files would be rw-r-----, and since the group running Yarn (and therefore the shuffle service), is no longer the group owner of the file, it does not have access to read the file. The sub directories still do not have world execute permissions, so there's no security issue opening up these files.

Both of these changes are done after creating a file so that umasks don't affect the resulting permissions.

Why are the changes needed?

External shuffle services are very useful for long running jobs and dynamic allocation. However, currently if an executor is removed (either through dynamic deallocation or through some error), the shuffle files created by that executor will live until the application finishes. This results in local disks slowly filling up over time, eventually causing problems for long running applications.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

New unit test. Not sure if there's a better way I could have tested for the files being deleted or any other tests I should add.

[Kimahriman]

I wasn't sure whether I should make this backward compatible or not, so I added a default implementation. I can remove it if it doesn't need to be backward compatible (which it's marked private so probably doesn't need to be?)

[mridulm]

I agree with your assessment @Kimahriman and would prefer to avoid this methough, though @Ngone51 had expressed concern in the past that some of the private classes might be getting used for some custom implementations.
Thoughts @Ngone51 ?

[Ngone51]

I'd be +1 for keeping backward compatible if you want to keep this method.

[AmplabJenkins]

Can one of the admins verify this patch?

[mridulm]

+CC @otterc Can you take a look at this PR ? It currently does not handle push based shuffle - would be great if you can suggest changes to accommodate that as well.

[mridulm]

Had a Q, @HyukjinKwon, @dongjoon-hyun - why is AmplabJenkins still commenting ? I thought we had turned jenkins off ?

[Kimahriman]

+CC @otterc Can you take a look at this PR ? It currently does not handle push based shuffle - would be great if you can suggest changes to accommodate that as well.

Yeah thought about that. Haven't used push based shuffle yet or fully know how it works so mostly didn't want to break it. I can look into that here with some hints (look at mergeStasus in addition to mapStatus?). Or that can be a separate follow up.

[dongjoon-hyun]

@mridulm I have no context for that either~ :)

[HyukjinKwon]

I have no idea too :-). I think the machines are still running assuming that it still links JIRA <> PRs. I remember that program is running separately in one of the Jenkins machines. So, i suspect that Jenkins builds are somehow shut down but the machines are still alive .. we can ignore it for now I think.

[otterc]

+CC @otterc Can you take a look at this PR ? It currently does not handle push based shuffle - would be great if you can suggest changes to accommodate that as well.

I think the same mechanism can be used for push-based shuffle. We can look at the mergeStatuses as well on the driver. On the ExternalShuffleService side, execId in the RemoveBlocks can differentiate whether it is a push-merged block or regular one and let RemoteBlockPushResolver handle removal of push-merged files.
I think we can handle this in a separate jira. cc. @mridulm @Kimahriman

[Kimahriman]

I tested this out in my actual environment and realized that it didn't actually work because the context cleaner remove the map output before removing the shuffle from the block manager master. I switched the order of those and updated to test to user the cleaner to replicate and it sort of started working.

I say sort of because I realized a major potential issue with this. We use a private secured Hadoop cluster and the shuffle files were not writable by the user running the shuffle service so they couldn't be deleted. I still need to do a little digging to see if the permissions for those files (via Spark or Hadoop) are configurable at all through the node managers.

[Kimahriman]

I added a utility function to create the shuffle files as group writable. I haven't tried this out in my environment yet but I think it'd be necessary for this to work (in a secured environment). If there are security concerns with this I can bring it up in the dev mailing list. Also don't know if this will break or be ignored on non-posix compliant systems, so definitely want to get some thoughts on this

[mridulm]

Took a quick pass through the PR, thanks for working on this @Kimahriman !

[mridulm]

+CC @otterc There were some concerns with PosixFilePermissions which resulted in addition of DiskBlockManager.createDirWithPermission770
Will this be affected by the same ?

[Kimahriman]

Yeah I saw that and wasn't sure what the background was that didn't work in the past.

[Kimahriman]

Based on the comment

* TODO: Find out why can't we create a dir using java api with permission 770
*  Files.createDirectories(mergeDir.toPath, PosixFilePermissions.asFileAttribute(
*  PosixFilePermissions.fromString("rwxrwx---")))

I'm thinking it might be a umask issue? The permissions you give it during dir creation will be affected by the umask, and the way around that is to create the dir and then change its permissions.

[mridulm]

IIRC this was not a umask issue (I had asked similar question during review :-) ).
Will let @otterc comment more though - she has better context.

[Kimahriman]

Think I figured it out while testing this out in our env. Java has no way to set or keep the setgid bit when setting file permissions, so you lose the setgid bit with any of these methods of setting file permissions. I'm trying to figure out how yarn creates these folders since it manages to keep the bit, but it looks like it's doing it all through a local FileContext, which seems to be not keeping the setgid bit when making directories, so I'm not sure how that works.

Will keep investigating but will probably use the existing method to create the 770 folder

[Kimahriman]

Weird, it worked when I tried for the test case and when I tested it on my environment

[otterc]

hmmm I think even @zhouyejoe tried that in our environment. @zhouyejoe do you remember whether that worked?

[otterc]

I will give it a try again. Maybe made some mistake. Can't find any reason for it to not work. If it works, then we can clean it up.

[akpatnam25]

I was looking into this in our environment and was able to get the below piece of code working:
if (dirToCreate.mkdirs() && dirToCreate.exists()) { Files.setPosixFilePermissions(dirToCreate.toPath, PosixFilePermissions.fromString("rwxrwx---")) created = dirToCreate }
@Kimahriman do you mind testing this piece of code in your environment? It seems like the code referenced in the TODO does not pass the unit tests and also does not work in our environment.
If this looks good we can clean it up.
cc: @zhouyejoe

[Kimahriman]

Yeah that's the umask issue. The todo does permission setting in the dir creation. If you setPosixFilePermissions after creating the directory you don't get limited by the umask (only weird behavior with the setgid bit mentioned in this thread). I've tried this approach in my env as well and it works in place of the createDirWith770 method

[Kimahriman]

I tried adding a setfacl -d -m g::rwx <path> on the blockmgr directory via processbuilder after it's created and that finally got it working in my environment. Theoretically this should only be a problem on Linux systems (there's a yarn shuffle service and spark standalone shuffle service, yarn is probably running on linux and spark is always the same user? not sure if spark standalone has run as user support?). But it still seems a little gross. I can't think of any other way though other than trying to get sh -c "umask 0007 && mkdir -p <path>" working for the create dir with 770 call

[Kimahriman]

I guess another way that could work too is create the sub dirs with 770 using setPosixPermissions, and then also make all shuffle files world readable to get around the fact that they won't be owned by the sub dir group because the setgid bit is dropped.

[otterc]

Created https://issues.apache.org/jira/browse/SPARK-38005 for push-based shuffle

[Kimahriman]

I went with the make subdirs 770 and make files world readable approach, seemed a lot less hacky than relying on ACLs. Will update the description with the approach

[mridulm]

Thanks for working on this @Kimahriman !
This is very useful, particularly in environments with dynamic resource allocation turned on.

[mridulm]

I agree with your assessment @Kimahriman and would prefer to avoid this methough, though @Ngone51 had expressed concern in the past that some of the private classes might be getting used for some custom implementations.
Thoughts @Ngone51 ?

[mridulm]

Add GROUP_READ and GROUP_EXECUTE (for listing) as well.

[Kimahriman]

The idea was it should already have everything else it needs (because it needs to to function), so change as little as possible and just add group write. I could just update to fully set as 770 instead

[mridulm]

Instead of Files.createDirectory, using DiskBlockManager.createDirWithPermission770 (or move it to utils) should suffice - and we wont need to change the perms.

It is unfortunate that jdk still has not fixed this bug.

[Kimahriman]

From what I understand, createDirWithPermission770 was made because it was thought it couldn't be done through the Java API, but this works in my environment through the Java API correctly. See later in that same thread for that discussion: #35085 (comment)

Because it might be possible to remove that helper after all (still waiting to hear if there's some other reason create then set permission won't work in other environments), I just stuck with the nio API.

From what I've found, losing the setgid bit isn't actually Java related. mkdir -p -m770 also removes the setgid bit because of the way the chmod Linux system call works (in the operating systems I tested at least). Despite the man page saying that it preservers the setgid bit on directories, the same behavior that the man page lists for files seems to apply to directories. If the user isn't in the group of the directory, it will remove the setgid bit.

[Kimahriman]

Now I can't recreate mkdir losing the setgid bit part...

[Kimahriman]

Yeah definitely not ideal, but the permissions of the parent folders should prevent this from actually opening up any security holes I believe

[tgravescs]

YARN has a deletion service that it uses to remove the files normally (as the user), I started to look at this but I'm not sure the shuffle manager can get a handle to it.

If we are going to take this approach, I want to only change permissions when this feature is enabled. Can we limit it to just when creating the shuffle files?
Ideally I would also like it documented.

[Kimahriman]

Yeah I hoped there was a way to hook into the yarn run as user in the shuffle service but I couldn't see anyway to do that

[Kimahriman]

I put all the permissions changes inside checks for the feature being enabled. I also removed the permission change from the temp file creation, I think the actual shuffle files only go through the createTempFileWith

[Kimahriman]

Just kidding not actually true, need to trace through and figure out but some final shuffle blocks must be created with createTempShuffleBlock

[tgravescs]

it would be nice to update description to say when we remove them. Is there a config to disable this functionality, didn't see one? I think the removeBlocks was added in 3.0 so probably fairly safe for compatibility but it might be nice to have it behind a feature flag just in case someone has weird shuffle service or there is just issues with the feature (unless we are really confident in it). open to thoughts on this..

[tgravescs]

thinking a bit more about this, I would like to see it behind a feature flag. how long does asking for the removal take? I assume its pretty small but would be good to measure. It does looks like waiting on removal is done in background by default (CLEANER_REFERENCE_TRACKING_BLOCKING_SHUFFLE) so won't block application side. On very busy external shuffle service this is just going to put more pressure on node manager as well as potential disk load.

[Kimahriman]

Feature flag defaulting on or off?

[tgravescs]

depends on how confident we are in it? have you done any testing on real cluster at scale?

[Kimahriman]

Yeah i've been using it on our production cluster (hundreds of nodes). When there were permission issues it failed pretty quickly. Haven't actually seen how long it takes but it's been removing ~10+ TB shuffles fine. Can do a little closer monitoring

[tgravescs]

seems fine to have on by default then.

[Kimahriman]

The second bullet is introduced by this PR, but since it's only a bug if you enable this new functionality is that also ok to be part of the separate issue?

[tgravescs]

sorry, I read that too quickly, yes we should fix the RDD fetch under this as we should not break one feature for another.
I think we essentially need something that says if we have modified the directory permissions which causes the setgid bit to be lost, then all files that need to be read by the external shuffle service under that need to be world readable, correct? I'm not sure about the paths off the top of my head, I can go look. @attilapiros any other input on that?

The only part that I guess we could leave off is updating the folder permissions for RDD fetch only.

[Kimahriman]

Yeah basically. It's kinda like:

• If RDD fetching or shuffle removal is enabled, make the sub dirs group writable (and lose the setgid bit)
  • If RDD fetching is enabled, make the rdd blocks world readable (the DiskStore update)
  • Shuffle files always have to be made world readable if either is enabled

[Kimahriman]

Added the update for RDD blocks and checking that setting, and added a few more checks to the tests as well

[Kimahriman]

Found an separate issue with the RDD fetching while testing this ^

[tgravescs]

just a couple doc requests

[mridulm]

Use shufflerStatus.withMapStatuses

[Kimahriman]

good call, done

[tgravescs]

Merged to master and branch-3.3. @Kimahriman if you want this on by default in next release please file a separate issue to discuss it.

[Kimahriman]

Sounds good!

[wangyum]

Do we need to upgrade shuffle service to enable this feature?

[zhouyejoe]

Which version are you using now? It is only available at or after 3.3.0

[wangyum]

Which version are you using now? It is only available at or after 3.3.0

Spark 3.5.0, but Spark shuffle service is Spark 3.0.0.

[Kimahriman]

Spark 3.5.0, but Spark shuffle service is Spark 3.0.0.

I think it might work with a 3.0.0 shuffle service. That's when #24499 was added which includes the deletion updates on the shuffle service this relies on.