Merge branch 'master' into release_102

Author: kevin-bates

tests/test_base.py

@@ -70,13 +70,12 @@ def test_uri_parsing(self):
         self.assertEqual(result_uri.hostname, '123.45.67.89')
         self.assertEqual(result_uri.port, 1234)
         self.assertEqual(result_uri.is_https, False)
-        
+
         result_uri = base.Uri('https://test-domain.com:1234')
         self.assertEqual(result_uri.scheme, 'https')
         self.assertEqual(result_uri.hostname, 'test-domain.com')
         self.assertEqual(result_uri.port, 1234)
         self.assertEqual(result_uri.is_https, True)
-        
 
     def get_client(self):
         client = base.BaseYarnAPI()

tests/test_hadoop_conf.py

@@ -3,6 +3,7 @@
 
 import mock
 from mock import patch
+from requests import RequestException
 from tests import TestCase
 
 import requests_mock
@@ -103,7 +104,7 @@ def test_get_resource_endpoint(self):
 
                 endpoint = hadoop_conf.get_resource_manager_endpoint()
 
-                self.assertEqual('example.com:8022', endpoint)
+                self.assertEqual('http://example.com:8022', endpoint)
                 parse_mock.assert_called_with(hadoop_conf_path + 'yarn-site.xml',
                                               'yarn.resourcemanager.webapp.address')
 
