Dispatcher log stdout (librenms#18953)

* Dispatcher service, allow specifying script output destination

* Style fixes

* style fix

* enum cleanup

* Add logger support

* Fix error if socket_timeout was not included

* fix incorrectly remove imports

* Only -q when log_output is none

* oopsie

* Fix fuckup

Author: murrant

LibreNMS/__init__.py

@@ -1,29 +1,30 @@
 import json
 import logging
 import os
+import re
 import signal
+import subprocess
 import sys
 import tempfile
 import threading
 import timeit
 from collections import deque
 from logging.handlers import RotatingFileHandler
-from math import ceil
 from queue import Queue
 from time import time
 
+from math import ceil
+
 from .command_runner import command_runner
 from .queuemanager import (
-    QueueManager,
-    TimedQueueManager,
+    AlertQueueManager,
     BillingQueueManager,
+    DiscoveryQueueManager,
     PingQueueManager,
-    ServicesQueueManager,
-    AlertQueueManager,
     PollerQueueManager,
-    DiscoveryQueueManager,
+    ServicesQueueManager,
 )
-from .service import Service, ServiceConfig
+from .service import Service, ServiceConfig, LogOutput
 
 # Hard limit script execution time so we don't get to "hang"
 DEFAULT_SCRIPT_TIMEOUT = 3600
@@ -190,13 +191,15 @@ def reset_signals():
     signal.signal(signal.SIGCHLD, signal.SIG_DFL)
 
 
-def call_script(script, args=()):
+def call_script(script, args=(), log_dest=None):
     """
     Run a LibreNMS script.  Captures all output returns exit code.
     Blocks parent signals (like SIGINT and SIGTERM).
     Kills script if it takes too long
     :param script: the name of the executable relative to the base directory
     :param args: a tuple of arguments to send to the command
+    :param log_dest: LogOutput enum or file path
+    :type log_dest: str | LogOutput | None
     :returns the output of the command
     """
     if script.endswith(".php"):
@@ -205,17 +208,61 @@ def call_script(script, args=()):
     else:
         base = ()
 
+    # preexec_fn=reset_signals ensures we don't receive signals from children (close_fds=True is default, but may become false in a future release)
+    kwargs = {
+        "preexec_fn": reset_signals,
+        "close_fds": True,
+        "start_new_session": True,
+        "timeout": DEFAULT_SCRIPT_TIMEOUT,
+    }
+
+    if isinstance(log_dest, str):
+        kwargs["stdout"] = log_dest
+        kwargs["stderr"] = log_dest
+    elif isinstance(log_dest, LogOutput):
+        if log_dest is LogOutput.FILE:
+            raise ValueError(
+                "For FILE mode, pass a file path string instead of LogOutput.FILE"
+            )
+
+        elif log_dest is LogOutput.NONE:
+            kwargs["stdout"] = subprocess.DEVNULL
+            kwargs["stderr"] = subprocess.DEVNULL
+
+        elif log_dest is LogOutput.PASSTHROUGH:
+            kwargs["stdout"] = sys.stdout
+            kwargs["stderr"] = sys.stderr
+
     base_dir = os.path.realpath(os.path.dirname(__file__) + "/..")
     cmd = base + ("{}/{}".format(base_dir, script),) + tuple(map(str, args))
     logger.debug("Running {}".format(cmd))
-    # preexec_fn=reset_signals ensures we don't receive signals from children (close_fds=True is default, but may become false in a future release)
-    return command_runner(
-        cmd,
-        preexec_fn=reset_signals,
-        close_fds=True,
-        start_new_session=True,
-        timeout=DEFAULT_SCRIPT_TIMEOUT,
-    )
+    exit_code, output = command_runner(cmd, **kwargs)
+
+    if log_dest is LogOutput.LOGGER:
+        if output:
+            for line in output.splitlines():
+                level = infer_log_level(line)
+                logger.log(level, line)
+
+    return exit_code, output
+
+
+def infer_log_level(line):
+    """Infer log level from a log line."""
+    upper_line = line.upper()
+
+    if re.match(r"^\s*(\[)?(CRITICAL|FATAL)(]|:|\s|-)", upper_line):
+        return logging.CRITICAL
+    elif re.match(r"^\s*(\[)?(ERROR|ERR)(]|:|\s|-)", upper_line):
+        return logging.ERROR
+    elif re.match(r"^\s*(\[)?(WARN|WARNING)(]|:|\s|-)", upper_line):
+        return logging.WARNING
+    elif re.match(r"^\s*(\[)?(INFO|INFORMATION)(]|:|\s|-)", upper_line):
+        return logging.INFO
+    elif re.match(r"^\s*(\[)?(DEBUG|TRACE)(]|:|\s|-)", upper_line):
+        return logging.DEBUG
+
+    return logging.INFO
 
 
 class DB:
@@ -472,9 +519,10 @@ def __init__(self, namespace="lock", sentinel_kwargs=None, **redis_kwargs):
             self._redis = redis.Redis(**kwargs)
         self._redis.ping()
         self._namespace = namespace
+        socket_timeout = redis_kwargs.get("socket_timeout")
         logger.debug(
             "Created redis lock manager with socket_timeout of {}s".format(
-                redis_kwargs["socket_timeout"]
+                socket_timeout if socket_timeout is not None else "default"
             )
         )
 

LibreNMS/queuemanager.py

@@ -375,12 +375,6 @@ def do_work(self, run_type, group):
                 "Error {} in {} billing:\n{}".format(exit_code, run_type, output)
             )
 
