[#22] - Ambari Service Actions

docs/ambari.md

@@ -24,8 +24,32 @@ The Ambari client can take anywhere between 0 and 5 different arguments:
 
 The defaults for each of the above are shown in the argument list.
 
+### `Ambari.service_action(cluster_name, service_name, action)`
 
-### `Ambari.getClusters(query='')`
+- `action`
+  - Must only be one one of: `'START'`, `'STOP'`, or `'RESTART'`
+
+Makes RESTful API call to `GET /api/v1/clusters/{cluster_name}/{service_name}` in order to Stop/Start/Restart a given service.
+
+Returns `True` or `False` based on whether or not the action was successful
+
+This function utilizes `self.service_wait_time` to determine the longest amount of time we wait before failing the process
+
+This function will return `True` or `False` based on whether or not the service was successful in starting, stopping, or restarting
+
+### `Ambari.get_service(cluster_name, service_name, query='')`
+
+Makes RESTful API call to `GET /api/v1/clusters/{cluster_name}/{service_name}`
+
+Returns a piece of JSON via `json.loads` from the API.
+
+If an error occurred when making the call to Ambari the error message from the curl client will be returned in the following format:
+
+	{
+		"message": error_message_string
+	}
+
+### `Ambari.get_clusters(query='')`
 
 Makes RESTful API call to `GET /api/v1/clusters`
 
@@ -37,7 +61,7 @@ If an error occurred when making the call to Ambari the error message from the c
 		"message": error_message_string
 	}
 
-### `Ambari.getServices(cluster_name, query='')`
+### `Ambari.get_services(cluster_name, query='')`
 
 
 Makes RESTful API call to `GET /api/v1/clusters/{CLUSTER_NAME}/services`
@@ -51,7 +75,7 @@ If an error occurred when making the call to Ambari the error message from the c
 	}
 
 
-### `Ambari.getClusterInfo(cluster_name, query='')`
+### `Ambari.get_cluster_info(cluster_name, query='')`
 
 Makes RESTful API call to `GET /api/v1/clusters/{CLUSTER_NAME}`
 
@@ -83,7 +107,11 @@ Updates the the desired ambari server hostname/IP Address to make REST calls to
 
 ### `Ambari.set_port(port)`
 
-Updates the desired port on which Ambari is running. **Must be an integer**.
+Updates the desired port on which Ambari is running. **Must be an integer between 0 and 65535**.
+
+### `Ambari.set_service_wait_time(wait_time)`
+
+Updates the total amount of **time in seconds** that we will wait before assuming a service has failed to Stop/Start. **Must be a number greater than 0**.
 
 
 

scripts/ambari.py

@@ -1,12 +1,11 @@
-import json
+import json, time
 from curl_client import CurlClient
 from logs import Logger
 
 logger = Logger('Ambari').getLogger()
 
 class Ambari:
 
-	client = CurlClient()
 	username = password = server = port = proto = ''
 
 	@staticmethod
@@ -33,23 +32,81 @@ def load_output(output):
 
 		return res
 
-	def getClusters(self, query=''):
+	def service_action(self, cluster_name, service_name, action):
+
+		logger.info('Attempting Service Action: ' + action + '; Service Name: ' + service_name)
+		if not (action == 'START' or action == 'RESTART' or action == 'STOP'):
+			logger.error('Service actions was not one of START, STOP, or RESTART')
+			raise ValueError('Service action must be one of: START, STOP, or RESTART')
+
+		logger.info(action + 'ing ' + service_name + ' via Ambari REST API')
+
+		request_payload = {}
+		request_payload['RequestInfo'] = {}
+		request_payload['RequestInfo']['context'] = action + ' ' + service_name + ' via REST'
+		request_payload['Body'] = {}
+		request_payload['Body']['ServiceInfo'] = {}
+
+		service_info = self.get_service(cluster_name, service_name, 'fields=ServiceInfo')[0]
+		before_state = service_info['ServiceInfo']['state']
+		logger.debug('Service state before attempting to change: ' + str(before_state))
+		after_state = ''
+
+		if action == 'STOP':
+			after_state = 'INSTALLED'
+
+		elif action == 'START':
+			after_state = 'STARTED'
+
+		elif action == 'RESTART':
+			r1 = self.service_action(service_name, cluster_name, 'STOP')
+			r2 = self.service_action(service_name, cluster_name, 'START')
+			return r1 and r2
+
+		request_payload['Body']['ServiceInfo']['state'] = after_state
+
+		payload = json.dumps(request_payload)
+		res = self.client.make_request('PUT', '/api/v1/clusters/' + cluster_name + '/services/' + service_name, '-i -d \'' + payload + '\' -H "X-Requested-By:ambari"', 'fields=ServiceInfo')
+
+		if not ('202 Accepted' in res[0] or '200 OK' in res[0]):
+			logger.error('No 200 Level status when attempting to change service state')
+			return False
+
+		service_state = ''
+		t = 0
+		while t < self.service_wait_time:
+			logger.debug('Checking for a change in service state')
+			service_state = self.get_service(cluster_name, service_name, 'fields=ServiceInfo')[0]['ServiceInfo']['state']
+			if service_state == after_state:
+				logger.info('Service action completed successfully')
+				return True
+			t += 1
+			time.sleep(1)
+
+		return False
+
+
+	def get_clusters(self, query=''):
 		logger.info('Making request to /api/v1/clusters/')
 		output = self.client.make_request('GET', '/api/v1/clusters', query)
 		res = self.load_output(output)
 		return res
 
-	def getServices(self, cluster_name, query=''):
+	def get_services(self, cluster_name, query=''):
 		logger.info('Making request to /api/v1/clusters/' + cluster_name + '/services')
 		output = self.client.make_request('GET', '/api/v1/clusters/' + cluster_name + '/services', query)
-		res = self.load_output(output)
-		return res
+		return self.load_output(output)
 
-	def getClusterInfo(self, cluster_name, query=''):
+	def get_cluster_info(self, cluster_name, query=''):
 		logger.info('Making request to /api/v1/clusters/' + cluster_name)
 		output = self.client.make_request('GET', '/api/v1/clusters/' + cluster_name, query)
-		res = self.load_output(output)
-		return res
+		return self.load_output(output)
+
+	def get_service(self, cluster_name, service_name, query=''):
+		logger.info('Making request to /api/v1/clusters/' + cluster_name)
+		output = self.client.make_request('GET', '/api/v1/clusters/' + cluster_name + '/' + service_name, query)
+		return self.load_output(output)
+
 
 	def set_username(self, user):
 		self.username = user
@@ -71,9 +128,17 @@ def set_port(self, port):
 		self.port = port
 		self.client.set_port(self.port)
 
+	def set_service_wait_time(self, wait_time):
+		if wait_time > 0:
+			self.service_wait_time = wait_time
+
 
-	def __init__(self, username='', password='', proto='http', server='127.0.0.1', port=8080):
+	def __init__(self, username='', password='', proto='http', server='127.0.0.1', port=8080, service_wait_time=60):
 		self.client = CurlClient()
+		if service_wait_time > 0:
+			self.set_service_wait_time(service_wait_time)
+		else:
+			self.set_service_wait_time(60)
 		if not username == '':
 			self.set_username(username)
 

scripts/curl_client.py

@@ -57,7 +57,6 @@ def make_request(self, verb, request, options='', query=''):
 		if not (verb == 'GET' or verb == 'POST' or verb == 'PUT' or verb == 'DELETE'):
 			raise ValueError('HTTP Verb must be one of GET|PUT|POST|DELETE')
 
-		query = '&'.join(query)
 		url = ''.join([self.proto, '://', self.server, ':', str(self.port), request])
 		url = url + '?'
 		url = url + query