Improve logging and diagnostic information

* Within worker, add logs when a message is received and immediatly
  after a response has been sent.

* Include HOSTNAME in the test metadata that is stored in Redis. For a
  K8s deployment, HOSTNAME will correspond to the name of the pod
  running the test. This will make it possible to retreive worker logs
  for a misbehaving test.

* When a test is complete, the worker and the associated logs, may no
  longer exist, however the logs will still be available in the log file
  contained within the test payload that is pushed to long term storage.

* These changes pertain to worker, however there is an existing log
  message within the web implementation that logs when a message is sent to
  the worker, but no response is received. Additionally, each message
  between web and worker is given a unique ID, therefore with all of
  this togethor there will be breadcrumbs if a message is dropped.

Author: kbenne

docs/redis.md

@@ -3,23 +3,24 @@
 BOPTEST-Service currently uses Redis for two purposes.
 
 1. Storing test metadata.
-2. A pub/sub message bus to communicate with workers. 
+2. A pub/sub message bus to communicate with workers.
 
 ## Test Metadata:
 
 A hash under the key `tests:${testid}` is used to store test metadata.
 
-| Key                                      | Field       | Value
-| ---------------------------------------- | ----------- | -----------------------------
-| `tests:${testid}`                        | `status`    | `Queued \| Running`
-| `tests:${testid}`                        | `timestamp` | Epoch seconds when test is queued
-| `tests:${testid}`                        | `user`      | The OAuth user "sub", or undefined for anonymous tests
+| Key               | Field       | Value                                                                                    |
+| ----------------- | ----------- | ---------------------------------------------------------------------------------------- |
+| `tests:${testid}` | `status`    | `Queued \| Running`                                                                      |
+| `tests:${testid}` | `timestamp` | Epoch seconds when test is queued.                                                       |
+| `tests:${testid}` | `user`      | The OAuth user "sub", or undefined for anonymous tests.                                  |
+| `tests:${testid}` | `host`      | The value of HOSTNAME for the worker running the test. On K8s this will be the pod name. |
 
 Tests that are started by an authenticated user are stored in a set associated with the user.
 
-| Key                                      |  Value
-| ---------------------------------------- | -------------------------------------------
-| `users:${userSub}:tests`                 | `[testid1, testid2, ...]`
+| Key                      | Value                     |
+| ------------------------ | ------------------------- |
+| `users:${userSub}:tests` | `[testid1, testid2, ...]` |
 
 ## Pub/Sub Messages
 

worker/jobs/boptest_run_test/job.py

@@ -7,6 +7,7 @@
 import redis
 import numpy as np
 import msgpack
+import logging
 from boptest.lib.testcase import TestCase
 
 class Job:
@@ -19,6 +20,13 @@ def __init__(self, parameters):
         self.abort = False
         self.last_message_time = datetime.now()
 
+        log_level = os.environ.get("BOPTEST_LOGLEVEL", "INFO")
+        logging.getLogger().setLevel(log_level)
+        logging.getLogger('botocore').setLevel(log_level)
+        logging.getLogger('s3transfer').setLevel(log_level)
+        logging.getLogger('urllib3').setLevel(log_level)
+        self.logger = logging.getLogger('worker')
+
         self.redis = redis.Redis(host=os.environ["BOPTEST_REDIS_HOST"])
         self.redis_pubsub = self.redis.pubsub()
         if self.redis.hexists(self.testKey, "user"):
@@ -91,7 +99,7 @@ def run(self):
     def check_idle_time(self):
         idle_time = datetime.now() - self.last_message_time
         if idle_time.total_seconds() > self.timeout:
-            print("Testid '%s' is terminating due to inactivity." % self.testid)
+            self.logger.info("Testid '%s' is terminating due to inactivity." % self.testid)
             self.keep_running = False
 
     # Begin methods for message passing between web and worker ###
@@ -149,10 +157,14 @@ def process_messages(self):
                     method = message_data.get("method")
                     params = message_data.get("params")
 
+                    self.logger.info("Request ID:   '%s', with method '%s', was received" % (request_id, method))
+
                     callback_result = self.call_message_handler(method, params)
                     packed_result = self.pack({"requestID": request_id, "payload": callback_result})
                     self.redis.publish(response_channel, packed_result)
 
+                    self.logger.info("Response for, '%s', was sent" % request_id)
+
                     self.last_message_time = datetime.now()
         except Job.InvalidRequestError as e:
             payload = {"status": 400, "message": "Bad Request", "payload": str(e)}
@@ -161,7 +173,7 @@ def process_messages(self):
         except Exception:
             # Generic exceptions are an internal error
             error_message = str(sys.exc_info()[1])
-            print(error_message)
+            self.logger.error(error_message)
 
     # End methods for message passing
 
@@ -258,6 +270,8 @@ def reset(self, tarinfo):
 
     # cleanup after the simulation is stopped
     def cleanup(self):
+        self.logger.info("Test '%s' is complete" % self.testid)
+
         self.redis.delete(self.testKey)
         if self.userTestsKey:
             self.redis.srem(self.userTestsKey, self.testid)
@@ -294,3 +308,6 @@ def keys_to_camel_case(self, a_dict):
 
     def init_sim_status(self):
         self.redis.hset(self.testKey, "status", "Running")
+        host = os.getenv("HOSTNAME")
+        if host:
+            self.redis.hset(self.testKey, "host", host)

worker/logger.py

@@ -11,10 +11,12 @@ class Logger:
     """A logger specific for the tasks of the Worker"""
 
     def __init__(self):
-        logging.basicConfig(level=os.environ.get("BOPTEST_LOGLEVEL", "INFO"))
         self.logger = logging.getLogger('worker')
-        self.formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        self.logger.setLevel(os.environ.get("BOPTEST_LOGLEVEL", "INFO"))
+        fmt = '%(asctime)s UTC\t%(name)-20s%(levelname)s\t%(message)s'
+        datefmt = '%m/%d/%Y %I:%M:%S %p'
+        formatter = logging.Formatter(fmt,datefmt)
 
-        self.fh = logging.FileHandler('worker.log')
-        self.fh.setFormatter(self.formatter)
-        self.logger.addHandler(self.fh)
+        stream_handler = logging.StreamHandler()
+        stream_handler.setFormatter(formatter)
+        self.logger.addHandler(stream_handler)