@@ -122,7 +123,7 @@ def test_get_resource_endpoint_with_ha(self, check_is_active_rm_mock, parse_mock
         check_is_active_rm_mock.return_value = True
         endpoint = hadoop_conf.get_resource_manager_endpoint()
 
-        self.assertEqual('example.com:8022', endpoint)
+        self.assertEqual('http://example.com:8022', endpoint)
         parse_mock.assert_called_with(hadoop_conf_path + 'yarn-site.xml',
                                       'yarn.resourcemanager.webapp.address.rm1')
 
@@ -171,23 +172,21 @@ def test_check_is_active_rm(self, is_https_only_mock):
 
         # Emulate requests library exception (socket timeout, etc)
         with requests_mock.mock() as requests_get_mock:
-            requests_get_mock.side_effect = Exception('error')
-            # requests_get_mock.get('https://example2:8022/cluster', status_code=200)
-            requests_get_mock.return_value = None
-            self.assertFalse(hadoop_conf.check_is_active_rm('https://example2:8022'))
+            requests_get_mock.get('example2:8022/cluster', exc=RequestException)
+            self.assertFalse(hadoop_conf.check_is_active_rm('example2:8022'))
 
     def test_get_resource_manager(self):
         with patch('yarn_api_client.hadoop_conf.parse') as parse_mock:
             parse_mock.return_value = 'example.com:8022'
 
             endpoint = hadoop_conf._get_resource_manager(hadoop_conf.CONF_DIR, None)
 
-            self.assertEqual('example.com:8022', endpoint)
+            self.assertEqual('http://example.com:8022', endpoint)
             parse_mock.assert_called_with(hadoop_conf_path + 'yarn-site.xml', 'yarn.resourcemanager.webapp.address')
 
             endpoint = hadoop_conf._get_resource_manager(hadoop_conf.CONF_DIR, 'rm1')
 
-            self.assertEqual(('example.com:8022'), endpoint)
+            self.assertEqual(('http://example.com:8022'), endpoint)
             parse_mock.assert_called_with(hadoop_conf_path + 'yarn-site.xml', 'yarn.resourcemanager.webapp.address.rm1')
 
             parse_mock.reset_mock()

yarn_api_client/application_master.py

@@ -1,9 +1,13 @@
 # -*- coding: utf-8 -*-
 from __future__ import unicode_literals
-from .base import BaseYarnAPI
+
+from .base import BaseYarnAPI, get_logger
 from .hadoop_conf import get_webproxy_endpoint
 
 
+log = get_logger(__name__)
+
+
 class ApplicationMaster(BaseYarnAPI):
     """
     The MapReduce Application Master REST API's allow the user to get status
@@ -24,7 +28,6 @@ class ApplicationMaster(BaseYarnAPI):
     """
     def __init__(self, service_endpoint=None, timeout=30, auth=None, verify=True):
         if not service_endpoint:
-            self.logger.debug('Get configuration from hadoop conf dir')
             service_endpoint = get_webproxy_endpoint(timeout, auth, verify)
 
         super(ApplicationMaster, self).__init__(service_endpoint, timeout, auth, verify)

yarn_api_client/base.py

@@ -2,8 +2,11 @@
 from __future__ import unicode_literals
 
 import logging
+import os
 import requests
 
+from datetime import datetime
+
 from .errors import APIError, ConfigurationError
 
 try:
@@ -12,6 +15,14 @@
     from urllib.parse import urlparse, urlunparse
 
 
+def get_logger(logger_name):
+    logger = logging.getLogger(logger_name)
+    return logger
+
+
+log = get_logger(__name__)
+
+
 class Response(object):
     """
     Basic container for response dictionary
@@ -49,7 +60,6 @@ def to_url(self, api_path=None):
 
 
 class BaseYarnAPI(object):
-    __logger = None
     response_class = Response
 
     def __init__(self, service_endpoint=None, timeout=None, auth=None, verify=True):
@@ -72,8 +82,6 @@ def request(self, api_path, method='GET', **kwargs):
         self._validate_configuration()
         api_endpoint = self.service_uri.to_url(api_path)
 
-        self.logger.info('API Endpoint {}'.format(api_endpoint))
-
         if method == 'GET':
             headers = {}
         else:
@@ -82,20 +90,26 @@ def request(self, api_path, method='GET', **kwargs):
         if 'headers' in kwargs and kwargs['headers']:
             headers.update(kwargs['headers'])
 
+        begin = datetime.now()
         response = self.session.request(method=method, url=api_endpoint, headers=headers, timeout=self.timeout, **kwargs)
+        end = datetime.now()
+        log.debug(
+            "'{method}' request against endpoint '{endpoint}' took {duration} ms".format(
+                method=method, 
+                endpoint=api_endpoint,
+                duration=round((end-begin).total_seconds()*1000,3)
+            )
+        )
 
         if response.status_code in (200, 202):
             return self.response_class(response)
         else:
-            msg = 'Response finished with status: %s. Details: %s' % (response.status_code, response.text)
+            msg = "Response finished with status: {status}. Details: {msg}".format(
+                status=response.status_code, 
+                msg=response.text
+            )
             raise APIError(msg)
 
     def construct_parameters(self, arguments):
         params = dict((key, value) for key, value in arguments if value is not None)
         return params
-
-    @property
-    def logger(self):
-        if self.__logger is None:
-            self.__logger = logging.getLogger(self.__module__)
-        return self.__logger

yarn_api_client/hadoop_conf.py

@@ -3,6 +3,10 @@
 import xml.etree.ElementTree as ET
 import requests
 
+from .base import get_logger
+
+log = get_logger(__name__)
+
 CONF_DIR = os.getenv('HADOOP_CONF_DIR', '/etc/hadoop/conf')
 
 
@@ -14,7 +18,8 @@ def _get_rm_ids(hadoop_conf_path):
 
 
 def _get_maximum_container_memory(hadoop_conf_path):
-    container_memory = int(parse(os.path.join(hadoop_conf_path,'yarn-site.xml'), 'yarn.nodemanager.resource.memory-mb'))
+    container_memory = int(parse(os.path.join(hadoop_conf_path, 'yarn-site.xml'),
+                                 'yarn.nodemanager.resource.memory-mb'))
     return container_memory
 
 
@@ -29,7 +34,9 @@ def _is_https_only():
 
 def _get_resource_manager(hadoop_conf_path, rm_id=None):
     # compose property name based on policy (and rm_id)
-    if _is_https_only():
+    is_https_only = _is_https_only()
+
+    if is_https_only:
         prop_name = 'yarn.resourcemanager.webapp.https.address'
     else:
         prop_name = 'yarn.resourcemanager.webapp.address'
@@ -38,25 +45,27 @@ def _get_resource_manager(hadoop_conf_path, rm_id=None):
     if rm_id:
         prop_name = "{name}.{rm_id}".format(name=prop_name, rm_id=rm_id)
 
-    rm_webapp_address = parse(os.path.join(hadoop_conf_path, 'yarn-site.xml'), prop_name)
+    rm_address = parse(os.path.join(hadoop_conf_path, 'yarn-site.xml'), prop_name)
 
-    return rm_webapp_address or None
+    return ('https://' if is_https_only else 'http://') + rm_address if rm_address else None
 
 
 def check_is_active_rm(url, timeout=30, auth=None, verify=True):
     try:
         response = requests.get(url + "/cluster", timeout=timeout, auth=auth, verify=verify)
-    except:
+    except requests.RequestException as e:
+        log.warning("Exception encountered accessing RM '{url}': '{err}', continuing...".format(url=url, err=e))
         return False
 
     if response.status_code != 200:
-        print("Error to access RM - HTTP Code {}".format(response.status_code))
+        log.warning("Failed to access RM '{url}' - HTTP Code '{status}', continuing...".format(url=url, status=response.status_code))
         return False
     else:
         return True
 
 
 def get_resource_manager_endpoint(timeout=30, auth=None, verify=True):
+    log.info('Getting resource manager endpoint from config: {config_path}'.format(config_path=os.path.join(CONF_DIR, 'yarn-site.xml')))
     hadoop_conf_path = CONF_DIR
     rm_ids = _get_rm_ids(hadoop_conf_path)
     if rm_ids:
@@ -72,18 +81,21 @@ def get_resource_manager_endpoint(timeout=30, auth=None, verify=True):
 
 def get_jobhistory_endpoint():
     config_path = os.path.join(CONF_DIR, 'mapred-site.xml')
+    log.info('Getting jobhistory endpoint from config: {config_path}'.format(config_path=config_path))
     prop_name = 'mapreduce.jobhistory.webapp.address'
     return parse(config_path, prop_name)
 
 
 def get_nodemanager_endpoint():
     config_path = os.path.join(CONF_DIR, 'yarn-site.xml')
+    log.info('Getting nodemanager endpoint from config: {config_path}'.format(config_path=config_path))
     prop_name = 'yarn.nodemanager.webapp.address'
     return parse(config_path, prop_name)
 
 
 def get_webproxy_endpoint(timeout=30, auth=None, verify=True):
     config_path = os.path.join(CONF_DIR, 'yarn-site.xml')
+    log.info('Getting webproxy endpoint from config: {config_path}'.format(config_path=config_path))
     prop_name = 'yarn.web-proxy.address'
     value = parse(config_path, prop_name)
     return value or get_resource_manager_endpoint(timeout, auth, verify)

yarn_api_client/history_server.py

@@ -1,10 +1,13 @@
 # -*- coding: utf-8 -*-
 from __future__ import unicode_literals
-from .base import BaseYarnAPI
+
+from .base import BaseYarnAPI, get_logger
 from .constants import JobStateInternal
 from .errors import IllegalArgumentError
 from .hadoop_conf import get_jobhistory_endpoint
 
+log = get_logger(__name__)
+
 
 class HistoryServer(BaseYarnAPI):
     """
@@ -24,7 +27,6 @@ class HistoryServer(BaseYarnAPI):
     """
     def __init__(self, service_endpoint=None, timeout=30, auth=None, verify=True):
         if not service_endpoint:
-            self.logger.debug('Get information from hadoop conf dir')
             service_endpoint = get_jobhistory_endpoint()
 
         super(HistoryServer, self).__init__(service_endpoint, timeout, auth, verify)

yarn_api_client/main.py

@@ -1,15 +1,14 @@
 # -*- coding: utf-8 -*-
 from __future__ import unicode_literals
 import argparse
-import logging
 from pprint import pprint
-import sys
 
+from .base import get_logger
 from .constants import (YarnApplicationState, FinalApplicationStatus,
                         ApplicationState, JobStateInternal)
 from . import ResourceManager, NodeManager, HistoryServer, ApplicationMaster
 
-logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
+log = get_logger(__name__)
 
 
 def get_parser():

yarn_api_client/node_manager.py

@@ -1,9 +1,11 @@
 # -*- coding: utf-8 -*-
-from .base import BaseYarnAPI
+from .base import BaseYarnAPI, get_logger
 from .constants import ApplicationState
 from .errors import IllegalArgumentError
 from .hadoop_conf import get_nodemanager_endpoint
 
+log = get_logger(__name__)
+
 LEGAL_APPLICATION_STATES = {s for s, _ in ApplicationState}
 
 
@@ -35,7 +37,6 @@ class NodeManager(BaseYarnAPI):
     """
     def __init__(self, service_endpoint=None, timeout=30, auth=None, verify=True):
         if not service_endpoint:
-            self.logger.debug('Get configuration from hadoop conf dir')
             service_endpoint = get_nodemanager_endpoint()
 
         super(NodeManager, self).__init__(service_endpoint, timeout, auth, verify)

yarn_api_client/resource_manager.py

@@ -1,11 +1,12 @@
 # -*- coding: utf-8 -*-
 from __future__ import unicode_literals
-from .base import BaseYarnAPI
+from .base import BaseYarnAPI, get_logger
 from .constants import YarnApplicationState, FinalApplicationStatus
 from .errors import IllegalArgumentError
 from .hadoop_conf import get_resource_manager_endpoint, check_is_active_rm, CONF_DIR, _get_maximum_container_memory
 from collections import deque
 
+log = get_logger(__name__)
 LEGAL_STATES = {s for s, _ in YarnApplicationState}
 LEGAL_FINAL_STATUSES = {s for s, _ in FinalApplicationStatus}
 
@@ -73,7 +74,6 @@ class ResourceManager(BaseYarnAPI):
     def __init__(self, service_endpoints=None, timeout=30, auth=None, verify=True):
         active_service_endpoint = None
         if not service_endpoints:
-            self.logger.debug('Get configuration from hadoop conf dir: {conf_dir}'.format(conf_dir=CONF_DIR))
             active_service_endpoint = get_resource_manager_endpoint(timeout, auth, verify)
         else:
             for endpoint in service_endpoints: