[Part 15]: Resource Cluster Abstraction and an implementation based on Akka actor (#166)

* [Part 15]: Resource Cluster Abstraction and an implementation based on Akka actor

Co-authored-by: Sundaram Ananthanarayanan <[email protected]>

Author: sundargates

mantis-common/build.gradle

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+apply plugin: 'java-test-fixtures'
+
 ext {
     jacksonVersion = '2.10.+'
     jctoolsVersion = '1.+'

mantis-common/src/testFixtures/java/io/mantisrx/common/util/DelegateClock.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.common.util;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+public class DelegateClock extends Clock {
+    private final AtomicReference<Clock> delegate;
+
+    @Override
+    public ZoneId getZone() {
+        return delegate.get().getZone();
+    }
+
+    @Override
+    public Clock withZone(ZoneId zone) {
+        return delegate.get().withZone(zone);
+    }
+
+    @Override
+    public Instant instant() {
+        return delegate.get().instant();
+    }
+}

mantis-control-plane/mantis-control-plane-core/build.gradle

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+apply plugin: 'java-test-fixtures'
 
 ext {
 

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/core/domain/WorkerId.java

@@ -19,6 +19,7 @@
 import io.mantisrx.shaded.com.fasterxml.jackson.annotation.JsonCreator;
 import io.mantisrx.shaded.com.fasterxml.jackson.annotation.JsonIgnore;
 import io.mantisrx.shaded.com.fasterxml.jackson.annotation.JsonProperty;
+import io.mantisrx.shaded.org.apache.curator.shaded.com.google.common.annotations.VisibleForTesting;
 import java.io.Serializable;
 import java.util.Optional;
 import org.slf4j.Logger;
@@ -98,6 +99,11 @@ public static Optional<WorkerId> fromId(final String id) {
         return Optional.empty();
     }
 
+    @VisibleForTesting
+    public static WorkerId fromIdUnsafe(String id) {
+        return fromId(id).get();
+    }
+
     public String getJobCluster() {
         return jobCluster;
     }

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

@@ -0,0 +1,101 @@
+/*
+ * 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 io.mantisrx.runtime.MachineDefinition;
+import io.mantisrx.server.core.domain.WorkerId;
+import io.mantisrx.server.worker.TaskExecutorGateway;
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import javax.annotation.Nullable;
+import lombok.Value;
+
+/**
+ * Abstraction to deal with all interactions with the resource cluster such as
+ * 1). listing the set of task executors registered
+ * 2). listing the set of task executors available
+ * 3). listing the set of task executors busy
+ * 4). get the current state of a task executor
+ * 5). get the current state of the system
+ * 6). assign a task executor for a given worker
+ */
+public interface ResourceCluster extends ResourceClusterGateway {
+    /**
+     * API that gets invoked when the resource cluster migrates from one machine to another and needs to be initialized.
+     *
+     * @param taskExecutorID taskExecutorID that was originally running the worker
+     * @param workerId workerID of the task that being run on the task executor
+     * @return Ack when the initialization is done
+     */
+    CompletableFuture<Ack> initializeTaskExecutor(TaskExecutorID taskExecutorID, WorkerId workerId);
+
+    CompletableFuture<List<TaskExecutorID>> getRegisteredTaskExecutors();
+
+    CompletableFuture<List<TaskExecutorID>> getAvailableTaskExecutors();
+
+    CompletableFuture<List<TaskExecutorID>> getBusyTaskExecutors();
+
+    CompletableFuture<List<TaskExecutorID>> getUnregisteredTaskExecutors();
+
+    CompletableFuture<ResourceOverview> resourceOverview();
+
+    /**
+     * Can throw {@link NoResourceAvailableException} wrapped within the CompletableFuture in case there
+     * are no task executors.
+     *
+     * @param machineDefinition machine definition that's requested for the worker
+     * @param workerId worker id of the task that's going to run on the node.
+     * @return task executor assigned for the particular task.
+     */
+    CompletableFuture<TaskExecutorID> getTaskExecutorFor(MachineDefinition machineDefinition, WorkerId workerId);
+
+    CompletableFuture<TaskExecutorGateway> getTaskExecutorGateway(TaskExecutorID taskExecutorID);
+
+    CompletableFuture<TaskExecutorRegistration> getTaskExecutorInfo(String hostName);
+
+    CompletableFuture<TaskExecutorRegistration> getTaskExecutorInfo(TaskExecutorID taskExecutorID);
+
+    CompletableFuture<TaskExecutorStatus> getTaskExecutorState(TaskExecutorID taskExecutorID);
+
+    class NoResourceAvailableException extends Exception {
+
+        public NoResourceAvailableException(String message) {
+            super(message);
+        }
+    }
+
+    @Value
+    class ResourceOverview {
+        long numRegisteredTaskExecutors;
+        long numAvailableTaskExecutors;
+        long numOccupiedTaskExecutors;
+        long numAssignedTaskExecutors;
+    }
+
+    @Value
+    class TaskExecutorStatus {
+        TaskExecutorRegistration registration;
+        boolean isRegistered;
+        boolean isRunningTask;
+        boolean isAssignedTask;
+        @Nullable
+        WorkerId workerId;
+        Instant lastHeartbeat;
+    }
+}

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

@@ -0,0 +1,49 @@
+/*
+ * 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.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import javax.annotation.Nullable;
+
+/**
+ * This is a helper abstraction to find cluster ids for task executors.
+ * For the foreseeable future, we expect only one implementation of this particular class.
+ */
+public interface ResourceClusterTaskExecutorMapper {
+    @Nullable
+    ClusterID getClusterFor(TaskExecutorID taskExecutorID);
+
+    void onTaskExecutorDiscovered(ClusterID clusterID, TaskExecutorID taskExecutorID);
+
+    static ResourceClusterTaskExecutorMapper inMemory() {
+        return new ResourceClusterTaskExecutorMapper() {
+            private final ConcurrentMap<TaskExecutorID, ClusterID> map =
+                new ConcurrentHashMap<>();
+
+            @Override
+            public ClusterID getClusterFor(TaskExecutorID taskExecutorID) {
+                return map.get(taskExecutorID);
+            }
+
+            @Override
+            public void onTaskExecutorDiscovered(ClusterID clusterID, TaskExecutorID taskExecutorID) {
+                map.putIfAbsent(taskExecutorID, clusterID);
+            }
+        };
+    }
+}

mantis-control-plane/mantis-control-plane-core/src/main/java/io/mantisrx/server/master/resourcecluster/ResourceClusters.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 java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * ResourceClusters is a factory class for getting and managing individual resource clusters.
+ */
+public interface ResourceClusters {
+  ResourceCluster getClusterFor(ClusterID clusterID);
+
+  CompletableFuture<Set<ClusterID>> listActiveClusters();
+}

mantis-server/mantis-server-worker/src/test/java/io/mantisrx/server/worker/TestingRpcService.java renamed to mantis-control-plane/mantis-control-plane-core/src/testFixtures/java/io/mantisrx/server/core/TestingRpcService.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.mantisrx.server.worker;
+package io.mantisrx.server.core;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 

mantis-control-plane/mantis-control-plane-server/build.gradle

@@ -48,6 +48,8 @@ dependencies {
     testCompile libraries.junit4
     testCompile libraries.mockitoAll
     testCompile "org.testng:testng:$testngVersion"
+    testImplementation testFixtures(project(":mantis-common"))
+    testImplementation testFixtures(project(":mantis-control-plane:mantis-control-plane-core"))
 }
 
 mainClassName = "io.mantisrx.server.master.MasterMain"