feat: Syslog handler and formatter for the metricq_command

metricq/cli/syslog.py

@@ -0,0 +1,30 @@
+import logging
+import socket
+import time
+from logging.handlers import SysLogHandler
+
+
+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 get_syslog_handler(address: str) -> SysLogHandler:
+    if ":" in address:
+        ip, port = address.split(":")
+        return SysLogHandler(address=(ip, int(port)))
+    else:
+        return SysLogHandler(address=address)

metricq/cli/wrapper.py

@@ -1,13 +1,14 @@
 import logging
-from typing import Callable, Optional, cast
+from typing import Any, Callable, Optional, cast
 
 import click
 import click_log  # type: ignore
-from click import option
+from click import Context, option
 from dotenv import find_dotenv, load_dotenv
 
 from .. import get_logger
 from .params import MetricParam, TemplateStringParam
+from .syslog import SyslogFormatter, get_syslog_handler
 from .types import FC
 
 # We do not interpolate (i.e. replace ${VAR} with corresponding environment variables).
@@ -41,6 +42,31 @@ def metricq_token_option(default: str) -> Callable[[FC], FC]:
     )
 
 
+def metricq_syslog_option(required: Optional[bool]) -> Callable[[FC], FC]:
+    def enable_syslog(ctx: Context, param: Any | None, value: Optional[str]) -> None:
+        if required and value is None:
+            raise ValueError("Input syslog is missing")
+        if value is not None:
+            logger = get_logger()
+
+            program_name: str = "metricq_program"
+            metric = ctx.params.get("metric", None)
+
+            if metric is not None:
+                program_name = metric
+
+            handler = get_syslog_handler(value)
+            handler.setFormatter(SyslogFormatter(name=program_name))
+            logger.addHandler(handler)
+
+    return option(
+        "--syslog",
+        help="Enable syslog logging by specifying the syslog path or URL for the logger.",
+        callback=enable_syslog,
+        expose_value=False,
+    )
+
+
 def metricq_metric_option(
     default: Optional[str] = None, multiple: bool = False
 ) -> Callable[[FC], FC]:
@@ -71,7 +97,9 @@ def get_metric_command_logger() -> logging.Logger:
 
 
 def metricq_command(
-    default_token: str, client_version: str | None = None
+    default_token: str,
+    client_version: str | None = None,
+    syslog_required: bool | None = None,
 ) -> Callable[[FC], click.Command]:
     logger = get_metric_command_logger()
 
@@ -95,8 +123,10 @@ def decorator(func: FC) -> click.Command:
             log_decorator(
                 metricq_token_option(default_token)(
                     metricq_server_option()(
-                        click.command(context_settings=context_settings, epilog=epilog)(
-                            func
+                        metricq_syslog_option(required=syslog_required)(
+                            click.command(
+                                context_settings=context_settings, epilog=epilog
+                            )(func)
                         )
                     )
                 )