feat: adding syslog wrapper

metricq/cli/__init__.py

@@ -10,6 +10,7 @@
     metricq_command,
     metricq_metric_option,
     metricq_server_option,
+    metricq_syslog,
     metricq_token_option,
 )
 
@@ -23,5 +24,6 @@
     "metricq_command",
     "metricq_metric_option",
     "metricq_server_option",
+    "metricq_syslog",
     "metricq_token_option",
 ]

metricq/cli/wrapper.py

@@ -1,10 +1,14 @@
 import logging
+import socket
+import time
+from logging.handlers import SysLogHandler
 from typing import Callable, Optional, cast
 
 import click
 import click_log  # type: ignore
 from click import option
 from dotenv import find_dotenv, load_dotenv
+from functools import wraps
 
 from .. import get_logger
 from .params import MetricParam, TemplateStringParam
@@ -104,3 +108,60 @@ def decorator(func: FC) -> click.Command:
         )
 
     return decorator
+
+
+class SyslogFormatter(logging.Formatter):
+    def __init__(self, *args, name: str = "metricq", **kwargs):  # type: ignore
+        super().__init__(*args, **kwargs)
+        self.program = name
+
+    def format(self, record: logging.LogRecord) -> str:
+        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(record.created))
+        hostname = socket.gethostname()
+        pid = record.process
+        program = self.program
+        # Custom Formatter based on rfc3164
+        # Format the header as "<PRI> TIMESTAMP HOSTNAME PROGRAM[PID]: MESSAGE"
+        # <PRI> is already beeing set by the SysLogHanlder, we only need to add the rest
+        syslog_header = f"{timestamp} {hostname} {program}[{pid}]: "
+
+        message = super().format(record)
+        return syslog_header + message
+
+
+def metricq_syslog(
+    required: bool = True, default: Optional[str] = None
+) -> Callable[[FC], FC]:
+    def get_syslog_handler(address: str) -> SysLogHandler:
+        if ":" in address:
+            ip, port = address.split(":")
+            return SysLogHandler(address=(ip, int(port)))
+        else:
+            return SysLogHandler(address=address)
+
+    def decorator(func):  # type: ignore
+        @click.option(
+            "--syslog",
+            default=default,
+            help="Setting this param will enable Syslog for the current program.",
+        )
+        @wraps(func)
+        def wrapper(*args, syslog, **kwargs):  # type: ignore
+            if required and syslog is None:
+                raise ValueError("Input syslog is missing")
+            if syslog is not None:
+                logger = get_logger()
+
+                program_name = "metricq_process"
+                if kwargs.get("token") is not None:
+                    program_name = str(kwargs.get("token"))
+
+                handler = get_syslog_handler(syslog)
+                handler.setFormatter(SyslogFormatter(name=program_name))
+                logger.addHandler(handler)
+
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator