Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ dependencies {
api "org.skife.config:config-magic:$configMagicVersion"

testCompile "com.typesafe.akka:akka-testkit_$scalaBinaryVersion:$akkaVersion"
testCompile "com.typesafe.akka:akka-http-testkit_$scalaBinaryVersion:$akkaHttpVersion"
testCompile libraries.junit4
testCompile libraries.mockitoCore3
testCompile "org.testng:testng:$testngVersion"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@
import io.mantisrx.server.master.ILeadershipManager;
import io.mantisrx.server.master.LeaderRedirectionFilter;
import io.mantisrx.server.master.persistence.IMantisStorageProvider;
import io.mantisrx.server.master.resourcecluster.ResourceClusters;
import io.mantisrx.server.master.scheduler.MantisScheduler;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CountDownLatch;
Expand All @@ -75,6 +76,7 @@ public class MasterApiAkkaService extends BaseService {
private final MasterMonitor masterMonitor;
private final MasterDescription masterDescription;
private final ActorRef jobClustersManagerActor;
private final ResourceClusters resourceClusters;
private final ActorRef statusEventBrokerActor;
private final int port;
private final IMantisStorageProvider storageProvider;
Expand All @@ -91,6 +93,7 @@ public MasterApiAkkaService(final MasterMonitor masterMonitor,
final MasterDescription masterDescription,
final ActorRef jobClustersManagerActor,
final ActorRef statusEventBrokerActor,
final ResourceClusters resourceClusters,
final int serverPort,
final IMantisStorageProvider mantisStorageProvider,
final MantisScheduler scheduler,
Expand All @@ -111,6 +114,7 @@ public MasterApiAkkaService(final MasterMonitor masterMonitor,
this.masterDescription = masterDescription;
this.jobClustersManagerActor = jobClustersManagerActor;
this.statusEventBrokerActor = statusEventBrokerActor;
this.resourceClusters = resourceClusters;
this.port = serverPort;
this.storageProvider = mantisStorageProvider;
this.scheduler = scheduler;
Expand Down Expand Up @@ -162,19 +166,20 @@ private MantisMasterRoute configureApiRoutes(final ActorSystem actorSystem, fina

final LeaderRedirectionFilter leaderRedirectionFilter = new LeaderRedirectionFilter(masterMonitor, leadershipManager);
return new MantisMasterRoute(leaderRedirectionFilter,
masterDescriptionRoute,
v0JobClusterRoute,
v0JobRoute,
v0JobDiscoveryRoute,
v0JobStatusRoute,
v0AgentClusterRoute,
v1JobClusterRoute,
v1JobsRoute,
v1AdminMasterRoute,
v1AgentClustersRoute,
v1JobDiscoveryStreamRoute,
v1LastSubmittedJobIdStreamRoute,
v1JobStatusStreamRoute);
masterDescriptionRoute,
v0JobClusterRoute,
v0JobRoute,
v0JobDiscoveryRoute,
v0JobStatusRoute,
v0AgentClusterRoute,
v1JobClusterRoute,
v1JobsRoute,
v1AdminMasterRoute,
v1AgentClustersRoute,
v1JobDiscoveryStreamRoute,
v1LastSubmittedJobIdStreamRoute,
v1JobStatusStreamRoute,
resourceClusters);
}

private void startAPIServer() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,12 +31,12 @@
import io.mantisrx.master.api.akka.route.v1.JobStatusStreamRoute;
import io.mantisrx.master.api.akka.route.v1.JobsRoute;
import io.mantisrx.master.api.akka.route.v1.LastSubmittedJobIdStreamRoute;
import io.mantisrx.master.api.akka.route.v1.ResourceClustersReadRoute;
import io.mantisrx.master.api.akka.route.v1.ResourceClustersWriteRoute;
import io.mantisrx.server.master.LeaderRedirectionFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.mantisrx.server.master.resourcecluster.ResourceClusters;

public class MantisMasterRoute extends AllDirectives {
private static final Logger logger = LoggerFactory.getLogger(MantisMasterRoute.class);

private final LeaderRedirectionFilter leaderRedirectionFilter;

Expand All @@ -55,22 +55,25 @@ public class MantisMasterRoute extends AllDirectives {
private final JobDiscoveryStreamRoute v1JobDiscoveryStreamRoute;
private final LastSubmittedJobIdStreamRoute v1LastSubmittedJobIdStreamRoute;
private final JobStatusStreamRoute v1JobStatusStreamRoute;
private final ResourceClustersReadRoute resourceClustersReadRoute;
private final ResourceClustersWriteRoute resourceClustersWriteRoute;

public MantisMasterRoute(
final LeaderRedirectionFilter leaderRedirectionFilter,
final MasterDescriptionRoute v0MasterDescriptionRoute,
final JobClusterRoute v0JobClusterRoute,
final JobRoute v0JobRoute,
final JobDiscoveryRoute v0JobDiscoveryRoute,
final JobStatusRoute v0JobStatusRoute,
final AgentClusterRoute v0AgentClusterRoute,
final JobClustersRoute v1JobClusterRoute,
final JobsRoute v1JobsRoute,
final AdminMasterRoute v1MasterRoute,
final AgentClustersRoute v1AgentClustersRoute,
final JobDiscoveryStreamRoute v1JobDiscoveryStreamRoute,
final LastSubmittedJobIdStreamRoute v1LastSubmittedJobIdStreamRoute,
final JobStatusStreamRoute v1JobStatusStreamRoute) {
final LeaderRedirectionFilter leaderRedirectionFilter,
final MasterDescriptionRoute v0MasterDescriptionRoute,
final JobClusterRoute v0JobClusterRoute,
final JobRoute v0JobRoute,
final JobDiscoveryRoute v0JobDiscoveryRoute,
final JobStatusRoute v0JobStatusRoute,
final AgentClusterRoute v0AgentClusterRoute,
final JobClustersRoute v1JobClusterRoute,
final JobsRoute v1JobsRoute,
final AdminMasterRoute v1MasterRoute,
final AgentClustersRoute v1AgentClustersRoute,
final JobDiscoveryStreamRoute v1JobDiscoveryStreamRoute,
final LastSubmittedJobIdStreamRoute v1LastSubmittedJobIdStreamRoute,
final JobStatusStreamRoute v1JobStatusStreamRoute,
final ResourceClusters resourceClusters) {
this.leaderRedirectionFilter = leaderRedirectionFilter;
this.v0MasterDescriptionRoute = v0MasterDescriptionRoute;
this.v0JobClusterRoute = v0JobClusterRoute;
Expand All @@ -86,6 +89,8 @@ public MantisMasterRoute(
this.v1JobDiscoveryStreamRoute = v1JobDiscoveryStreamRoute;
this.v1LastSubmittedJobIdStreamRoute = v1LastSubmittedJobIdStreamRoute;
this.v1JobStatusStreamRoute = v1JobStatusStreamRoute;
this.resourceClustersReadRoute = new ResourceClustersReadRoute(resourceClusters);
this.resourceClustersWriteRoute = new ResourceClustersWriteRoute(resourceClusters);
}

public Route createRoute() {
Expand All @@ -102,7 +107,9 @@ public Route createRoute() {
v1AgentClustersRoute.createRoute(leaderRedirectionFilter::redirectIfNotLeader),
v1JobDiscoveryStreamRoute.createRoute(leaderRedirectionFilter::redirectIfNotLeader),
v1LastSubmittedJobIdStreamRoute.createRoute(leaderRedirectionFilter::redirectIfNotLeader),
v1JobStatusStreamRoute.createRoute(leaderRedirectionFilter::redirectIfNotLeader)
v1JobStatusStreamRoute.createRoute(leaderRedirectionFilter::redirectIfNotLeader),
resourceClustersReadRoute.createRoute(leaderRedirectionFilter::redirectIfNotLeader),
resourceClustersWriteRoute.createRoute(leaderRedirectionFilter::rejectIfNotLeader)
);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import akka.japi.pf.PFBuilder;
import akka.pattern.AskTimeoutException;
import com.netflix.spectator.api.BasicTag;
import io.mantisrx.master.api.akka.route.Jackson;
import io.mantisrx.master.api.akka.route.MasterApiMetrics;
import io.mantisrx.master.jobcluster.proto.BaseResponse;
import io.mantisrx.shaded.com.fasterxml.jackson.databind.node.JsonNodeFactory;
Expand All @@ -53,6 +54,7 @@
import java.util.Arrays;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
Expand Down Expand Up @@ -109,7 +111,7 @@ protected Cache<Uri, RouteResult> createCache(ActorSystem actorSystem, int initi

protected abstract Route constructRoutes();

protected Route createRoute(Function<Route, Route> routeFilter) {
public Route createRoute(Function<Route, Route> routeFilter) {

final ExceptionHandler jsonExceptionHandler = ExceptionHandler
.newBuilder()
Expand Down Expand Up @@ -316,4 +318,11 @@ Boolean parseBoolean(String val) {
return Boolean.valueOf(val);
}
}

protected <T> Route withFuture(CompletableFuture<T> tFuture) {
return onComplete(tFuture,
t -> t.fold(
throwable -> complete(StatusCodes.INTERNAL_SERVER_ERROR, throwable, Jackson.marshaller()),
r -> complete(StatusCodes.OK, r, Jackson.marshaller())));
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
/*
* 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.api.akka.route.v1;

import static akka.http.javadsl.server.PathMatchers.segment;

import akka.http.javadsl.server.PathMatcher0;
import akka.http.javadsl.server.PathMatchers;
import akka.http.javadsl.server.Route;
import io.mantisrx.server.master.resourcecluster.ClusterID;
import io.mantisrx.server.master.resourcecluster.ResourceCluster;
import io.mantisrx.server.master.resourcecluster.ResourceClusters;
import io.mantisrx.server.master.resourcecluster.TaskExecutorID;
import java.util.concurrent.CompletableFuture;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
* Resource Cluster Route
* Defines the following end points:
* /api/v1/resourceClusters/list (GET)
* <p>
* /api/v1/resourceClusters/{}/getResourceOverview (GET)
* /api/v1/resourceClusters/{}/getRegisteredTaskExecutors (GET)
* /api/v1/resourceClusters/{}/getBusyTaskExecutors (GET)
* /api/v1/resourceClusters/{}/getAvailableTaskExecutors (GET)
* /api/v1/resourceClusters/{}/getUnregisteredTaskExecutors (GET)
* <p>
* /api/v1/resourceClusters/{}/taskExecutors/{}/getTaskExecutorState (GET)
*/
@Slf4j
@RequiredArgsConstructor
public class ResourceClustersReadRoute extends BaseRoute {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe add some test coverage for these?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done.


private static final PathMatcher0 RESOURCECLUSTERS_API_PREFIX =
segment("api").slash("v1").slash("resourceClusters");

private final ResourceClusters gateway;

@Override
protected Route constructRoutes() {
Route result = pathPrefix(
RESOURCECLUSTERS_API_PREFIX,
() -> concat(
// /list
pathPrefix(
"list",
() -> concat(
// GET
get(this::listClusters))
),
// /{}/getResourceOverview
path(
PathMatchers.segment().slash("getResourceOverview"),
(clusterName) -> pathEndOrSingleSlash(() -> concat(get(() -> getResourceOverview(getClusterID(clusterName)))))
),
// /{}/getRegisteredTaskExecutors
path(
PathMatchers.segment().slash("getRegisteredTaskExecutors"),
(clusterName) -> pathEndOrSingleSlash(() -> concat(get(() -> withFuture(gateway.getClusterFor(getClusterID(clusterName)).getRegisteredTaskExecutors()))))
),
// /{}/getBusyTaskExecutors
path(
PathMatchers.segment().slash("getBusyTaskExecutors"),
(clusterName) -> pathEndOrSingleSlash(() -> concat(get(() -> withFuture(gateway.getClusterFor(getClusterID(clusterName)).getBusyTaskExecutors()))))
),
// /{}/getAvailableTaskExecutors
path(
PathMatchers.segment().slash("getAvailableTaskExecutors"),
(clusterName) -> pathEndOrSingleSlash(() -> concat(get(() -> withFuture(gateway.getClusterFor(getClusterID(clusterName)).getAvailableTaskExecutors()))))
),
// /{}/getUnregisteredTaskExecutors
path(
PathMatchers.segment().slash("getUnregisteredTaskExecutors"),
(clusterName) -> pathEndOrSingleSlash(() -> concat(get(() -> withFuture(gateway.getClusterFor(getClusterID(clusterName)).getUnregisteredTaskExecutors()))))
),

// /api/v1/resourceClusters/{}/taskExecutors/{}/getTaskExecutorState
pathPrefix(
PathMatchers.segment().slash("taskExecutors"),
(clusterName) ->
path(
PathMatchers.segment().slash("getTaskExecutorState"),
(taskExecutorId) ->
pathEndOrSingleSlash(() -> concat(get(() -> getTaskExecutorState(getClusterID(clusterName), getTaskExecutorID(taskExecutorId))))))
)
));

return result;
}

private Route listClusters() {
return withFuture(gateway.listActiveClusters());
}

private Route getResourceOverview(ClusterID clusterID) {
CompletableFuture<ResourceCluster.ResourceOverview> resourceOverview =
gateway.getClusterFor(clusterID).resourceOverview();
return withFuture(resourceOverview);
}

private Route getTaskExecutorState(ClusterID clusterID, TaskExecutorID taskExecutorID) {
CompletableFuture<ResourceCluster.TaskExecutorStatus> statusOverview =
gateway.getClusterFor(clusterID).getTaskExecutorState(taskExecutorID);
return withFuture(statusOverview);
}

private ClusterID getClusterID(String clusterName) {
return ClusterID.of(clusterName);
}

private TaskExecutorID getTaskExecutorID(String resourceName) {
return TaskExecutorID.of(resourceName);
}
}
Loading