-        if self.config.log_output:
-            with open(
-                "{}/dispatch_billing-{}.log".format(self.config.logdir, run_type), "a"
-            ) as log_file:
-                log_file.write(output)
-
 
 class PingQueueManager(TimedQueueManager):
     def __init__(self, config, lock_manager):
@@ -415,15 +409,6 @@ def do_work(self, context, group):
                 )
                 exit_code, output = LibreNMS.call_script("lnms", args)
 
-                if self.config.log_output:
-                    with open(
-                        "{}/dispatch_group_{}_ping.log".format(
-                            self.config.logdir, group
-                        ),
-                        "a",
-                    ) as log_file:
-                        log_file.write(output)
-
                 if exit_code != 0:
                     logger.warning(
                         "Running fast ping for {} failed with error code {}: {}".format(
@@ -461,17 +446,17 @@ def do_dispatch(self):
     def do_work(self, device_id, group):
         if self.lock(device_id, timeout=self.config.services.frequency):
             logger.info("Checking services on device {}".format(device_id))
-            args = ("-d", "-h", device_id) if self.config.debug else ("-h", device_id)
-            exit_code, output = LibreNMS.call_script("check-services.php", args)
 
-            if self.config.log_output:
-                with open(
-                    "{}/dispatch_device_{}_services.log".format(
-                        self.config.logdir, device_id
-                    ),
-                    "a",
-                ) as log_file:
-                    log_file.write(output)
+            output = (
+                "{}/dispatch_device_{}_services.log".format(
+                    self.config.logdir, device_id
+                )
+                if self.config.log_output == LibreNMS.LogOutput.FILE
+                else self.config.log_output
+            )
+
+            args = ("-d", "-h", device_id) if self.config.debug else ("-h", device_id)
+            exit_code, output = LibreNMS.call_script("check-services.php", args, output)
 
             if exit_code == 0:
                 self.unlock(device_id)
@@ -511,15 +496,15 @@ def do_dispatch(self):
 
     def do_work(self, device_id, group):
         logger.info("Checking alerts")
-        args = ("-d", "-f") if self.config.debug else ("-f",)
-        exit_code, output = LibreNMS.call_script("alerts.php", args)
 
-        if self.config.log_output:
-            with open(
-                "{}/dispatch_alerts.log".format(self.config.logdir),
-                "a",
-            ) as log_file:
-                log_file.write(output)
+        output = (
+            "{}/dispatch_alerts.log".format(self.config.logdir)
+            if self.config.log_output == LibreNMS.LogOutput.FILE
+            else self.config.log_output
+        )
+
+        args = ("-d", "-f") if self.config.debug else ("-f",)
+        exit_code, output = LibreNMS.call_script("alerts.php", args, output)
 
         if exit_code != 0:
             if exit_code == 1:
@@ -544,21 +529,20 @@ def do_work(self, device_id, group):
         if self.lock(device_id, timeout=self.config.poller.frequency):
             logger.info("Polling device {}".format(device_id))
 
-            args = (
-                ("device:poll", device_id, "-vv")
-                if self.config.debug
-                else ("device:poll", device_id, "-q")
+            output = (
+                "{}/dispatch_device_{}_poller.log".format(self.config.logdir, device_id)
+                if self.config.log_output == LibreNMS.LogOutput.FILE
+                else self.config.log_output
             )
-            exit_code, output = LibreNMS.call_script("lnms", args)
 
-            if self.config.log_output:
-                with open(
-                    "{}/dispatch_device_{}_poller.log".format(
-                        self.config.logdir, device_id
-                    ),
-                    "a",
-                ) as log_file:
-                    log_file.write(output)
+            args_list = ["device:poll", device_id]
+            if self.config.debug:
+                args_list.append("-vv")
+            elif self.config.log_output is LibreNMS.LogOutput.NONE:
+                args_list.append("-q")
+            args = tuple(args_list)
+
+            exit_code, output = LibreNMS.call_script("lnms", args, output)
 
             if exit_code == 0:
                 self.unlock(device_id)
@@ -613,21 +597,22 @@ def do_work(self, device_id, group):
         ):
             logger.info("Discovering device {}".format(device_id))
 
-            args = (
-                ("device:discover", device_id, "-vv")
-                if self.config.debug
-                else ("device:discover", device_id, "-q")
+            output = (
+                "{}/dispatch_device_{}_discovery.log".format(
+                    self.config.logdir, device_id
+                )
+                if self.config.log_output == LibreNMS.LogOutput.FILE
+                else self.config.log_output
             )
-            exit_code, output = LibreNMS.call_script("lnms", args)
-
-            if self.config.log_output:
-                with open(
-                    "{}/dispatch_device_{}_discovery.log".format(
-                        self.config.logdir, device_id
-                    ),
-                    "a",
-                ) as log_file:
-                    log_file.write(output)
+
+            args_list = ["device:discover", device_id]
+            if self.config.debug:
+                args_list.append("-vv")
+            elif self.config.log_output is LibreNMS.LogOutput.NONE:
+                args_list.append("-q")
+            args = tuple(args_list)
+
+            exit_code, output = LibreNMS.call_script("lnms", args, output)
 
             if exit_code == 0:
                 self.unlock(device_id)

LibreNMS/service.py

@@ -1,10 +1,11 @@
 import logging
 import os
-import pymysql  # pylint: disable=import-error
 import sys
 import threading
 import time
 
+import pymysql  # pylint: disable=import-error
+
 import LibreNMS
 from LibreNMS.config import DBConfig
 
@@ -15,6 +16,7 @@
 
 from datetime import timedelta
 from datetime import datetime
+from enum import Enum
 from platform import python_version
 from time import sleep
 from socket import gethostname
@@ -38,6 +40,13 @@ class RedisConnectionError(Exception):
 logger = logging.getLogger(__name__)
 
 
+class LogOutput(Enum):
+    NONE = "none"
+    PASSTHROUGH = "passthrough"
+    LOGGER = "logger"
+    FILE = "file"
+
+
 class ServiceConfig(DBConfig):
     def __init__(self):
         """
@@ -101,7 +110,7 @@ def __init__(self, workers, frequency, calculate=None):
     redis_sentinel_service = None
     redis_timeout = 60
 
-    log_output = False
+    log_output = LogOutput.NONE
     logdir = "logs"
 
     watchdog_enabled = False
@@ -246,9 +255,11 @@ def populate(self):
         self.redis_timeout = int(
             os.getenv(
                 "REDIS_TIMEOUT",
-                self.alerting.frequency
-                if self.alerting.frequency != 0
-                else self.redis_timeout,
+                (
+                    self.alerting.frequency
+                    if self.alerting.frequency != 0
+                    else self.redis_timeout
+                ),
             )
         )
 
@@ -513,15 +524,21 @@ def start(self):
         )
         logger.info(
             "Queue Workers: Discovery={} Poller={} Services={} Alerting={} Billing={} Ping={}".format(
-                self.config.discovery.workers
-                if self.config.discovery.enabled
-                else "disabled",
-                self.config.poller.workers
-                if self.config.poller.enabled
-                else "disabled",
-                self.config.services.workers
-                if self.config.services.enabled
-                else "disabled",
+                (
+                    self.config.discovery.workers
+                    if self.config.discovery.enabled
+                    else "disabled"
+                ),
+                (
+                    self.config.poller.workers
+                    if self.config.poller.enabled
+                    else "disabled"
+                ),
+                (
+                    self.config.services.workers
+                    if self.config.services.enabled
+                    else "disabled"
+                ),
                 "enabled" if self.config.alerting.enabled else "disabled",
                 "enabled" if self.config.billing.enabled else "disabled",
                 "enabled" if self.config.ping.enabled else "disabled",