chore: restructure common logs

tubesync/common/logger.py

@@ -1,22 +1,32 @@
 import logging
 from django.conf import settings
-##from .logging import default_handler, syslog_handler
-from .utils import getenv
+from .logs import app_logger, default_handler
+##from .logs.syslog.std import default_handler as syslog_default_handler
+from .logs.syslog.hat import (
+    default_handler as hat_syslog_default_handler,
+    handler as hat_syslog_handler,
+)
 
 
-##if settings.DEBUG:
-##    default_handler.setLevel(logging.DEBUG)
+log = app_logger
 
-
-app_name = getenv('DJANGO_SETTINGS_MODULE')
-first_part = app_name.split('.', 1)[0]
-log = app_logger = logging.getLogger(first_part)
-##app_logger.propagate = False
-##app_logger.addHandler(default_handler)
-##app_logger.addHandler(syslog_handler)
-app_logger.setLevel(logging.INFO)
+default_handler.setLevel(logging.INFO)
 if settings.DEBUG:
-    app_logger.setLevel(logging.DEBUG)
+    default_handler.setLevel(logging.DEBUG)
+
+hat_syslog_tcp_handler = hat_syslog_handler(
+    host=hat_syslog_default_handler.host,
+    port=hat_syslog_default_handler.port,
+    comm_type='TCP',
+)
+hat_syslog_tcp_handler.setLevel(logging.DEBUG)
+##if not settings.DEBUG:
+##    hat_syslog_tcp_handler.setLevel(logging.INFO)
+
+app_logger.propagate = False
+app_logger.addHandler(default_handler)
+##app_logger.addHandler(syslog_default_handler)
+app_logger.addHandler(hat_syslog_tcp_handler)
 
 if (
     hasattr(settings, 'DATABASES') and

tubesync/common/logs/__init__.py

@@ -0,0 +1,16 @@
+from . import syslog
+from ._default import default_formatter, default_handler
+from ._filters import RemoveSpecificLogFilter
+from ._logger import app_logger, logger
+
+
+logger = logger(__name__)
+
+__all__ = [
+    'app_logger',
+    'default_formatter',
+    'default_handler',
+    'logger',
+    'syslog',
+    'RemoveSpecificLogFilter',
+]

tubesync/common/logs/_default.py

@@ -0,0 +1,12 @@
+import logging
+
+
+default_formatter = logging.Formatter(
+    '%(asctime)s [%(name)s/%(levelname)s] %(message)s'
+)
+
+default_handler = logging.StreamHandler()
+default_handler.setFormatter(default_formatter)
+default_handler.setLevel(logging.INFO)
+
+__all__ = ['default_formatter', 'default_handler']

tubesync/common/logging.py → tubesync/common/logs/_filters.py

@@ -1,5 +1,4 @@
 import logging
-from logging.handlers import SysLogHandler
 
 
 class RemoveSpecificLogFilter(logging.Filter):
@@ -69,21 +68,7 @@ def filter(self, record):
         return False
 
 
-default_formatter = logging.Formatter(
-    '%(asctime)s [%(name)s/%(levelname)s] %(message)s'
-)
-default_handler = logging.StreamHandler()
-default_handler.setFormatter(default_formatter)
-default_handler.setLevel(logging.INFO)
-
-syslog_formatter = logging.Formatter(
-    '%(asctime)s %(name)s: %(message)s',
-    '%b %d %H:%M:%S',
-)
-syslog_handler = SysLogHandler(
-    address='/dev/log',
-    facility=SysLogHandler.LOG_LOCAL0,
-)
-syslog_handler.setFormatter(syslog_formatter)
-syslog_handler.setLevel(logging.DEBUG)
+__all__ = [
+    'RemoveSpecificLogFilter',
+]
 

tubesync/common/logs/_logger.py

@@ -0,0 +1,16 @@
+import logging
+import os
+
+
+logger = lambda name=None: logging.getLogger(
+    __name__.rsplit('.', 1)[0] if name is None else name
+)
+
+app_logger = logger()
+app_name = os.getenv('DJANGO_SETTINGS_MODULE', str()).strip()
+if app_name:
+    first_part = app_name.split('.', 1)[0]
+    app_logger = logger(first_part)
+app_logger.setLevel(logging.DEBUG)
+
+__all__ = ['app_logger', 'logger']

tubesync/common/logs/syslog/__init__.py

@@ -0,0 +1,2 @@
+from . import hat as hat
+from . import std as std

tubesync/common/logs/syslog/hat/__init__.py

@@ -0,0 +1,3 @@
+from ._default import *  # noqa: F403
+from ._logger import logger as logger
+logger = logger(__name__)

tubesync/common/logs/syslog/hat/__main__.py

@@ -0,0 +1,106 @@
+import contextlib
+import logging
+import socket
+import threading
+import time
+import unittest
+
+from ._default import handler
+
+
+class MockSyslogServer:
+    """Stands up an isolated local background socket server to harvest transport streams."""
+    def __init__(self, host='127.0.0.1', port=0):
+        self.host = host
+        self.port = port
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.sock.bind((self.host, self.port))
+        self.port = self.sock.getsockname()[1]
+
+        self.received_messages = []
+        self.running = threading.Event()
+        self._thread = None
+
+    def start(self):
+        self.running.set()
+        self._thread = threading.Thread(target=self._listen_loop, daemon=True)
+        self._thread.start()
+
+    def stop(self):
+        self.running.clear()
+        s = None
+        try:
+            s = socket.create_connection((self.host, self.port), timeout=0.1)
+        except Exception:
+            pass
+        finally:
+            if s is not None:
+                s.close()
+        if self._thread:
+            self._thread.join(timeout=1.0)
+        self.sock.close()
+
+    def _listen_loop(self):
+        self.sock.listen(1)
+        while self.running.is_set():
+            try:
+                conn, _ = self.sock.accept()
+                if not self.running.is_set():
+                    conn.close()
+                    break
+
+                with conn:
+                    while self.running.is_set():
+                        data = conn.recv(4096)
+                        if not data:
+                            break
+                        self.received_messages.append(data.decode('utf-8'))
+            except Exception:
+                break
+
+class TestSyslogHandlerIntegration(unittest.TestCase):
+    def setUp(self):
+        """Initializes the background mock network collection service before running assertions."""
+        self.server = MockSyslogServer()
+        self.server.start()
+
+        self.handler = handler(
+            host=self.server.host,
+            port=self.server.port,
+            comm_type='tcp',
+            queue_size=10,
+            reconnect_delay=1,
+        )
+
+        self.test_logger = logging.getLogger('integration_test')
+        self.test_logger.setLevel(logging.DEBUG)
+        self.test_logger.addHandler(self.handler)
+
+    def tearDown(self):
+        """Cleans up the network service topology profiles safely upon validation teardown."""
+        with contextlib.suppress(Exception):
+            self.handler.close()
+        self.server.stop()
+
+    def test_pipeline_delivery_and_flush(self):
+        """Verifies that items are completely delivered down the wire before flush unblocks."""
+        self.test_logger.debug('Message A')
+        self.test_logger.info('Message B')
+
+        start_time = time.monotonic()
+        self.handler.flush()
+        elapsed = time.monotonic() - start_time
+
+        self.assertLess(elapsed, 2.0, 'The flush operations deadlocked the execution loop context')
+        self.assertTrue(any('Message A' in msg for msg in self.server.received_messages))
+        self.assertTrue(any('Message B' in msg for msg in self.server.received_messages))
+
+    def test_graceful_close_lifecycle(self):
+        """Confirms that close drains remaining log states and tears down the worker thread."""
+        self.test_logger.info('Shutdown Message')
+        self.handler.close()
+        self.assertTrue(any('Shutdown Message' in msg for msg in self.server.received_messages))
+
+unittest.main()