[Part 1]: Support for hot task executors in Mantis

.github/CODEOWNERS

@@ -1 +1 @@
-* @codyrioux @neerajrj @nickmahilani @jeffchao @piygoyal @calvin681 @sundargates @Andyz26
+* @codyrioux @nickmahilani @jeffchao @piygoyal @calvin681 @sundargates @Andyz26 @hmitnflx

build.gradle

@@ -56,6 +56,9 @@ subprojects {
     // Apply lombok plugin and disabled the default config file generation.
     apply plugin: "io.freefair.lombok"
     generateLombokConfig.enabled = false
+    lombok {
+        version = "1.18.20"
+    }
 
     group = 'io.mantisrx'
 

lombok.config

@@ -0,0 +1 @@
+lombok.anyConstructor.addConstructorProperties = true

mantis-common/src/main/java/io/mantisrx/common/Ack.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.common;
+
+import java.io.Serializable;
+import lombok.Value;
+
+/**
+ * Ack is sent whenever an effect has taken place on the side of the receiver.
+ * Ack instance supports both JSON / Java serialization.
+ */
+@Value(staticConstructor = "getInstance")
+public class Ack implements Serializable {
+}

mantis-common/src/main/java/io/mantisrx/common/JsonSerializer.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.common;
+
+import io.mantisrx.shaded.com.fasterxml.jackson.core.type.TypeReference;
+import io.mantisrx.shaded.com.fasterxml.jackson.databind.DeserializationFeature;
+import io.mantisrx.shaded.com.fasterxml.jackson.databind.ObjectMapper;
+import io.mantisrx.shaded.com.fasterxml.jackson.databind.SerializationFeature;
+import io.mantisrx.shaded.com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
+import io.mantisrx.shaded.com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import java.io.IOException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JsonSerializer {
+    private static final Logger logger = LoggerFactory.getLogger(JsonSerializer.class);
+
+    private static final ObjectMapper defaultObjectMapper = new ObjectMapper()
+            .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+            .configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false)
+            .registerModule(new Jdk8Module());
+    public static final SimpleFilterProvider DEFAULT_FILTER_PROVIDER;
+
+    static {
+        DEFAULT_FILTER_PROVIDER = new SimpleFilterProvider();
+        DEFAULT_FILTER_PROVIDER.setFailOnUnknownId(false);
+    }
+
+    public <T> T fromJSON(String json, Class<T> expectedType) throws IOException {
+        return defaultObjectMapper.readerFor(expectedType).readValue(json);
+    }
+
+    public <T> T fromJSON(String json, TypeReference<T> expectedType) throws IOException {
+        return defaultObjectMapper.readerFor(expectedType).readValue(json);
+    }
+
+    public String toJson(Object object) throws IOException {
+        return defaultObjectMapper.writeValueAsString(object);
+    }
+}

mantis-common/src/test/java/io/mantisrx/common/AckTest.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.common;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+
+public class AckTest {
+    private final JsonSerializer serializer = new JsonSerializer();
+
+    @Test
+    public void testAckInstance() throws Exception {
+        Ack ack = Ack.getInstance();
+        String serialized = serializer.toJson(ack);
+        Ack actual = serializer.fromJSON(serialized, Ack.class);
+        assertEquals(ack, actual);
+    }
+}

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/master/resourcecluster/ClusterID.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.server.master.resourcecluster;
+
+import lombok.RequiredArgsConstructor;
+import lombok.Value;
+
+/**
+ * Represents the cluster that the task executor belongs to.
+ */
+@RequiredArgsConstructor(staticName="of")
+@Value
+public class ClusterID {
+  String resourceID;
+}

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/master/resourcecluster/ResourceClusterGateway.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.server.master.resourcecluster;
+
+import io.mantisrx.common.Ack;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Gateway for performing all actions corresponding to the resource cluster from the task executor.
+ */
+public interface ResourceClusterGateway {
+  /**
+   * triggered at the start of a task executor or when the task executor has lost connection to the resource cluster
+   * previously
+   */
+  CompletableFuture<Ack> registerTaskExecutor(TaskExecutorRegistration registration);
+
+  /**
+   * Triggered when the task executor needs to send a heartbeat every epoch after registration.
+   * Absence of a heartbeat from the task executor implies loss of the task executor or a network partition.
+   */
+  CompletableFuture<Ack> heartBeatFromTaskExecutor(TaskExecutorHeartbeat heartbeat);
+
+  /**
+   * Triggered whenever the task executor gets occupied with a worker request or is available to do some work
+   */
+  CompletableFuture<Ack> notifyTaskExecutorStatusChange(TaskExecutorStatusChange taskExecutorStatusChange);
+
+  /**
+   * Triggered by the task executor when it's about to shut itself down.
+   */
+  CompletableFuture<Ack> disconnectTaskExecutor(TaskExecutorDisconnection taskExecutorDisconnection);
+
+  /**
+   * Exception thrown by the resource cluster whenever anyone of the state transitions of the task executor are invalid.
+   * Note that the exception generally gets wrapped inside {@link java.util.concurrent.CompletionException}.
+   */
+  class InvalidStateTransitionException extends Exception {
+
+    public InvalidStateTransitionException(String message) {
+      super(message);
+    }
+  }
+}

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/master/resourcecluster/TaskExecutorDisconnection.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.server.master.resourcecluster;
+
+import lombok.Value;
+
+/**
+ * Datastructure representing the task executor that needs to be disconnected from the resource cluster.
+ */
+@Value
+public class TaskExecutorDisconnection {
+  TaskExecutorID taskExecutorID;
+
+  ClusterID clusterID;
+}

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/master/resourcecluster/TaskExecutorHeartbeat.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.server.master.resourcecluster;
+
+import lombok.Value;
+
+@Value
+public class TaskExecutorHeartbeat {
+  TaskExecutorID taskExecutorID;
+
+  ClusterID clusterID;
+
+  TaskExecutorReport taskExecutorReport;
+}

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/master/resourcecluster/TaskExecutorID.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.server.master.resourcecluster;
+
+import java.util.UUID;
+import lombok.RequiredArgsConstructor;
+import lombok.Value;
+
+/**
+ * ID of the task executor. Reason this is wrapped inside a type is to have strong typing when dealing
+ * with different types of IDs within the system.
+ */
+@RequiredArgsConstructor(staticName="of")
+@Value
+public class TaskExecutorID {
+  String resourceId;
+
+  public static TaskExecutorID generate() {
+    return new TaskExecutorID(UUID.randomUUID().toString());
+  }
+}

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/master/resourcecluster/TaskExecutorRegistration.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2022 Netflix, Inc.
+ *
+ * Licensed 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 io.mantisrx.server.master.resourcecluster;
+
+import io.mantisrx.common.WorkerPorts;
+import io.mantisrx.runtime.MachineDefinition;
+import lombok.Value;
+
+/**
+ * Data structure used at the time of registration by the task executor.
+ * Different fields help identify the task executor in different dimensions.
+ */
+@Value
+public class TaskExecutorRegistration {
+  TaskExecutorID taskExecutorID;
+
+  ClusterID clusterID;
+
+  // RPC address that's used to talk to the task executor
+  String taskExecutorAddress;
+
+  // host name of the task executor
+  String hostname;
+
+  // ports used by the task executor for various purposes.
+  WorkerPorts workerPorts;
+
+  // machine information identifies the cpu/mem/disk/network resources of the task executor.
+  MachineDefinition machineDefinition;
+}