Enable logging for all components, remove print stmts (#80)

This change enables a logger for each module of yarn-api-client.
By default, nothing will be logged unless the calling application has
configured logging.  Once configured, entries will be logged relative
to the configured log level. The calling application is also responsible
for specifying the appropriate log formats, etc.

An accessor method, get_logger(logger_name), has been created in case we
need to add configuration for the loggers.  In those cases, we'd probably
configure the logger "yarn_api_client", so that each of the submodule
loggers would inherit from that logger.  Likewise, applications can
configure "yarn_api_client" themselves and those changes will be picked
up by the individual loggers.

The previous INFO log statement has been changed to DEBUG and I've
added timing results for the request. In addition, the previous print
statements have been replaced with calls to apiLogger().warning().

Fixes: #73

Co-authored-by: Dmitry Romanenko <[email protected]>

Author: 2 people

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()

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
 
 
@@ -46,17 +51,19 @@ def _get_resource_manager(hadoop_conf_path, rm_id=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 Exception 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 +79,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: