Modernise logging setup

Author: danpalmer

django_lightweight_queue/app_settings.py

@@ -11,6 +11,8 @@ def setting(suffix, default):
 WORKERS = setting('WORKERS', {})
 BACKEND = setting('BACKEND', 'django_lightweight_queue.backends.synchronous.SynchronousBackend')
 
+LOGGER_FACTORY = setting('LOGGER_FACTORY', 'logging.getLogger')
+
 # Allow per-queue overrides of the backend.
 BACKEND_OVERRIDES = setting('BACKEND_OVERRIDES', {})
 
@@ -35,8 +37,4 @@ def setting(suffix, default):
 # Workers will export metrics on this port, and ports following it
 PROMETHEUS_START_PORT = setting('PROMETHEUS_START_PORT', 9300)
 
-ENABLE_FLUENT_LOGGING = setting('ENABLE_FLUENT_LOGGING', False)
-FLUENT_HOST = setting('FLUENT_HOST', '127.0.0.1')
-FLUENT_PORT = setting('FLUENT_PORT', 24224)
-
 ATOMIC_JOBS = setting('ATOMIC_JOBS', True)

django_lightweight_queue/cron_scheduler.py

@@ -1,50 +1,32 @@
 import re
 import imp
 import time
-import logging
 import datetime
 import threading
 
 from django.apps import apps
 from django.core.management import call_command
 
 from .task import task
-from .utils import get_backend, configure_logging, \
-    contribute_implied_queue_name
+from .utils import get_logger, get_backend, contribute_implied_queue_name
 
 CRON_QUEUE_NAME = 'cron_scheduler'
 
 
 class CronScheduler(threading.Thread):
-    def __init__(self, log_level, log_filename, config):
-        self.log_level = log_level
-        self.log_filename = log_filename
+    def __init__(self, config):
         self.config = config
-
-        # Logfiles must be opened in child process
-        self.log = None
-
+        self.logger = get_logger('dlq.cron')
         super(CronScheduler, self).__init__(daemon=True)
 
     def run(self):
-        self.log = logging.getLogger()
-        for x in self.log.handlers:
-            self.log.removeHandler(x)
-
-        configure_logging(
-            level=self.log_level,
-            format='%%(asctime)-15s %%(process)d %s %%(levelname).1s: '
-                '%%(message)s' % (CRON_QUEUE_NAME,),
-            filename=self.log_filename,
-            extra={
-                'queue': CRON_QUEUE_NAME,
-            },
-        )
-
-        self.log.debug("Starting")
+        self.logger.debug("Starting cron thread")
 
         backend = get_backend(CRON_QUEUE_NAME)
-        self.log.info("Loaded backend %s", backend)
+        self.logger.info(
+            "Loaded backend {}".format(backend),
+            extra={'backend': backend},
+        )
 
         while True:
             # This will run until the process terminates.
@@ -55,7 +37,10 @@ def run(self):
             time.sleep((1 - time.time() % 1))
 
     def tick(self, backend):
-        self.log.debug("tick()")
+        self.logger.debug(
+            "Cron thread checking for work",
+            extra={'backend': backend},
+        )
 
         t = datetime.datetime.utcnow()
 
@@ -71,7 +56,13 @@ def tick(self, backend):
             ):
                 continue
 
-            self.log.info("Enqueueing %s", row['command'])
+            self.logger.debug(
+                "Enqueueing {}".format(row['command']),
+                extra={
+                    'target_queue': row['queue'],
+                    'command': row['command'],
+                },
+            )
 
             execute(
                 row['command'],
@@ -82,7 +73,13 @@ def tick(self, backend):
                 **row.get('command_kwargs', {})
             )
 
-            self.log.debug("Enqueued %s", row)
+            self.logger.info(
+                "Enqueued {}".format(row['command']),
+                extra={
+                    'target_queue': row['queue'],
+                    'command': row['command'],
+                },
+            )
 
 
 def get_cron_config():

django_lightweight_queue/management/commands/queue_runner.py

@@ -1,21 +1,17 @@
-import logging
 import daemonize
 
 from django.apps import apps
 from django.core.management.base import BaseCommand, CommandError
 
-from ...machine_types import PooledMachine, DirectlyConfiguredMachine
-from ...utils import get_backend, get_middleware, load_extra_config, \
-    configure_logging
+from ...utils import get_logger, get_backend, get_middleware, load_extra_config
 from ...runner import runner
+from ...machine_types import PooledMachine, DirectlyConfiguredMachine
 
 
 class Command(BaseCommand):
     def add_arguments(self, parser):
         parser.add_argument('--pidfile', action='store', dest='pidfile', default=None,
             help="Fork and write pidfile to this file.")
-        parser.add_argument('--logfile', action='store', dest='logfile', default=None,
-            help="Log to the specified file.")
         parser.add_argument('--touchfile', action='store', dest='touchfile', default=None,
             help="touch(1) the specified file after running a job.")
         parser.add_argument('--machine', action='store', dest='machine_number', default=None,
@@ -65,59 +61,34 @@ def validate_and_normalise(self, options):
                 )
 
     def handle(self, **options):
-        # Django < 1.8.3 leaves options['verbosity'] as a string so we cast to
-        # ensure an int.
-        verbosity = int(options['verbosity'])
-
-        level = {
-            0: logging.WARNING,
-            1: logging.INFO,
-            2: logging.DEBUG,
-        }[verbosity]
+        logger = get_logger('dlq.master')
 
         self.validate_and_normalise(options)
 
-        def log_filename(name):
-            try:
-                return options['logfile'] % name
-            except TypeError:
-                return options['logfile']
-
         def touch_filename(name):
             try:
                 return options['touchfile'] % name
             except TypeError:
                 return None
 
-        log_fd = configure_logging(
-            level=level,
-            format='%(asctime)-15s %(process)d %(levelname).1s: %(message)s',
-            filename=log_filename('master'),
-            extra={
-                'queue': 'master',
-            },
-        )
-
-        log = logging.getLogger()
-
         # Configuration overrides
         extra_config = options['config']
         if extra_config is not None:
             load_extra_config(extra_config)
 
-        log.info("Starting queue runner")
+        logger.info("Starting queue master")
 
         # Ensure children will be able to import our backend
         get_backend('dummy')
 
         get_middleware()
-        log.info("Loaded middleware")
+        logger.debug("Loaded middleware")
 
         # Ensure children will be able to import most things, but also try and
         # save memory by importing as much as possible before the fork() as it
         # has copy-on-write semantics.
         apps.get_models()
-        log.info("Loaded models")
+        logger.debug("Loaded models")
 
         if options['exact_configuration']:
             machine = DirectlyConfiguredMachine()
@@ -129,12 +100,7 @@ def touch_filename(name):
             )
 
         def run():
-            runner(
-                log,
-                log_filename,
-                touch_filename,
-                machine,
-            )
+            runner(touch_filename, machine, logger)
 
         # fork() only after we have started enough to catch failure, including
         # being able to write to our pidfile.
@@ -143,7 +109,6 @@ def run():
                 app='queue_runner',
                 pid=options['pidfile'],
                 action=run,
-                keep_fds=[log_fd],
             )
             daemon.start()
 

django_lightweight_queue/management/commands/queue_worker.py

@@ -1,7 +1,3 @@
-import sys
-import logging
-import argparse
-
 from django.core.management.base import BaseCommand
 
 from ...worker import Worker
@@ -25,18 +21,6 @@ def add_arguments(self, parser):
             type=int,
             help="port number on which to run Prometheus",
         )
-        parser.add_argument(
-            '--log-level',
-            choices=[x.lower() for x in logging._nameToLevel.keys()],
-            default='warning',
-            help="log level to set",
-        )
-        parser.add_argument(
-            '--log-file',
-            type=str,
-            dest='log_filename',
-            help="log destination",
-        )
         parser.add_argument(
             '--touch-file',
             type=str,
@@ -50,17 +34,13 @@ def handle(
         queue,
         number,
         prometheus_port,
-        log_level,
-        log_filename,
         touch_filename,
         **options
     ):
         worker = Worker(
             queue=queue,
             worker_num=number,
             prometheus_port=prometheus_port,
-            log_level=logging._nameToLevel[log_level.upper()],
-            log_filename=log_filename,
             touch_filename=touch_filename,
         )
         worker.run()

django_lightweight_queue/middleware/logging.py

@@ -1,43 +1,19 @@
-import json
 import logging
-import traceback
 
-log = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
-class LoggingMiddleware(object):
-    def process_job(self, job):
-        log.info("Running job %s", job)
 
-        self.fluent_log(job, state='processing')
+class LoggingMiddleware:
+    def process_job(self, job, queue, worker_num):
+        logger.info("Running job {}".format(job))
 
     def process_result(self, job, result, duration):
-        log.info("Finished job => %r (Time taken: %.2fs)",
+        logger.info("Finished job => {!r} (Time taken: {:.2f}s)".format(
             result,
             duration,
-        )
-
-        self.fluent_log(job, state='finished', duration=duration)
+        ))
 
     def process_exception(self, job, duration, *exc_info):
-        exception = ''.join(traceback.format_exception(*exc_info))
-
-        log.error("Exception when processing job (duration: %.2fs): %s",
+        logger.exception("Exception when processing job (duration: {:.2f}s)".format(
             duration,
-            exception,
-        )
-
-        self.fluent_log(
-            job,
-            state='exception',
-            duration=duration,
-            exception=exception,
-        )
-
-    def fluent_log(self, job, **kwargs):
-        data = job.as_dict()
-
-        data.update(kwargs)
-
-        data['fluent_log'] = True
-
-        log.info(json.dumps(data))
+        ))