1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase;
20
21 import com.sun.jersey.api.client.Client;
22 import com.sun.jersey.api.client.ClientResponse;
23 import com.sun.jersey.api.client.filter.HTTPBasicAuthFilter;
24
25 import org.apache.commons.logging.Log;
26 import org.apache.commons.logging.LogFactory;
27 import org.apache.hadoop.conf.Configuration;
28 import org.apache.hadoop.conf.Configured;
29 import org.apache.hadoop.util.ReflectionUtils;
30 import org.codehaus.jackson.JsonNode;
31 import org.codehaus.jackson.map.ObjectMapper;
32
33 import javax.ws.rs.core.MediaType;
34 import javax.ws.rs.core.Response;
35 import javax.ws.rs.core.UriBuilder;
36 import javax.xml.ws.http.HTTPException;
37 import java.io.IOException;
38 import java.net.URI;
39 import java.util.HashMap;
40 import java.util.Map;
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63 public class RESTApiClusterManager extends Configured implements ClusterManager {
64
65
66 private static final String REST_API_CLUSTER_MANAGER_HOSTNAME =
67 "hbase.it.clustermanager.restapi.hostname";
68 private static final String REST_API_CLUSTER_MANAGER_USERNAME =
69 "hbase.it.clustermanager.restapi.username";
70 private static final String REST_API_CLUSTER_MANAGER_PASSWORD =
71 "hbase.it.clustermanager.restapi.password";
72 private static final String REST_API_CLUSTER_MANAGER_CLUSTER_NAME =
73 "hbase.it.clustermanager.restapi.clustername";
74
75
76 private static final String DEFAULT_SERVER_HOSTNAME = "http://localhost:7180";
77 private static final String DEFAULT_SERVER_USERNAME = "admin";
78 private static final String DEFAULT_SERVER_PASSWORD = "admin";
79 private static final String DEFAULT_CLUSTER_NAME = "Cluster 1";
80
81
82
83 private String serverHostname;
84 private String serverUsername;
85 private String serverPassword;
86 private String clusterName;
87
88
89
90 private static final String API_VERSION = "v6";
91
92
93 private Client client = Client.create();
94
95
96
97 private ClusterManager hBaseClusterManager;
98
99 private static final Log LOG = LogFactory.getLog(RESTApiClusterManager.class);
100
101 RESTApiClusterManager() {
102 hBaseClusterManager = ReflectionUtils.newInstance(HBaseClusterManager.class,
103 new IntegrationTestingUtility().getConfiguration());
104 }
105
106 @Override
107 public void setConf(Configuration conf) {
108 super.setConf(conf);
109 if (conf == null) {
110
111 return;
112 }
113 serverHostname = conf.get(REST_API_CLUSTER_MANAGER_HOSTNAME, DEFAULT_SERVER_HOSTNAME);
114 serverUsername = conf.get(REST_API_CLUSTER_MANAGER_USERNAME, DEFAULT_SERVER_USERNAME);
115 serverPassword = conf.get(REST_API_CLUSTER_MANAGER_PASSWORD, DEFAULT_SERVER_PASSWORD);
116 clusterName = conf.get(REST_API_CLUSTER_MANAGER_CLUSTER_NAME, DEFAULT_CLUSTER_NAME);
117
118
119 client.addFilter(new HTTPBasicAuthFilter(serverUsername, serverPassword));
120 }
121
122 @Override
123 public void start(ServiceType service, String hostname, int port) throws IOException {
124 performClusterManagerCommand(service, hostname, RoleCommand.START);
125 }
126
127 @Override
128 public void stop(ServiceType service, String hostname, int port) throws IOException {
129 performClusterManagerCommand(service, hostname, RoleCommand.STOP);
130 }
131
132 @Override
133 public void restart(ServiceType service, String hostname, int port) throws IOException {
134 performClusterManagerCommand(service, hostname, RoleCommand.RESTART);
135 }
136
137 @Override
138 public String isRunning(ServiceType service, String hostname, int port) throws IOException {
139 String serviceName = getServiceName(roleServiceType.get(service));
140 String hostId = getHostId(hostname);
141 String roleState = getRoleState(serviceName, service.toString(), hostId);
142 String healthSummary = getHealthSummary(serviceName, service.toString(), hostId);
143 boolean isRunning = false;
144
145
146
147 if ("STARTED".equals(roleState) && "GOOD".equals(healthSummary)) {
148 isRunning = true;
149 }
150
151
152 return isRunning ? roleState + " " + healthSummary : "";
153 }
154
155 @Override
156 public void kill(ServiceType service, String hostname, int port) throws IOException {
157 hBaseClusterManager.kill(service, hostname, port);
158 }
159
160 @Override
161 public void suspend(ServiceType service, String hostname, int port) throws IOException {
162 hBaseClusterManager.kill(service, hostname, port);
163 }
164
165 @Override
166 public void resume(ServiceType service, String hostname, int port) throws IOException {
167 hBaseClusterManager.kill(service, hostname, port);
168 }
169
170
171
172
173 private void performClusterManagerCommand(ServiceType role, String hostname, RoleCommand command)
174 throws IOException {
175 LOG.info("Performing " + command + " command against " + role + " on " + hostname + "...");
176 String serviceName = getServiceName(roleServiceType.get(role));
177 String hostId = getHostId(hostname);
178 String roleName = getRoleName(serviceName, role.toString(), hostId);
179 doRoleCommand(serviceName, roleName, command);
180 }
181
182
183 private void doRoleCommand(String serviceName, String roleName, RoleCommand roleCommand) {
184 URI uri = UriBuilder.fromUri(serverHostname)
185 .path("api")
186 .path(API_VERSION)
187 .path("clusters")
188 .path(clusterName)
189 .path("services")
190 .path(serviceName)
191 .path("roleCommands")
192 .path(roleCommand.toString())
193 .build();
194 String body = "{ \"items\": [ \"" + roleName + "\" ] }";
195 LOG.info("Executing POST against " + uri + " with body " + body + "...");
196 ClientResponse response = client.resource(uri)
197 .type(MediaType.APPLICATION_JSON)
198 .post(ClientResponse.class, body);
199
200 int statusCode = response.getStatus();
201 if (statusCode != Response.Status.OK.getStatusCode()) {
202 throw new HTTPException(statusCode);
203 }
204 }
205
206
207 private String getHealthSummary(String serviceName, String roleType, String hostId)
208 throws IOException {
209 return getRolePropertyValue(serviceName, roleType, hostId, "healthSummary");
210 }
211
212
213 private String getHostId(String hostname) throws IOException {
214 String hostId = null;
215
216 URI uri = UriBuilder.fromUri(serverHostname)
217 .path("api")
218 .path(API_VERSION)
219 .path("hosts")
220 .build();
221 JsonNode hosts = getJsonNodeFromURIGet(uri);
222 if (hosts != null) {
223
224 for (JsonNode host : hosts) {
225 if (host.get("hostname").getTextValue().equals(hostname)) {
226 hostId = host.get("hostId").getTextValue();
227 break;
228 }
229 }
230 } else {
231 hostId = null;
232 }
233
234 return hostId;
235 }
236
237
238 private JsonNode getJsonNodeFromURIGet(URI uri) throws IOException {
239 LOG.info("Executing GET against " + uri + "...");
240 ClientResponse response = client.resource(uri)
241 .accept(MediaType.APPLICATION_JSON_TYPE)
242 .get(ClientResponse.class);
243
244 int statusCode = response.getStatus();
245 if (statusCode != Response.Status.OK.getStatusCode()) {
246 throw new HTTPException(statusCode);
247 }
248
249 return new ObjectMapper().readTree(response.getEntity(String.class)).get("items");
250 }
251
252
253 private String getRoleName(String serviceName, String roleType, String hostId)
254 throws IOException {
255 return getRolePropertyValue(serviceName, roleType, hostId, "name");
256 }
257
258
259 private String getRolePropertyValue(String serviceName, String roleType, String hostId,
260 String property) throws IOException {
261 String roleValue = null;
262 URI uri = UriBuilder.fromUri(serverHostname)
263 .path("api")
264 .path(API_VERSION)
265 .path("clusters")
266 .path(clusterName)
267 .path("services")
268 .path(serviceName)
269 .path("roles")
270 .build();
271 JsonNode roles = getJsonNodeFromURIGet(uri);
272 if (roles != null) {
273
274 for (JsonNode role : roles) {
275 if (role.get("hostRef").get("hostId").getTextValue().equals(hostId) &&
276 role.get("type")
277 .getTextValue()
278 .toLowerCase()
279 .equals(roleType.toLowerCase())) {
280 roleValue = role.get(property).getTextValue();
281 break;
282 }
283 }
284 }
285
286 return roleValue;
287 }
288
289
290 private String getRoleState(String serviceName, String roleType, String hostId)
291 throws IOException {
292 return getRolePropertyValue(serviceName, roleType, hostId, "roleState");
293 }
294
295
296 private String getServiceName(Service service) throws IOException {
297 String serviceName = null;
298 URI uri = UriBuilder.fromUri(serverHostname)
299 .path("api")
300 .path(API_VERSION)
301 .path("clusters")
302 .path(clusterName)
303 .path("services")
304 .build();
305 JsonNode services = getJsonNodeFromURIGet(uri);
306 if (services != null) {
307
308 for (JsonNode serviceEntry : services) {
309 if (serviceEntry.get("type").getTextValue().equals(service.toString())) {
310 serviceName = serviceEntry.get("name").getTextValue();
311 break;
312 }
313 }
314 }
315
316 return serviceName;
317 }
318
319
320
321
322
323
324
325 private enum RoleCommand {
326 START, STOP, RESTART;
327
328
329 @Override
330 public String toString() {
331 return name().toLowerCase();
332 }
333 }
334
335
336
337
338 private static Map<ServiceType, Service> roleServiceType = new HashMap<ServiceType, Service>();
339 static {
340 roleServiceType.put(ServiceType.HADOOP_NAMENODE, Service.HDFS);
341 roleServiceType.put(ServiceType.HADOOP_DATANODE, Service.HDFS);
342 roleServiceType.put(ServiceType.HADOOP_JOBTRACKER, Service.MAPREDUCE);
343 roleServiceType.put(ServiceType.HADOOP_TASKTRACKER, Service.MAPREDUCE);
344 roleServiceType.put(ServiceType.HBASE_MASTER, Service.HBASE);
345 roleServiceType.put(ServiceType.HBASE_REGIONSERVER, Service.HBASE);
346 }
347
348 private enum Service {
349 HBASE, HDFS, MAPREDUCE
350 }
351 }