Skip to content

Commit f82a59e

Browse files
narendlyJiajun Wang
narendly
authored and
Jiajun Wang
committed
Add REST API endpoints for WAGED Rebalancer (#611)
We want to make WAGED rebalancer (weight-aware) easier to use. One way to do this is to allow the user to easily add resources with weight configuration set by providing REST endpoints. This change adds the relevant REST endpoints based on the HelixAdmin APIs added in (#570). Basically, this commit uses existing REST endpoints whose hierarchy is defined by REST resource. What this commit does to the existing endpoints is 1) Add extra commands 2) Add a WAGED command as a QueryParam so that WAGED logic could be included. This change is backward-compatible because it keeps the original behavior when no commands are provided by using @DefaultValue annotation.
1 parent 5da3a19 commit f82a59e

11 files changed

Lines changed: 490 additions & 74 deletions

File tree

‎helix-rest/src/main/java/org/apache/helix/rest/server/resources/AbstractResource.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,15 @@ public enum Command {
6767
rebalance,
6868
reset,
6969
resetPartitions,
70-
removeInstanceTag
70+
removeInstanceTag,
71+
addResource,
72+
addWagedResource,
73+
getResource,
74+
validateWeight,
75+
enableWagedRebalance,
76+
enableWagedRebalanceForAllResources,
77+
getInstance,
78+
getAllInstances
7179
}
7280

7381
@Context

‎helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -238,7 +238,15 @@ public Response updateCluster(@PathParam("clusterId") String clusterId,
238238
helixAdmin.manuallyEnableMaintenanceMode(clusterId, command == Command.enableMaintenanceMode,
239239
content, customFieldsMap);
240240
break;
241-
241+
case enableWagedRebalanceForAllResources:
242+
// Enable WAGED rebalance for all resources in the cluster
243+
List<String> resources = helixAdmin.getResourcesInCluster(clusterId);
244+
try {
245+
helixAdmin.enableWagedRebalance(clusterId, resources);
246+
} catch (HelixException e) {
247+
return badRequest(e.getMessage());
248+
}
249+
break;
242250
default:
243251
return badRequest("Unsupported command " + command);
244252
}

‎helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/InstancesAccessor.java‎

Lines changed: 66 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,24 @@
11
package org.apache.helix.rest.server.resources.helix;
22

3+
/*
4+
* Licensed to the Apache Software Foundation (ASF) under one
5+
* or more contributor license agreements. See the NOTICE file
6+
* distributed with this work for additional information
7+
* regarding copyright ownership. The ASF licenses this file
8+
* to you under the Apache License, Version 2.0 (the
9+
* "License"); you may not use this file except in compliance
10+
* with the License. You may obtain a copy of the License at
11+
*
12+
* http://www.apache.org/licenses/LICENSE-2.0
13+
*
14+
* Unless required by applicable law or agreed to in writing,
15+
* software distributed under the License is distributed on an
16+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
17+
* KIND, either express or implied. See the License for the
18+
* specific language governing permissions and limitations
19+
* under the License.
20+
*/
21+
322
import java.io.IOException;
423
import java.util.ArrayList;
524
import java.util.Collections;
@@ -8,6 +27,7 @@
827
import java.util.Map;
928
import java.util.Set;
1029
import java.util.TreeSet;
30+
import javax.ws.rs.DefaultValue;
1131
import javax.ws.rs.GET;
1232
import javax.ws.rs.POST;
1333
import javax.ws.rs.Path;
@@ -61,41 +81,65 @@ public enum InstanceHealthSelectionBase {
6181
}
6282

6383
@GET
64-
public Response getAllInstances(@PathParam("clusterId") String clusterId) {
84+
public Response getAllInstances(@PathParam("clusterId") String clusterId,
85+
@DefaultValue("getAllInstances") @QueryParam("command") String command) {
86+
// Get the command. If not provided, the default would be "getAllInstances"
87+
Command cmd;
88+
try {
89+
cmd = Command.valueOf(command);
90+
} catch (Exception e) {
91+
return badRequest("Invalid command : " + command);
92+
}
93+
6594
HelixDataAccessor accessor = getDataAccssor(clusterId);
6695
List<String> instances = accessor.getChildNames(accessor.keyBuilder().instanceConfigs());
67-
6896
if (instances == null) {
6997
return notFound();
7098
}
7199

72-
ObjectNode root = JsonNodeFactory.instance.objectNode();
73-
root.put(Properties.id.name(), JsonNodeFactory.instance.textNode(clusterId));
100+
switch (cmd) {
101+
case getAllInstances:
102+
ObjectNode root = JsonNodeFactory.instance.objectNode();
103+
root.put(Properties.id.name(), JsonNodeFactory.instance.textNode(clusterId));
74104

75-
ArrayNode instancesNode = root.putArray(InstancesAccessor.InstancesProperties.instances.name());
76-
instancesNode.addAll((ArrayNode) OBJECT_MAPPER.valueToTree(instances));
77-
ArrayNode onlineNode = root.putArray(InstancesAccessor.InstancesProperties.online.name());
78-
ArrayNode disabledNode = root.putArray(InstancesAccessor.InstancesProperties.disabled.name());
105+
ArrayNode instancesNode =
106+
root.putArray(InstancesAccessor.InstancesProperties.instances.name());
107+
instancesNode.addAll((ArrayNode) OBJECT_MAPPER.valueToTree(instances));
108+
ArrayNode onlineNode = root.putArray(InstancesAccessor.InstancesProperties.online.name());
109+
ArrayNode disabledNode = root.putArray(InstancesAccessor.InstancesProperties.disabled.name());
79110

80-
List<String> liveInstances = accessor.getChildNames(accessor.keyBuilder().liveInstances());
81-
ClusterConfig clusterConfig = accessor.getProperty(accessor.keyBuilder().clusterConfig());
111+
List<String> liveInstances = accessor.getChildNames(accessor.keyBuilder().liveInstances());
112+
ClusterConfig clusterConfig = accessor.getProperty(accessor.keyBuilder().clusterConfig());
82113

83-
for (String instanceName : instances) {
84-
InstanceConfig instanceConfig =
85-
accessor.getProperty(accessor.keyBuilder().instanceConfig(instanceName));
86-
if (instanceConfig != null) {
87-
if (!instanceConfig.getInstanceEnabled() || (clusterConfig.getDisabledInstances() != null
88-
&& clusterConfig.getDisabledInstances().containsKey(instanceName))) {
89-
disabledNode.add(JsonNodeFactory.instance.textNode(instanceName));
90-
}
114+
for (String instanceName : instances) {
115+
InstanceConfig instanceConfig =
116+
accessor.getProperty(accessor.keyBuilder().instanceConfig(instanceName));
117+
if (instanceConfig != null) {
118+
if (!instanceConfig.getInstanceEnabled() || (clusterConfig.getDisabledInstances() != null
119+
&& clusterConfig.getDisabledInstances().containsKey(instanceName))) {
120+
disabledNode.add(JsonNodeFactory.instance.textNode(instanceName));
121+
}
91122

92-
if (liveInstances.contains(instanceName)){
93-
onlineNode.add(JsonNodeFactory.instance.textNode(instanceName));
123+
if (liveInstances.contains(instanceName)) {
124+
onlineNode.add(JsonNodeFactory.instance.textNode(instanceName));
125+
}
94126
}
95127
}
128+
return JSONRepresentation(root);
129+
case validateWeight:
130+
// Validate all instances for WAGED rebalance
131+
HelixAdmin admin = getHelixAdmin();
132+
Map<String, Boolean> validationResultMap;
133+
try {
134+
validationResultMap = admin.validateInstancesForWagedRebalance(clusterId, instances);
135+
} catch (HelixException e) {
136+
return badRequest(e.getMessage());
137+
}
138+
return JSONRepresentation(validationResultMap);
139+
default:
140+
_logger.error("Unsupported command :" + command);
141+
return badRequest("Unsupported command :" + command);
96142
}
97-
98-
return JSONRepresentation(root);
99143
}
100144

101145
@POST

‎helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/PerInstanceAccessor.java‎

Lines changed: 39 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,12 @@
2020
*/
2121

2222
import java.io.IOException;
23+
import java.util.Collections;
2324
import java.util.List;
25+
import java.util.Map;
2426
import javax.ws.rs.Consumes;
2527
import javax.ws.rs.DELETE;
28+
import javax.ws.rs.DefaultValue;
2629
import javax.ws.rs.GET;
2730
import javax.ws.rs.POST;
2831
import javax.ws.rs.PUT;
@@ -81,16 +84,41 @@ public enum PerInstanceProperties {
8184

8285
@GET
8386
public Response getInstanceById(@PathParam("clusterId") String clusterId,
84-
@PathParam("instanceName") String instanceName) throws IOException {
85-
ObjectMapper objectMapper = new ObjectMapper();
86-
HelixDataAccessor dataAccessor = getDataAccssor(clusterId);
87-
// TODO reduce GC by dependency injection
88-
InstanceService instanceService =
89-
new InstanceServiceImpl(new HelixDataAccessorWrapper((ZKHelixDataAccessor) dataAccessor), getConfigAccessor());
90-
InstanceInfo instanceInfo = instanceService.getInstanceInfo(clusterId, instanceName,
91-
InstanceService.HealthCheck.STARTED_AND_HEALTH_CHECK_LIST);
87+
@PathParam("instanceName") String instanceName,
88+
@DefaultValue("getInstance") @QueryParam("command") String command) throws IOException {
89+
// Get the command. If not provided, the default would be "getInstance"
90+
Command cmd;
91+
try {
92+
cmd = Command.valueOf(command);
93+
} catch (Exception e) {
94+
return badRequest("Invalid command : " + command);
95+
}
9296

93-
return OK(objectMapper.writeValueAsString(instanceInfo));
97+
switch (cmd) {
98+
case getInstance:
99+
ObjectMapper objectMapper = new ObjectMapper();
100+
HelixDataAccessor dataAccessor = getDataAccssor(clusterId);
101+
// TODO reduce GC by dependency injection
102+
InstanceService instanceService = new InstanceServiceImpl(
103+
new HelixDataAccessorWrapper((ZKHelixDataAccessor) dataAccessor), getConfigAccessor());
104+
InstanceInfo instanceInfo = instanceService.getInstanceInfo(clusterId, instanceName,
105+
InstanceService.HealthCheck.STARTED_AND_HEALTH_CHECK_LIST);
106+
return OK(objectMapper.writeValueAsString(instanceInfo));
107+
case validateWeight:
108+
// Validates instanceConfig for WAGED rebalance
109+
HelixAdmin admin = getHelixAdmin();
110+
Map<String, Boolean> validationResultMap;
111+
try {
112+
validationResultMap = admin.validateInstancesForWagedRebalance(clusterId,
113+
Collections.singletonList(instanceName));
114+
} catch (HelixException e) {
115+
return badRequest(e.getMessage());
116+
}
117+
return JSONRepresentation(validationResultMap);
118+
default:
119+
LOG.error("Unsupported command :" + command);
120+
return badRequest("Unsupported command :" + command);
121+
}
94122
}
95123

96124
@POST
@@ -345,7 +373,8 @@ public Response getResourceOnInstance(@PathParam("clusterId") String clusterId,
345373
return notFound();
346374
}
347375

348-
@GET @Path("errors")
376+
@GET
377+
@Path("errors")
349378
public Response getErrorsOnInstance(@PathParam("clusterId") String clusterId,
350379
@PathParam("instanceName") String instanceName) throws IOException {
351380
HelixDataAccessor accessor = getDataAccssor(clusterId);

0 commit comments

Comments
 (0)