[Part 5]: Rewriting Mantis Worker to be able to support starting tasks dynamically from the mantis master #153

sundargates · 2022-03-25T19:05:18Z

Context

Right now, the mantis worker assumes that the task it needs to perform will be seeded once as part of the environment or by MesosExecutor as part of the launchTask method. In the new Titus model, these tasks are received and acted upon asynchronously at the request of the mantis scheduler. This diff establishes the model for that by decoupling the act of running a task from how it is being received (for instance, Mesos Executor).

This will be followed by the TaskExecutor implementation which will use the building blocks created in this diff to execute tasks on demand.

Checklist

./gradlew build compiles code correctly
Added new tests where applicable
./gradlew test passes all tests
Extended README or added javadocs where applicable
Added copyright headers for new files from CONTRIBUTING.md

…s dyanmically

sundargates · 2022-03-25T20:06:51Z

...ane/mantis-control-plane-core/src/main/java/io/mantisrx/server/core/ExecuteStageRequest.java



-public class ExecuteStageRequest {
+public class ExecuteStageRequest implements Serializable {


Add a comment on why this needs to be serializable

should we do javadoc for these comments or is it fine to be not.

I don't have a preference. let me know if you prefer one format over another. Added documentation though.

...ane/mantis-control-plane-core/src/main/java/io/mantisrx/server/core/ExecuteStageRequest.java

sundargates · 2022-03-25T20:07:35Z

...ane/mantis-control-plane-core/src/main/java/io/mantisrx/server/core/ExecuteStageRequest.java

    private List<Parameter> parameters = new LinkedList<Parameter>();
    private SchedulingInfo schedulingInfo;
    private MantisJobDurationType durationType;
+    private Optional<String> nameOfJobProviderClass;


Documentation.

+1. Why jobProviderClass is in the request now? do we ignore the jobProviderClass inside the zip?

@calvin681 no this is just for testing purposes. Imagine running a job that's defined in the same codebase. In that case, the only definition that's needed is the job provider class.

sundargates · 2022-03-25T20:08:59Z

...mantis-server-worker/src/main/java/io/mantisrx/server/worker/ExecuteStageRequestService.java

+        WorkerExecutionOperations executionOperations,
+        Optional<String> jobProviderClass,
+        ClassLoaderHandle classLoaderHandle, Collection<URL> requiredClasspaths,
+        Job mantisJob) {


nit: make this nullable

Job is used in the spring boot path. We need to figure out how to load class for spring boot and get the Job bean.

I haven't looked too deep into this but I'm guessing this should be some version of the code you have here: https://sourcegraph.netflix.net/stash.corp.netflix.com/CPIE/nfmantis/-/blob/runtime-spring-boot/src/main/java/com/netflix/mantis/runtime/command/SpringReadJobFromZip.java?L65#tab=references

Basically, create an AnnotationBasedApplicationContext based on the classloader and then get the bean for the job class. I'll try to get a test spring-boot app working after the nodequark job migration.

sundargates · 2022-03-25T20:09:17Z

...mantis-server-worker/src/main/java/io/mantisrx/server/worker/ExecuteStageRequestService.java

-                        }
-                        logger.info("Creating job classpath with pathLocation " + pathLocation);
-                        ClassLoader cl = URLClassLoader.newInstance(new URL[] {pathLocation});
+


sundargates · 2022-03-25T20:11:10Z

...er/mantis-server-worker/src/main/java/io/mantisrx/server/worker/SimpleClassLoaderHandle.java

+import org.apache.flink.util.UserCodeClassLoader;
+
+@RequiredArgsConstructor(staticName = "of")
+public class SimpleClassLoaderHandle implements ClassLoaderHandle {


Documentation

sundargates · 2022-03-25T20:11:19Z

...ntis-server-worker/src/main/java/io/mantisrx/server/worker/SinkSubscriptionStateHandler.java

+import io.mantisrx.server.core.Service;
+import java.util.function.Function;
+
+public interface SinkSubscriptionStateHandler extends Service {


Documentation

hmitnflx · 2022-03-25T21:17:40Z

...ane/mantis-control-plane-core/src/main/java/io/mantisrx/server/core/ExecuteStageRequest.java



-public class ExecuteStageRequest {
+public class ExecuteStageRequest implements Serializable {


should we do javadoc for these comments or is it fine to be not.

hmitnflx · 2022-03-25T21:32:28Z

...mantis-server-worker/src/main/java/io/mantisrx/server/worker/ExecuteStageRequestService.java

-                            executeRequest.getStatus().onError(e1);
-                            return Observable.empty();
-                        }
-                        logger.info("Creating job classpath with pathLocation " + pathLocation);


nit: i prefer the logger.info("msg {}", pathLocation); format over string concat. It might be performant if we aren't logging the info log.

Agreed for both performance and stylistic reasons.

Actually, this was existing code that's been deleted.

hmitnflx · 2022-03-25T22:02:35Z

...ntis-server-worker/src/main/java/io/mantisrx/server/worker/SinkSubscriptionStateHandler.java

+
+  }
+
+  public static SinkSubscriptionStateHandler noop() {


I'd convert this to a static final constant to promote reuse.

Hmm, this is expected to be used only for testing purposes. I'm not sure if I want to expose this as a field and as a method.

hmitnflx · 2022-03-25T22:04:32Z

...ntis-server-worker/src/main/java/io/mantisrx/server/worker/SubscriptionStateHandlerImpl.java

+                                            subscriptionTimeoutSecs + " secs")
+                                            .subscribe();
+                                    // wait for kill to happen, we won't be running if it succeeds
+                                    try {Thread.sleep(60000);} // arbitrary sleep before we retry the kill


formatting and also a constant preferably configuration.

As before, this is existing code actually. If you refresh the PR, you should be able to see it now.

hmitnflx · 2022-03-25T22:05:28Z

...ntis-server-worker/src/main/java/io/mantisrx/server/worker/SubscriptionStateHandlerImpl.java

+        // let's call that instead.
+        onSinkSubscribed();
+        // let's shutdown the executor and reset it.
+        executor.shutdown();


there could be a NPE here that needs to be handled.

Cleaned up.

hmitnflx · 2022-03-25T22:06:16Z

...ntis-server-worker/src/main/java/io/mantisrx/server/worker/SubscriptionStateHandlerImpl.java

+        this.masterClientApi = masterClientApi;
+        this.subscriptionTimeoutSecs = subscriptionTimeoutSecs;
+        this.minRuntimeSecs = minRuntimeSecs;
+        executor = this.subscriptionTimeoutSecs > 0L ? new ScheduledThreadPoolExecutor(1) : null;


this might be a good candidate for optional IMO.

Agreed. Do we have a team preference for the Java Optional or the Vavr one? I suppose this sort of a larger discussion about which libraries we do or don't consider part of the canonical set in our code bases. I know introducing something like Vavr has larger implications for how code should be written.

This is existing code. I just deleted the file that this was supposed to replace, and you should be able to see the changes I made to this file now.

I have completely refactored the code. Can you guys take a look now? It should be way simpler than the original version.

codyrioux

LGTM with the exception of the noted changes.

codyrioux · 2022-03-25T22:21:54Z

...mantis-server-worker/src/main/java/io/mantisrx/server/worker/ExecuteStageRequestService.java

-                            executeRequest.getStatus().onError(e1);
-                            return Observable.empty();
-                        }
-                        logger.info("Creating job classpath with pathLocation " + pathLocation);


Agreed for both performance and stylistic reasons.

codyrioux · 2022-03-25T22:26:59Z

...mantis-server-worker/src/main/java/io/mantisrx/server/worker/ExecuteStageRequestService.java

+        final UserCodeClassLoader userCodeClassLoader =
+             classLoaderHandle.getOrResolveClassLoader(ImmutableList.of(executeStageRequest.getJobJarUrl().toURI()), requiredClasspaths);
+
+        log.info(


It may be wise to also add a metric for this so we can alert in the event that our artifact downloads go latent.

codyrioux · 2022-03-25T22:31:53Z

...s-server/mantis-server-worker/src/main/java/io/mantisrx/server/worker/InstantiationUtil.java

+   */
+  public static <T> T instantiate(Class<T> clazz) {
+    if (clazz == null) {
+      throw new NullPointerException();


Give me an error message here so I know what was null. :)

Actually, this is not expected at all as the clazz is always supposed to be some value. So, the null pointer exception along with the stack trace should be enough to understand the semantics.

codyrioux · 2022-03-25T22:37:42Z

...s-server/mantis-server-worker/src/main/java/io/mantisrx/server/worker/InstantiationUtil.java

+    final Class<? extends T> clazz;
+    try {
+      clazz = Class.forName(className, false, classLoader).asSubclass(targetType);
+    } catch (ClassNotFoundException e) {


We may want to add either specific catches for ExceptionInInitializerError and LinkageError. They aren't checked exceptions but this method can throw them. See the forName(String, boolean, ClassLoader) docs.

codyrioux · 2022-03-25T22:43:16Z

...ntis-server-worker/src/main/java/io/mantisrx/server/worker/SubscriptionStateHandlerImpl.java

+        this.masterClientApi = masterClientApi;
+        this.subscriptionTimeoutSecs = subscriptionTimeoutSecs;
+        this.minRuntimeSecs = minRuntimeSecs;
+        executor = this.subscriptionTimeoutSecs > 0L ? new ScheduledThreadPoolExecutor(1) : null;


Agreed. Do we have a team preference for the Java Optional or the Vavr one? I suppose this sort of a larger discussion about which libraries we do or don't consider part of the canonical set in our code bases. I know introducing something like Vavr has larger implications for how code should be written.

codyrioux · 2022-03-25T22:46:36Z

...ntis-server-worker/src/main/java/io/mantisrx/server/worker/SubscriptionStateHandlerImpl.java

+        this.masterClientApi = masterClientApi;
+        this.subscriptionTimeoutSecs = subscriptionTimeoutSecs;
+        this.minRuntimeSecs = minRuntimeSecs;
+        executor = this.subscriptionTimeoutSecs > 0L ? new ScheduledThreadPoolExecutor(1) : null;


On this note, what if we have a subscription timeout of zero? Is that a valid value for users to input? We may want to consider >= 0L here. Further is it valid at all to instantiate this class if the executor will be null?

codyrioux · 2022-03-25T23:14:05Z

mantis-server/mantis-server-worker/src/main/java/io/mantisrx/server/worker/Task.java

+
+@Slf4j
+@RequiredArgsConstructor
+public class Task extends AbstractIdleService {


Documentation for this one.

calvin681 · 2022-03-26T00:04:35Z

...ane/mantis-control-plane-core/src/main/java/io/mantisrx/server/core/ExecuteStageRequest.java

    private List<Parameter> parameters = new LinkedList<Parameter>();
    private SchedulingInfo schedulingInfo;
    private MantisJobDurationType durationType;
+    private Optional<String> nameOfJobProviderClass;


+1. Why jobProviderClass is in the request now? do we ignore the jobProviderClass inside the zip?

calvin681 · 2022-03-26T00:09:49Z

.../mantis-control-plane-core/src/main/java/io/mantisrx/server/core/metrics/MetricsFactory.java

+   * @param request request for which the metrics need to be published
+   * @return MetricsServerService server
+   */
+  public static MetricsServerService newMetricsServer(CoreConfiguration configuration, ExecuteStageRequest request) {


If we are moving to a factory, should we make these singletons as well?

calvin681 · 2022-03-26T00:20:49Z

...mantis-server-worker/src/main/java/io/mantisrx/server/worker/ExecuteStageRequestService.java

-                        executionOperations.shutdownStage();
+                        try {
+                            executionOperations.shutdownStage();
+                        } catch (IOException e) {


Should you still close cl?

CL closure has been added recently. Can you refresh the set of commits and check if looks okay?

calvin681 · 2022-03-26T00:23:23Z

...mantis-server-worker/src/main/java/io/mantisrx/server/worker/ExecuteStageRequestService.java

+        WorkerExecutionOperations executionOperations,
+        Optional<String> jobProviderClass,
+        ClassLoaderHandle classLoaderHandle, Collection<URL> requiredClasspaths,
+        Job mantisJob) {


Job is used in the spring boot path. We need to figure out how to load class for spring boot and get the Job bean.

...s-server/mantis-server-worker/src/main/java/io/mantisrx/server/worker/InstantiationUtil.java

sundargates added 2 commits March 24, 2022 10:39

Adding some code

1dad4dc

[Part 5]: Rewriting Mantis Worker to be able to support starting task…

e88862b

…s dyanmically

sundargates requested review from Andyz26, calvin681, codyrioux, hmitnflx, jeffchao, nickmahilani and piygoyal as code owners March 25, 2022 19:05

sundargates commented Mar 25, 2022

View reviewed changes

hmitnflx reviewed Mar 25, 2022

View reviewed changes

codyrioux suggested changes Mar 25, 2022

View reviewed changes

sundargates added 2 commits March 25, 2022 17:05

Addressing comments

92d7f33

Addressing comments

51c9414

calvin681 reviewed Mar 26, 2022

View reviewed changes

sundargates added 3 commits March 26, 2022 19:26

Addressing comments

246b9c0

Adding tests and cleaning up the code

161687c

Addressing comments

01f4780

calvin681 approved these changes Mar 27, 2022

View reviewed changes

Made sure the task updates are reported back to the gateway

8ceb5c2

sundargates merged commit 5f18638 into Netflix:master Mar 27, 2022



		public class ExecuteStageRequest {
		public class ExecuteStageRequest implements Serializable {

[Part 5]: Rewriting Mantis Worker to be able to support starting tasks dynamically from the mantis master #153

[Part 5]: Rewriting Mantis Worker to be able to support starting tasks dynamically from the mantis master #153

Uh oh!

Conversation

sundargates commented Mar 25, 2022

Context

Checklist

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codyrioux left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sundargates Mar 26, 2022 •

edited

Loading