unifying logger usage thus preventing multiple instantiation of logging-singleton

Author: kirschte

market_maker/bitmex.py

@@ -6,11 +6,12 @@
 import json
 import base64
 import uuid
-import logging
 from market_maker.auth import APIKeyAuthWithExpires
-from market_maker.utils import constants, errors
+from market_maker.utils import constants, errors, log
 from market_maker.ws.ws_thread import BitMEXWebsocket
 
+logger = log.setup_custom_logger('root')
+
 
 # https://www.bitmex.com/api/explorer/
 class BitMEX(object):
@@ -20,7 +21,6 @@ class BitMEX(object):
     def __init__(self, base_url=None, symbol=None, apiKey=None, apiSecret=None,
                  orderIDPrefix='mm_bitmex_', shouldWSAuth=True, postOnly=False, timeout=7):
         """Init connector."""
-        self.logger = logging.getLogger('root')
         self.base_url = base_url
         self.symbol = symbol
         self.postOnly = postOnly
@@ -255,7 +255,7 @@ def retry():
         # Make the request
         response = None
         try:
-            self.logger.info("sending req to %s: %s" % (url, json.dumps(postdict or query or '')))
+            logger.info("sending req to %s: %s" % (url, json.dumps(postdict or query or '')))
             req = requests.Request(verb, url, json=postdict, auth=auth, params=query)
             prepped = self.session.prepare_request(req)
             response = self.session.send(prepped, timeout=timeout)
@@ -268,25 +268,25 @@ def retry():
 
             # 401 - Auth error. This is fatal.
             if response.status_code == 401:
-                self.logger.error("API Key or Secret incorrect, please check and restart.")
-                self.logger.error("Error: " + response.text)
+                logger.error("API Key or Secret incorrect, please check and restart.")
+                logger.error("Error: " + response.text)
                 if postdict:
-                    self.logger.error(postdict)
+                    logger.error(postdict)
                 # Always exit, even if rethrow_errors, because this is fatal
                 exit(1)
 
             # 404, can be thrown if order canceled or does not exist.
             elif response.status_code == 404:
                 if verb == 'DELETE':
-                    self.logger.error("Order not found: %s" % postdict['orderID'])
+                    logger.error("Order not found: %s" % postdict['orderID'])
                     return
