Resource Cluster Manager Actor

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

@@ -0,0 +1,91 @@
+/*
+ * 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.master.resourcecluster;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import io.mantisrx.server.master.config.ConfigurationProvider;
+import io.mantisrx.server.master.config.MasterConfiguration;
+import io.mantisrx.server.master.persistence.MantisJobStore;
+import io.mantisrx.server.master.resourcecluster.ClusterID;
+import io.mantisrx.server.master.resourcecluster.ResourceCluster;
+import io.mantisrx.server.master.resourcecluster.ResourceClusterTaskExecutorMapper;
+import io.mantisrx.server.master.resourcecluster.ResourceClusters;
+import java.time.Clock;
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import lombok.AccessLevel;
+import lombok.RequiredArgsConstructor;
+import org.apache.flink.runtime.rpc.RpcService;
+
+/**
+ * This class is an implementation of {@link ResourceClusters} that uses the Akka actor implementation under the hood.
+ * You can think of this class as a more java typed-way of sharing the functionalities of the akka actor.
+ */
+@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
+public class ResourceClustersAkkaImpl implements ResourceClusters {
+
+    private final ActorRef resourceClustersManagerActor;
+    private final Duration askTimeout;
+    private final ResourceClusterTaskExecutorMapper mapper;
+    private final ConcurrentMap<ClusterID, ResourceCluster> cache =
+        new ConcurrentHashMap<>();
+
+    @Override
+    public ResourceCluster getClusterFor(ClusterID clusterID) {
+        cache.computeIfAbsent(
+            clusterID,
+            dontCare ->
+                new ResourceClusterAkkaImpl(
+                    resourceClustersManagerActor,
+                    askTimeout,
+                    clusterID,
+                    mapper));
+        return cache.get(clusterID);
+    }
+
+    @Override
+    public CompletableFuture<Set<ClusterID>> listActiveClusters() {
+        return
+            Patterns.ask(resourceClustersManagerActor,
+                    new ResourceClustersManagerActor.ListActiveClusters(), askTimeout)
+                .toCompletableFuture()
+                .thenApply(ResourceClustersManagerActor.ClusterIdSet.class::cast)
+                .thenApply(clusterIdSet -> clusterIdSet.getClusterIDS());
+    }
+
+    public static ResourceClusters load(
+        MasterConfiguration masterConfiguration,
+        RpcService rpcService,
+        ActorSystem actorSystem,
+        MantisJobStore mantisJobStore) {
+        final ActorRef resourceClusterManagerActor =
+            actorSystem.actorOf(
+                ResourceClustersManagerActor.props(masterConfiguration, Clock.systemDefaultZone(),
+                    rpcService, mantisJobStore));
+        final ResourceClusterTaskExecutorMapper globalMapper =
+            ResourceClusterTaskExecutorMapper.inMemory();
+
+        final Duration askTimeout = java.time.Duration.ofMillis(
+            ConfigurationProvider.getConfig().getMasterApiAskTimeoutMs());
+        return new ResourceClustersAkkaImpl(resourceClusterManagerActor, askTimeout, globalMapper);
+    }
+}

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

@@ -0,0 +1,156 @@
+/*
+ * 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.master.resourcecluster;
+
+import akka.actor.AbstractActor;
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import akka.actor.SupervisorStrategy;
+import akka.japi.pf.ReceiveBuilder;
+import io.mantisrx.master.akka.MantisActorSupervisorStrategy;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.GetAvailableTaskExecutorsRequest;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.GetBusyTaskExecutorsRequest;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.GetRegisteredTaskExecutorsRequest;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.GetTaskExecutorStatusRequest;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.GetUnregisteredTaskExecutorsRequest;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.ResourceOverviewRequest;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.TaskExecutorAssignmentRequest;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.TaskExecutorGatewayRequest;
+import io.mantisrx.master.resourcecluster.ResourceClusterActor.TaskExecutorInfoRequest;
+import io.mantisrx.server.master.config.MasterConfiguration;
+import io.mantisrx.server.master.persistence.MantisJobStore;
+import io.mantisrx.server.master.resourcecluster.ClusterID;
+import io.mantisrx.server.master.resourcecluster.TaskExecutorDisconnection;
+import io.mantisrx.server.master.resourcecluster.TaskExecutorHeartbeat;
+import io.mantisrx.server.master.resourcecluster.TaskExecutorRegistration;
+import io.mantisrx.server.master.resourcecluster.TaskExecutorStatusChange;
+import java.time.Clock;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import lombok.Value;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.runtime.rpc.RpcService;
+
+/**
+ * Supervisor actor responsible for creating/deleting/listing all resource clusters in the system.
+ */
+@Slf4j
+class ResourceClustersManagerActor extends AbstractActor {
+
+    private final MasterConfiguration masterConfiguration;
+    private final Clock clock;
+    private final RpcService rpcService;
+    private final MantisJobStore mantisJobStore;
+
+    private final Map<ClusterID, ActorRef> resourceClusterActorMap;
+
+    public static Props props(MasterConfiguration masterConfiguration, Clock clock, RpcService rpcService, MantisJobStore mantisJobStore) {
+        return Props.create(ResourceClustersManagerActor.class, masterConfiguration, clock, rpcService, mantisJobStore);
+    }
+
+    public ResourceClustersManagerActor(
+        MasterConfiguration masterConfiguration, Clock clock,
+        RpcService rpcService,
+        MantisJobStore mantisJobStore) {
+        this.masterConfiguration = masterConfiguration;
+        this.clock = clock;
+        this.rpcService = rpcService;
+        this.mantisJobStore = mantisJobStore;
+
+        this.resourceClusterActorMap = new HashMap<>();
+    }
+
+    @Override
+    public Receive createReceive() {
+        return
+            ReceiveBuilder
+                .create()
+                .match(ListActiveClusters.class, req -> sender().tell(getActiveClusters(), self()))
+
+                .match(GetRegisteredTaskExecutorsRequest.class, req -> getRCActor(req.getClusterID()).forward(req, context()))
+                .match(GetBusyTaskExecutorsRequest.class, req -> getRCActor(req.getClusterID()).forward(req, context()))
+                .match(GetAvailableTaskExecutorsRequest.class, req -> getRCActor(req.getClusterID()).forward(req, context()))
+                .match(GetUnregisteredTaskExecutorsRequest.class, req -> getRCActor(req.getClusterID()).forward(req, context()))
+                .match(GetTaskExecutorStatusRequest.class, req -> getRCActor(req.getClusterID()).forward(req, context()))
+
+                .match(TaskExecutorRegistration.class, registration ->
+                    getRCActor(registration.getClusterID()).forward(registration, context()))
+                .match(TaskExecutorHeartbeat.class, heartbeat ->
+                    getRCActor(heartbeat.getClusterID()).forward(heartbeat, context()))
+                .match(TaskExecutorStatusChange.class, statusChange ->
+                    getRCActor(statusChange.getClusterID()).forward(statusChange, context()))
+                .match(TaskExecutorDisconnection.class, disconnection ->
+                    getRCActor(disconnection.getClusterID()).forward(disconnection, context()))
+                .match(TaskExecutorAssignmentRequest.class, req ->
+                    getRCActor(req.getClusterID()).forward(req, context()))
+                .match(ResourceOverviewRequest.class, req ->
+                    getRCActor(req.getClusterID()).forward(req, context()))
+                .match(TaskExecutorInfoRequest.class, req ->
+                    getRCActor(req.getClusterID()).forward(req, context()))
+                .match(TaskExecutorGatewayRequest.class, req ->
+                    getRCActor(req.getClusterID()).forward(req, context()))
+                .build();
+    }
+
+    private ActorRef createResourceClusterActorFor(ClusterID clusterID) {
+        log.info("Creating resource cluster actor for {}", clusterID);
+        ActorRef clusterActor =
+            getContext().actorOf(
+                ResourceClusterActor.props(
+                    clusterID,
+                    Duration.ofMillis(masterConfiguration.getHeartbeatIntervalInMs()),
+                    Duration.ofMillis(masterConfiguration.getAssignmentIntervalInMs()),
+                    clock,
+                    rpcService,
+                    mantisJobStore),
+                "ResourceClusterActor-" + clusterID.getResourceID());
+        log.info("Created resource cluster actor for {}", clusterID);
+        return clusterActor;
+    }
+
+    private ActorRef getRCActor(ClusterID clusterID) {
+        if (resourceClusterActorMap.get(clusterID) != null) {
+            return resourceClusterActorMap.get(clusterID);
+        } else {
+            return resourceClusterActorMap.computeIfAbsent(clusterID, (dontCare) -> {
+                ActorRef actorRef = createResourceClusterActorFor(clusterID);
+                getContext().watch(actorRef);
+                return actorRef;
+            });
+        }
+    }
+
+    private ClusterIdSet getActiveClusters() {
+        return new ClusterIdSet(resourceClusterActorMap.keySet());
+    }
+
+    @Value
+    static class ListActiveClusters {
+    }
+
+    @Value
+    static class ClusterIdSet {
+        Set<ClusterID> clusterIDS;
+    }
+
+    @Override
+    public SupervisorStrategy supervisorStrategy() {
+        return MantisActorSupervisorStrategy.getInstance().create();
+    }
+}

mantis-control-plane/mantis-control-plane-server/src/main/java/io/mantisrx/server/master/config/MasterConfiguration.java

@@ -18,6 +18,7 @@
 
 import io.mantisrx.server.core.CoreConfiguration;
 import io.mantisrx.server.master.store.MantisStorageProvider;
+import java.time.Duration;
 import org.skife.config.Config;
 import org.skife.config.Default;
 import org.skife.config.DefaultNull;
@@ -302,4 +303,20 @@ public interface MasterConfiguration extends CoreConfiguration {
     @Config("mantis.master.api.cache.size.min")
     @Default("5")
     int getApiCacheMinSize();
-}
+
+    @Config("mantis.agent.heartbeat.interval.ms")
+    @Default("300000") // 5 minutes
+    int getHeartbeatIntervalInMs();
+
+    @Config("mantis.agent.assignment.interval.ms")
+    @Default("60000") // 1 minute
+    int getAssignmentIntervalInMs();
+
+    default Duration getHeartbeatInterval() {
+        return Duration.ofMillis(getHeartbeatIntervalInMs());
+    }
+
+    default Duration getMaxAssignmentThreshold() {
+        return Duration.ofMillis(getAssignmentIntervalInMs());
+    }
+}