[#5] - Ambari client

* Adding curl client for ambari requests

* covering case where string is a port

* Removing status bar from curl output

* [#5] - Adding ambari client

* Ambari client not using correct setters

* Fixing issue with command line arguments in curl request

* Better coverage for ambari tests


With this commit we get the ability to interact with Ambari via the rest API to do any cluster management necessary.

requirements.txt

@@ -1,2 +1,3 @@
 nose
-coverage
+coverage
+mock

scripts/ambari.py

@@ -0,0 +1,86 @@
+import json
+from curl_client import CurlClient
+
+
+class Ambari:
+
+	client = CurlClient()
+	username = password = server = port = proto = ''
+
+	@staticmethod
+	def load_output(output):
+		json_str = ''
+		if not len(output[0]) == 0:
+			json_str = output[0]
+		elif not len(output[1]) == 0:
+			# StdErr output
+			json_str = '{ "message": "" }'
+			res = json.loads(json_str) # Built from json_str no chance for error
+			res['message'] = output[1]
+			return res
+
+
+		else:
+			json_str = '{ "message" : "No output was returned." }'
+
+		res = ''
+
+		try:
+			res = json.loads(json_str)
+		except ValueError as e:
+			raise ValueError(e)
+
+		return res
+
+	def getClusters(self, query=''):
+		output = self.client.make_request('GET', '/api/v1/clusters', query)
+		res = self.load_output(output)
+		return res
+
+	def getServices(self, cluster_name, query=''):
+		output = self.client.make_request('GET', '/api/v1/clusters/' + cluster_name + '/services', query)
+		res = self.load_output(output)
+		return res
+
+	def getClusterInfo(self, cluster_name, query=''):
+		output = self.client.make_request('GET', '/api/v1/clusters/' + cluster_name, query)
+		res = self.load_output(output)
+		return res
+
+	def set_username(self, user):
+		self.username = user
+		self.client.set_username(self.username)
+
+	def set_password(self, password):
+		self.password = password
+		self.client.set_password(self.password)
+
+	def set_proto(self, proto):
+		self.proto = proto
+		self.client.set_proto(self.proto)
+
+	def set_server(self, server):
+		self.server = server
+		self.client.set_server(self.server)
+
+	def set_port(self, port):
+		self.port = port
+		self.client.set_port(self.port)
+
+
+	def __init__(self, username='', password='', proto='http', server='127.0.0.1', port=8080):
+		self.client = CurlClient()
+		if not username == '':
+			self.set_username(username)
+
+		if not password == '':
+			self.set_password(password)
+
+		if not proto == '':
+			self.set_proto(proto)
+
+		if not server == '':
+			self.set_server(server)
+
+		if not len(str(port)) == 0:
+			self.set_port(port)

scripts/curl_client.py

@@ -0,0 +1,90 @@
+import json
+from shell import Shell
+
+
+class CurlClient:
+	username = ''
+	password = ''
+	server = ''
+	port = ''
+	proto = ''
+	cmd = Shell()
+
+	def set_username(self, username):
+		self.username = username
+
+	def set_password(self, password):
+		self.password = password
+
+	# Set HTTP protocol (http or https)
+	def set_proto(self, proto):
+		if proto == 'http' or proto == 'https':
+			self.proto = proto
+		else:
+			raise ValueError('Protocol must be http or https')
+
+	def set_server(self, server):
+		self.server = server
+
+	# A number between 0 and 65535 
+	def set_port(self, port):
+		if not type(port) is int:
+			raise ValueError('Server port was not of type: int')
+		if port > 0 and port <= 65535:
+			self.port = port
+		else:
+			raise ValueError('Server port must be between 0 and 65535. Value was ' + str(port))
+
+
+	# Make a request via cURL
+	# Must pass an HTTP verb GET|PUT|POST|DELETE
+	# The request to the server (possibly something like /api/v1/resource...)
+	# A list of query parameters
+	# 	['param1=value1', 'param2=value2']
+
+	def make_request(self, verb, request, 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
+
+
+		credentials = ':'.join([self.username, self.password])
+		credentials = '-u ' + credentials
+
+		method = '-X ' + verb
+
+		call = ' '.join(['curl -sS', credentials, method, url])
+		output = self.cmd.run(call)
+		return output
+
+
+	def __init__(self, username='', password='', proto='http', server='127.0.0.1', port=8080):
+		self.set_username(username)
+		self.set_password(password)
+		self.set_server(server)
+		self.set_port(port)
+		self.set_proto(proto)
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+

scripts/shell.py

@@ -26,7 +26,7 @@ def run(self, command, args=''):
 		if len(self.cwd) > 0:
 			process = subprocess.Popen(command, shell=True, cwd=path, stdout=subprocess.PIPE, stdin=subprocess.PIPE)
 		else:
-			process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+			process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
 		output = process.communicate()
 		return output
 

tests/test_ambari.py

@@ -0,0 +1,116 @@
+import unittest, json, mock
+from env import scripts
+from mock import Mock
+from scripts.ambari import Ambari
+
+
+sample_cluster_res = '{"href":"http://demo-server:8080/api/v1/clusters","items":[{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster","Clusters":{"cluster_name":"demo_cluster","version":"HDP-2.4"}}]}'
+
+sample_services_res = '{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster?fields=services","Clusters":{"cluster_name":"demo_cluster","version":"HDP-2.4"},"services":[{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/ACCUMULO","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"ACCUMULO"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/AMBARI_METRICS","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"AMBARI_METRICS"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/ATLAS","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"ATLAS"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/FALCON","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"FALCON"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/HBASE","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"HBASE"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/HDFS","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"HDFS"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/HIVE","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"HIVE"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/KAFKA","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"KAFKA"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/MAPREDUCE2","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"MAPREDUCE2"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/OOZIE","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"OOZIE"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/PIG","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"PIG"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/SPARK","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"SPARK"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/STORM","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"STORM"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/TEZ","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"TEZ"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/YARN","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"YARN"}},{"href":"http://demo-server:8080/api/v1/clusters/demo_cluster/services/ZOOKEEPER","ServiceInfo":{"cluster_name":"demo_cluster","service_name":"ZOOKEEPER"}}]}'
+
+no_cluster_res = '{"status":404,"message":"The requested resource doesn\'t exist: Cluster not found, clusterName=d"}'
+
+err_res = 'error message here'
+empty_res = ''
+bad_res = '{}]}}}'
+bad_err_res = '{}{}{}{}{}{}{}{}STRY()'
+
+def mocked_request(*args, **kwargs):
+
+	if '/api/v1/clusters?' in args[0]:
+		return [sample_cluster_res, '']
+	elif ('/api/v1/clusters/demo_cluster?' in args[0]):
+		return [sample_services_res, '']
+	elif ('/api/v1/clusters/demo_cluster/services?' in args[0]):
+		return [sample_services_res, '']
+	elif ('bad/request' in args[0]):
+		return ['', err_res]
+	elif ('empty/res' in args[0]):
+		return ['', empty_res]
+	elif ('bad/json/res' in args[0]):
+		return [bad_res, '']
+	elif ('err/json/res' in args[0]):
+		return ['', bad_err_res]
+	else:
+		return [no_cluster_res, '']
+
+class TestAmbariClient(unittest.TestCase):
+
+	un = 'admin'
+	pw = 'admin'
+	proto = 'http'
+	server = 'demo-server'
+	port = 8080
+
+
+	@mock.patch('scripts.shell.Shell.run', side_effect=mocked_request)
+	def test_clusters_request(self, mock):
+		client = Ambari(self.un, self.pw, self.proto, self.server, self.port)
+		data = client.getClusters();
+		assert 'demo-server:8080' in data['href']
+		assert data['items'][0]['Clusters']['cluster_name'] == 'demo_cluster'
+		assert data['items'][0]['Clusters']['version'] == 'HDP-2.4'
+
+
+	@mock.patch('scripts.shell.Shell.run', side_effect=mocked_request)	
+	def test_services_request(self, mock):
+		client = Ambari(self.un, self.pw, self.proto, self.server, self.port)
+		data = client.getServices('demo_cluster');
+		assert 'demo-server:8080' in data['href']
+		assert len(data['services']) == 16
+
+
+	@mock.patch('scripts.shell.Shell.run', side_effect=mocked_request)
+	def test_cluster_info_request(self, mock):
+		client = Ambari(self.un, self.pw, self.proto, self.server, self.port)
+		data = client.getClusterInfo('demo_cluster');
+		assert 'demo-server:8080' in data['href']
+		assert len(data['services']) == 16
+
+	@mock.patch('scripts.shell.Shell.run', side_effect=mocked_request)
+	def test_missing_cluster(self, mock):
+		client = Ambari(self.un, self.pw, self.proto, self.server, self.port)
+		data = client.getClusterInfo('');
+		assert data['status'] == 404
+		assert 'resource doesn\'t exist' in data['message']
+
+	@mock.patch('scripts.shell.Shell.run', side_effect=mocked_request)
+	def test_err_str(self, mock):
+		client = Ambari(self.un, self.pw, self.proto, self.server, self.port)
+		data = client.getClusterInfo('bad/request');
+		assert not len(data['message']) == 0
+		assert data['message'] == err_res
+
+	@mock.patch('scripts.shell.Shell.run', side_effect=mocked_request)
+	def test_empty_str(self, mock):
+		client = Ambari(self.un, self.pw, self.proto, self.server, self.port)
+		data = client.getClusterInfo('empty/res');
+		assert not len(data['message']) == 0
+		assert data['message'] == 'No output was returned.'
+
+	@mock.patch('scripts.shell.Shell.run', side_effect=mocked_request)
+	def test_bad_json_res(self, mock):
+		client = Ambari(self.un, self.pw, self.proto, self.server, self.port)
+		try:
+			data = client.getClusterInfo('bad/json/res');
+			self.fail('Should have thrown an exception: ValueError')
+		except ValueError as e:
+			assert ('Extra data:' in str(e.message))
+			assert not len(str(e.message)) == 0
+			pass
+
+	@mock.patch('scripts.shell.Shell.run', side_effect=mocked_request)
+	def test_err_json_res(self, mock):
+		client = Ambari(self.un, self.pw, self.proto, self.server, self.port)
+		data = client.getClusterInfo('err/json/res');
+		assert not len(str(data['message'])) == 0
+		assert data['message'] == bad_err_res
+
+
+
+
+
+
+
+
+