-                self.logger.error("Unable to contact the BitMEX API (404). " +
+                logger.error("Unable to contact the BitMEX API (404). " +
                                   "Request: %s \n %s" % (url, json.dumps(postdict)))
                 exit_or_throw(e)
 
             # 429, ratelimit; cancel orders & wait until X-RateLimit-Reset
             elif response.status_code == 429:
-                self.logger.error("Ratelimited on current request. Sleeping, then trying again. Try fewer " +
+                logger.error("Ratelimited on current request. Sleeping, then trying again. Try fewer " +
                                   "order pairs or contact support@bitmex.com to raise your limits. " +
                                   "Request: %s \n %s" % (url, json.dumps(postdict)))
 
@@ -296,18 +296,18 @@ def retry():
                 reset_str = datetime.datetime.fromtimestamp(int(ratelimit_reset)).strftime('%X')
 
                 # We're ratelimited, and we may be waiting for a long time. Cancel orders.
-                self.logger.warning("Canceling all known orders in the meantime.")
+                logger.warning("Canceling all known orders in the meantime.")
                 self.cancel([o['orderID'] for o in self.open_orders()])
 
-                self.logger.error("Your ratelimit will reset at %s. Sleeping for %d seconds." % (reset_str, to_sleep))
+                logger.error("Your ratelimit will reset at %s. Sleeping for %d seconds." % (reset_str, to_sleep))
                 time.sleep(to_sleep)
 
                 # Retry the request.
                 return retry()
 
             # 503 - BitMEX temporary downtime, likely due to a deploy. Try again
             elif response.status_code == 503:
-                self.logger.warning("Unable to contact the BitMEX API (503), retrying. " +
+                logger.warning("Unable to contact the BitMEX API (503), retrying. " +
                                     "Request: %s \n %s" % (url, json.dumps(postdict)))
                 time.sleep(3)
                 return retry()
@@ -336,22 +336,22 @@ def retry():
                     return orderResults
 
                 elif 'insufficient available balance' in message:
-                    self.logger.error('Account out of funds. The message: %s' % error['message'])
+                    logger.error('Account out of funds. The message: %s' % error['message'])
                     exit_or_throw(Exception('Insufficient Funds'))
 
 
             # If we haven't returned or re-raised yet, we get here.
-            self.logger.error("Unhandled Error: %s: %s" % (e, response.text))
-            self.logger.error("Endpoint was: %s %s: %s" % (verb, path, json.dumps(postdict)))
+            logger.error("Unhandled Error: %s: %s" % (e, response.text))
+            logger.error("Endpoint was: %s %s: %s" % (verb, path, json.dumps(postdict)))
             exit_or_throw(e)
 
         except requests.exceptions.Timeout as e:
             # Timeout, re-run this request
-            self.logger.warning("Timed out on request: %s (%s), retrying..." % (path, json.dumps(postdict or '')))
+            logger.warning("Timed out on request: %s (%s), retrying..." % (path, json.dumps(postdict or '')))
             return retry()
 
         except requests.exceptions.ConnectionError as e:
-            self.logger.warning("Unable to contact the BitMEX API (%s). Please check the URL. Retrying. " +
+            logger.warning("Unable to contact the BitMEX API (%s). Please check the URL. Retrying. " +
                                 "Request: %s %s \n %s" % (e, url, json.dumps(postdict)))
             time.sleep(1)
             return retry()

market_maker/utils/log.py

@@ -1,14 +1,19 @@
 import logging
 from market_maker.settings import settings
 
+loggers = {}
+
 
 def setup_custom_logger(name, log_level=settings.LOG_LEVEL):
-    formatter = logging.Formatter(fmt='%(asctime)s - %(levelname)s - %(module)s - %(message)s')
+    if loggers.get(name):
+        return loggers[name]
+
+    logger = logging.getLogger(name)
+    loggers[name] = logger
 
+    formatter = logging.Formatter(fmt='%(asctime)s - %(levelname)s - %(module)s - %(message)s')
     handler = logging.StreamHandler()
     handler.setFormatter(formatter)
-
-    logger = logging.getLogger(name)
     logger.setLevel(log_level)
     logger.addHandler(handler)
     return logger

market_maker/ws/ws_thread.py

@@ -9,14 +9,17 @@
 import logging
 from market_maker.settings import settings
 from market_maker.auth.APIKeyAuth import generate_expires, generate_signature
-from market_maker.utils.log import setup_custom_logger
+from market_maker.utils import log
 from market_maker.utils.math import toNearest
 from future.utils import iteritems
 from future.standard_library import hooks
 with hooks():  # Python 2/3 compat
     from urllib.parse import urlparse, urlunparse
 
 
+logger = log.setup_custom_logger('root')
+
+
 # Connects to BitMEX websocket for streaming realtime data.
 # The Marketmaker still interacts with this as if it were a REST Endpoint, but now it can get
 # much more realtime data without heavily polling the API.
@@ -31,7 +34,6 @@ class BitMEXWebsocket():
     MAX_TABLE_LEN = 200
 
     def __init__(self):
-        self.logger = logging.getLogger('root')
         self.__reset()
 
     def __del__(self):
@@ -40,7 +42,7 @@ def __del__(self):
     def connect(self, endpoint="", symbol="XBTN15", shouldAuth=True):
         '''Connect to the websocket and initialize data stores.'''
 
-        self.logger.debug("Connecting WebSocket.")
+        logger.debug("Connecting WebSocket.")
         self.symbol = symbol
         self.shouldAuth = shouldAuth
 
@@ -57,15 +59,15 @@ def connect(self, endpoint="", symbol="XBTN15", shouldAuth=True):
         urlParts[0] = urlParts[0].replace('http', 'ws')
         urlParts[2] = "/realtime?subscribe=" + ",".join(subscriptions)
         wsURL = urlunparse(urlParts)
-        self.logger.info("Connecting to %s" % wsURL)
+        logger.info("Connecting to %s" % wsURL)
         self.__connect(wsURL)
-        self.logger.info('Connected to WS. Waiting for data images, this may take a moment...')
+        logger.info('Connected to WS. Waiting for data images, this may take a moment...')
 
         # Connected. Wait for partials
         self.__wait_for_symbol(symbol)
         if self.shouldAuth:
             self.__wait_for_account()
-        self.logger.info('Got all market data. Starting.')
+        logger.info('Got all market data. Starting.')
 
     #
     # Data methods
@@ -132,7 +134,7 @@ def recent_trades(self):
     #
     def error(self, err):
         self._error = err
-        self.logger.error(err)
+        logger.error(err)
         self.exit()
 
     def exit(self):
@@ -145,7 +147,7 @@ def exit(self):
 
     def __connect(self, wsURL):
         '''Connect to the websocket in a thread.'''
-        self.logger.debug("Starting thread")
+        logger.debug("Starting thread")
 
         ssl_defaults = ssl.get_default_verify_paths()
         sslopt_ca_certs = {'ca_certs': ssl_defaults.cafile}
@@ -157,11 +159,10 @@ def __connect(self, wsURL):
                                          header=self.__get_auth()
                                          )
 
-        setup_custom_logger('websocket', log_level=settings.LOG_LEVEL)
         self.wst = threading.Thread(target=lambda: self.ws.run_forever(sslopt=sslopt_ca_certs))
         self.wst.daemon = True
         self.wst.start()
-        self.logger.info("Started thread")
+        logger.info("Started thread")
 
         # Wait for connect before continuing
         conn_timeout = 5
@@ -170,7 +171,7 @@ def __connect(self, wsURL):
             conn_timeout -= 1
 
         if not conn_timeout or self._error:
-            self.logger.error("Couldn't connect to WS! Exiting.")
+            logger.error("Couldn't connect to WS! Exiting.")
             self.exit()
             sys.exit(1)
 
@@ -180,7 +181,7 @@ def __get_auth(self):
         if self.shouldAuth is False:
             return []
 
-        self.logger.info("Authenticating with API Key.")
+        logger.info("Authenticating with API Key.")
         # To auth to the WS using an API key, we generate a signature of a nonce and
         # the WS API endpoint.
         nonce = generate_expires()
@@ -208,14 +209,14 @@ def __send_command(self, command, args):
     def __on_message(self, message):
         '''Handler for parsing WS messages.'''
         message = json.loads(message)
-        self.logger.debug(json.dumps(message))
+        logger.debug(json.dumps(message))
 
         table = message['table'] if 'table' in message else None
         action = message['action'] if 'action' in message else None
         try:
             if 'subscribe' in message:
                 if message['success']:
-                    self.logger.debug("Subscribed to %s." % message['subscribe'])
+                    logger.debug("Subscribed to %s." % message['subscribe'])
                 else:
                     self.error("Unable to subscribe to %s. Error: \"%s\" Please check and restart." %
                                (message['request']['args'][0], message['error']))
@@ -238,13 +239,13 @@ def __on_message(self, message):
                 # 'update'  - update row
                 # 'delete'  - delete row
                 if action == 'partial':
-                    self.logger.debug("%s: partial" % table)
+                    logger.debug("%s: partial" % table)
                     self.data[table] += message['data']
                     # Keys are communicated on partials to let you know how to uniquely identify
                     # an item. We use it for updates.
                     self.keys[table] = message['keys']
                 elif action == 'insert':
-                    self.logger.debug('%s: inserting %s' % (table, message['data']))
+                    logger.debug('%s: inserting %s' % (table, message['data']))
                     self.data[table] += message['data']
 
                     # Limit the max length of the table to avoid excessive memory usage.
@@ -253,7 +254,7 @@ def __on_message(self, message):
                         self.data[table] = self.data[table][(BitMEXWebsocket.MAX_TABLE_LEN // 2):]
 
                 elif action == 'update':
-                    self.logger.debug('%s: updating %s' % (table, message['data']))
+                    logger.debug('%s: updating %s' % (table, message['data']))
                     # Locate the item in the collection and update it.
                     for updateData in message['data']:
                         item = findItemByKeys(self.keys[table], self.data[table], updateData)
@@ -267,7 +268,7 @@ def __on_message(self, message):
                                 contExecuted = updateData['cumQty'] - item['cumQty']
                                 if contExecuted > 0:
                                     instrument = self.get_instrument(item['symbol'])
-                                    self.logger.info("Execution: %s %d Contracts of %s at %.*f" %
+                                    logger.info("Execution: %s %d Contracts of %s at %.*f" %
                                              (item['side'], contExecuted, item['symbol'],
                                               instrument['tickLog'], item['price']))
 
@@ -279,21 +280,21 @@ def __on_message(self, message):
                             self.data[table].remove(item)
 
                 elif action == 'delete':
-                    self.logger.debug('%s: deleting %s' % (table, message['data']))
+                    logger.debug('%s: deleting %s' % (table, message['data']))
                     # Locate the item in the collection and remove it.
                     for deleteData in message['data']:
                         item = findItemByKeys(self.keys[table], self.data[table], deleteData)
                         self.data[table].remove(item)
                 else:
                     raise Exception("Unknown action: %s" % action)
         except:
-            self.logger.error(traceback.format_exc())
+            logger.error(traceback.format_exc())
 
     def __on_open(self):
-        self.logger.debug("Websocket Opened.")
+        logger.debug("Websocket Opened.")
 
     def __on_close(self):
-        self.logger.info('Websocket Closed')
+        logger.info('Websocket Closed')
         self.exit()
 
     def __on_error(self, error):
@@ -318,16 +319,8 @@ def findItemByKeys(keys, table, matchData):
 
 if __name__ == "__main__":
     # create console handler and set level to debug
-    logger = logging.getLogger()
-    logger.setLevel(logging.DEBUG)
-    ch = logging.StreamHandler()
-    # create formatter
-    formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
-    # add formatter to ch
-    ch.setFormatter(formatter)
-    logger.addHandler(ch)
+    logger = log.setup_custom_logger('websocket', logging.DEBUG)
     ws = BitMEXWebsocket()
-    ws.logger = logger
     ws.connect("https://testnet.bitmex.com/api/v1")
     while(ws.ws.sock.connected):
         sleep(1)