[BEAM-4778] add option to flink job server to clean staged artifacts per-job #5958

ryan-williams · 2018-07-16T06:19:39Z

when set, the InMemoryJobService subscribes to invoked jobs' state-changes, and when it sees them complete, removes all associated artifacts

A few incidental moves/changes:

expose BeamFileSystemArtifactRetrievalService.loadManifest publicly
moved StagingSessionToken JSON serde into methods

R: @angoenka

Post-Commit Tests Status (on master branch)

Lang	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go	---	---	---	---	---	---
Java
Python	---		---	---	---	---

ryan-williams · 2018-07-16T18:37:37Z

Looking over this again, I'm going to revert the bits that make StagingSessionToken public, I thought it was nicer to deal with the structured object in a few places instead of an opaque String, but the test usages in particular make it seem like keeping it as an opaque String is deliberate / has advantages

angoenka

Thanks for taking this up.
Added a few comments.
Please add some test cases for the this.

angoenka · 2018-07-16T19:28:58Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java

  private final JobInvoker invoker;

  private InMemoryJobService(
      Endpoints.ApiServiceDescriptor stagingServiceDescriptor,
      Function<String, String> stagingServiceTokenProvider,
+      ThrowingConsumer<String> cleanupJobFn,


Lets make it a jobTerminationListener

angoenka · 2018-07-16T19:30:08Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java

+              state.equals(JobState.Enum.DONE) || state.equals(JobState.Enum.FAILED)) {
+                String stagingSessionToken = stagingSessionTokens.get(preparationId);
+                try {
+                  cleanupJobFn.accept(stagingSessionToken);


Given its job Termination/Cleanup listener, we should be passing the job id or job object.

angoenka · 2018-07-16T19:31:56Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java

@@ -70,26 +71,43 @@
  public static InMemoryJobService create(
      Endpoints.ApiServiceDescriptor stagingServiceDescriptor,
      Function<String, String> stagingServiceTokenProvider,
+      ThrowingConsumer<String> cleanupJobFn,
+      Boolean cleanArtifactsPerJob,


We don't need the boolean. We can simply set the jobTerminationListener in previous line or not set it.
Alternatively the jobTerminationListener can have an internal check for this boolean.

angoenka · 2018-07-16T19:45:50Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java

  }

  private final ConcurrentMap<String, JobPreparation> preparations;
  private final ConcurrentMap<String, JobInvocation> invocations;
+  private final ConcurrentMap<String, String> stagingSessionTokens;


We should also cleanup the stagingSessionTokens on job completion.

ok; I've made it so that we always clean them up on job termination, regardless of whether a cleanup hook is set.

my understanding is that we have no plans to use them beyond a job's life-span, so this falls under "not leaking memory from this map".

angoenka · 2018-07-16T19:49:48Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

@@ -305,7 +311,7 @@ public void onCompleted() {
   * Serializable StagingSessionToken used to stage files with {@link
   * BeamFileSystemArtifactStagingService}.
   */
-  private static class StagingSessionToken implements Serializable {
+  static class StagingSessionToken implements Serializable {


We should keep the StagingSessionToken private as no one else should care about its structure outside this class.

angoenka · 2018-07-16T19:51:36Z

...ava/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactRetrievalService.java

+    return loadManifest(manifestResourceId);
+  }
+
+  public static ProxyManifest loadManifest(ResourceId manifestResourceId) throws IOException {


Lets keep it package private as BeamFileSystemArtifactRetricalService is the only other user.

angoenka · 2018-07-16T19:57:08Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

+    LOG.info("Removing manifest: {}", manifestResourceId);
+    FileSystems.delete(Collections.singletonList(manifestResourceId));
+    LOG.info("Removing empty dir: {}", dir);
+    FileSystems.delete(Collections.singletonList(dir));


We should check if the directory is empty or not.

If it's not, currently we'll get a DirectoryNotEmptyException, which seems basically desirable?

afaict FileSystems offers no way to recursively delete, cf. BEAM-4843

for now, it seems like throwing an exception if the structure has changed from what is assumed here is the correct behavior, to me

DirectoryNotEmptyException is good enough.

ryan-williams

I think I responded to everything.

I added test logic to BeamFileSystemArtifactServicesTest and InMemoryJobServiceTest that covers most of what's new here.

However, some of the plumbing for the directory-removal callback happens in FlinkJobServerDriver which doesn't have its own test atm.

I looked at adding one but it feels like it gets into IT / VR test territory a bit, or would want a TestPortableRunner, so I held off on that for now, but if you'd like me to forge ahead with one of those approaches lmk!

ryan-williams · 2018-07-22T20:48:59Z

...ava/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactRetrievalService.java

+    return loadManifest(manifestResourceId);
+  }
+
+  public static ProxyManifest loadManifest(ResourceId manifestResourceId) throws IOException {


ryan-williams · 2018-07-22T20:59:13Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

+    LOG.info("Removing manifest: {}", manifestResourceId);
+    FileSystems.delete(Collections.singletonList(manifestResourceId));
+    LOG.info("Removing empty dir: {}", dir);
+    FileSystems.delete(Collections.singletonList(dir));


If it's not, currently we'll get a DirectoryNotEmptyException, which seems basically desirable?

afaict FileSystems offers no way to recursively delete, cf. BEAM-4843

for now, it seems like throwing an exception if the structure has changed from what is assumed here is the correct behavior, to me

ryan-williams · 2018-07-22T21:01:20Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java

  }

  private final ConcurrentMap<String, JobPreparation> preparations;
  private final ConcurrentMap<String, JobInvocation> invocations;
+  private final ConcurrentMap<String, String> stagingSessionTokens;


ok; I've made it so that we always clean them up on job termination, regardless of whether a cleanup hook is set.

my understanding is that we have no plans to use them beyond a job's life-span, so this falls under "not leaking memory from this map".

ryan-williams · 2018-07-22T21:02:44Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java

@@ -70,26 +71,43 @@
  public static InMemoryJobService create(
      Endpoints.ApiServiceDescriptor stagingServiceDescriptor,
      Function<String, String> stagingServiceTokenProvider,
+      ThrowingConsumer<String> cleanupJobFn,
+      Boolean cleanArtifactsPerJob,


ryan-williams · 2018-07-22T21:31:09Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java

+              state.equals(JobState.Enum.DONE) || state.equals(JobState.Enum.FAILED)) {
+                String stagingSessionToken = stagingSessionTokens.get(preparationId);
+                try {
+                  cleanupJobFn.accept(stagingSessionToken);


ryan-williams · 2018-07-22T21:31:18Z

...tion/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java

  private final JobInvoker invoker;

  private InMemoryJobService(
      Endpoints.ApiServiceDescriptor stagingServiceDescriptor,
      Function<String, String> stagingServiceTokenProvider,
+      ThrowingConsumer<String> cleanupJobFn,


ryan-williams · 2018-07-22T21:32:35Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

@@ -305,7 +311,7 @@ public void onCompleted() {
   * Serializable StagingSessionToken used to stage files with {@link
   * BeamFileSystemArtifactStagingService}.
   */
-  private static class StagingSessionToken implements Serializable {
+  static class StagingSessionToken implements Serializable {


ryan-williams · 2018-07-23T03:52:10Z

Run Java PreCommit

ryan-williams · 2018-07-23T04:28:51Z

First Jenkins build failed with a lot of messages like:

Expiring Daemon because JVM Tenured space is exhausted

and eventually:

Build timed out (after 90 minutes). Marking the build as aborted.

Now I'm seeing the first message in the latest build as well.

Any ideas what that's about are welcome!

ryan-williams · 2018-07-23T14:15:48Z

Run Java PreCommit

ryan-williams · 2018-07-23T16:06:11Z

Same thing 3x here, not sure how to triage.

Seeing a timed out build on #6018 with a lot of OOMs but not the Tenured space is exhausted msg.

ryan-williams · 2018-07-23T21:43:16Z

Looks like this is tracked at BEAM-4847, per @lukecwik on Slack (thanks!)

Not sure if I should keep trying here or wait until it seems like there's some movement on that issue?

angoenka

Thanks!

angoenka · 2018-07-24T17:30:00Z

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobServerDriver.java

+        configuration.cleanArtifactsPerJob ?
+            (String stagingSessionToken) ->
+                artifactStagingService.getService().removeArtifacts(stagingSessionToken)
+            : null,


This will throw NullPointer on accept call if configuration.cleanArtifactsPerJob is false.

I was suggesting:

(String stagingSessionToken) -> if(configuration.cleanArtifactsPerJob) artifactStagingService.getService().removeArtifacts(stagingSessionToken)

This will throw NullPointer on accept call if configuration.cleanArtifactsPerJob is false.

I don't think that's true, as written? I check cleanupJobFn != null in the only place it's referenced in InMemoryJobService.

Or maybe I'm missing what you mean?

Anyway, if you think this way is cleaner I'm almost fine with it, lmk!

I changed it to what you suggested, anyway

angoenka · 2018-07-24T17:32:44Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

+    LOG.info("Removing manifest: {}", manifestResourceId);
+    FileSystems.delete(Collections.singletonList(manifestResourceId));
+    LOG.info("Removing empty dir: {}", dir);
+    FileSystems.delete(Collections.singletonList(dir));


DirectoryNotEmptyException is good enough.

ryan-williams · 2018-07-24T19:51:22Z

Merged in HEAD, let's see if the Java PreCommit issue has been resolved 🤞🏻

ryan-williams · 2018-07-25T00:48:57Z

whew, finally passed precommit. lmk if you want me to squash+rebase @angoenka (or if there are any other comments I missed! I think I got everything)

angoenka · 2018-07-30T19:41:08Z

The PR looks good.
We can simplify it a bit by reusing FinkJobInvokation#addStateListener to listen to the termination instead of adding a separate listener.
But we can take it up later.
Otherwise, the PR looks good.

angoenka · 2018-08-01T20:44:02Z

@ryan-williams Is the PR good for merge?
cc: @tweise

ryan-williams · 2018-08-02T03:07:09Z

I removed the terminationListener per your suggestion @angoenka; it should be ready to go!

tweise · 2018-08-02T03:49:52Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

+      try {
+        return MAPPER.writeValueAsString(this);
+      } catch (JsonProcessingException e) {
+        LOG.error("Error {} occurred while serializing {}.", e.getMessage(), this);


Why logging vs. using StatusRuntimeException as shown below?

good catch, done

tweise · 2018-08-02T03:50:46Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

-    } catch (JsonProcessingException e) {
-      LOG.error("Error {} occurred while serializing {}.", e.getMessage(), stagingSessionToken);
-      throw e;
+    LOG.info("Removing dir {}", dir);


Do we really this and following logging at info level? Why not debug?

good point; I made just the last one info and the rest debug, lmk if you want them all as debug

ryan-williams

thanks @tweise, I think I addressed both comments (and hopefully the CI issue)

ryan-williams · 2018-08-02T04:25:03Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

-    } catch (JsonProcessingException e) {
-      LOG.error("Error {} occurred while serializing {}.", e.getMessage(), stagingSessionToken);
-      throw e;
+    LOG.info("Removing dir {}", dir);


good point; I made just the last one info and the rest debug, lmk if you want them all as debug

ryan-williams · 2018-08-02T04:25:24Z

.../java/org/apache/beam/runners/fnexecution/artifact/BeamFileSystemArtifactStagingService.java

+      try {
+        return MAPPER.writeValueAsString(this);
+      } catch (JsonProcessingException e) {
+        LOG.error("Error {} occurred while serializing {}.", e.getMessage(), this);


good catch, done

ryan-williams · 2018-08-02T13:44:03Z

Thanks for the spotless fix; I was mistakenly only running it on beam-runners-java-fn-execution!

I see another spotless nit locally, will push it if this build fails (which it presumably will?)

tweise · 2018-08-02T15:03:33Z

FYI this can be avoided by running ./gradlew spotlessApply prior to opening the PR.

tweise · 2018-08-02T15:12:35Z

Run Java Precommit

ryan-williams requested review from aaltay, kennknowles and lukecwik as code owners July 16, 2018 06:19

add option to flink job server to clean staged artifacts per-job

a60b9b5

ryan-williams force-pushed the rm-staged-artifacts branch from 1407f70 to a60b9b5 Compare July 16, 2018 13:45

re-privatize StagingSessionToken

f4b1c02

angoenka reviewed Jul 16, 2018

View reviewed changes

Merge remote-tracking branch 'upstream/HEAD' into staging

872e3c8

ryan-williams force-pushed the rm-staged-artifacts branch from fcd8069 to 7ca46ab Compare July 23, 2018 02:07

ryan-williams commented Jul 23, 2018

View reviewed changes

code review feedback; begin adding tests

d3c9465

ryan-williams force-pushed the rm-staged-artifacts branch from 7ca46ab to d3c9465 Compare July 23, 2018 02:16

angoenka reviewed Jul 24, 2018

View reviewed changes

Merge remote-tracking branch 'upstream/HEAD' into staging

748c255

ryan-williams added 4 commits July 24, 2018 20:40

spotless nits

cd77363

cr feedback

da17413

Merge remote-tracking branch 'upstream/HEAD' into staging

588e29a

style nit

2fdeaea

angoenka approved these changes Jul 30, 2018

View reviewed changes

Merge remote-tracking branch 'upstream/HEAD' into rm-staged-artifacts

0396bec

rm JobInvocation.terminationListener, reuse stateListener

ebd733d

ryan-williams force-pushed the rm-staged-artifacts branch from aae17aa to ebd733d Compare August 2, 2018 03:06

tweise reviewed Aug 2, 2018

View reviewed changes

spotless nit, cr feedback

38135c5

ryan-williams commented Aug 2, 2018

View reviewed changes

Fix checkstyle errors

0465d67

ryan-williams force-pushed the rm-staged-artifacts branch from 1d4a880 to 3711dbf Compare August 2, 2018 14:16

another spotless nit

9c7130e

ryan-williams force-pushed the rm-staged-artifacts branch from 3711dbf to 9c7130e Compare August 2, 2018 14:47

fix test compile

dc8da43

tweise merged commit 8c28d3e into apache:master Aug 2, 2018

ryan-williams mentioned this pull request Aug 12, 2018

[BEAM-4176] clean up artifacts per-job in portable VR tests #6209

Merged

[BEAM-4778] add option to flink job server to clean staged artifacts per-job #5958

[BEAM-4778] add option to flink job server to clean staged artifacts per-job #5958

Conversation

ryan-williams commented Jul 16, 2018 • edited

Post-Commit Tests Status (on master branch)

ryan-williams commented Jul 16, 2018

angoenka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

angoenka Jul 16, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

angoenka Jul 16, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

angoenka Jul 16, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryan-williams left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryan-williams commented Jul 23, 2018

ryan-williams commented Jul 23, 2018

ryan-williams commented Jul 23, 2018

ryan-williams commented Jul 23, 2018

ryan-williams commented Jul 23, 2018

angoenka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryan-williams commented Jul 24, 2018

ryan-williams commented Jul 25, 2018

angoenka commented Jul 30, 2018

angoenka commented Aug 1, 2018

ryan-williams commented Aug 2, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tweise Aug 2, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryan-williams left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryan-williams commented Aug 2, 2018

tweise commented Aug 2, 2018

tweise commented Aug 2, 2018

ryan-williams commented Jul 16, 2018 •

edited

angoenka Jul 16, 2018 •

edited

angoenka Jul 16, 2018 •

edited

angoenka Jul 16, 2018 •

edited

tweise Aug 2, 2018 •

edited