Logstash for DID finders

did_finder_cernopendata/pyproject.toml

@@ -9,6 +9,7 @@ packages = [{include = "did_finder_cernopendata", from = "src"}]
 [tool.poetry.dependencies]
 python = "~3.10"
 servicex-did-finder-lib = "^3.1.1"
+python-logstash = "^0.4.8"
 
 [tool.poetry.group.test]
 optional = true

did_finder_cernopendata/src/did_finder_cernopendata/__init__.py

@@ -0,0 +1,142 @@
+# Copyright (c) 2025, IRIS-HEP
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# * Redistributions of source code must retain the above copyright notice, this
+#   list of conditions and the following disclaimer.
+#
+# * Redistributions in binary form must reproduce the above copyright notice,
+#   this list of conditions and the following disclaimer in the documentation
+#   and/or other materials provided with the distribution.
+#
+# * Neither the name of the copyright holder nor the names of its
+#   contributors may be used to endorse or promote products derived from
+#   this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+import logging
+import os
+
+import logstash
+
+
+instance = os.environ.get("INSTANCE_NAME", "Unknown")
+
+
+class LogstashFormatter(logstash.formatter.LogstashFormatterBase):
+
+    def format(self, record):
+        message = {
+            "@timestamp": self.format_timestamp(record.created),
+            "@version": "1",
+            "message": record.getMessage(),
+            "path": record.pathname,
+            "tags": self.tags,
+            "type": self.message_type,
+            "instance": instance,
+            "component": "transformer sidecar",
+            # Extra Fields
+            "level": record.levelname,
+        }
+
+        # Add extra fields
+        message.update(self.get_extra_fields(record))
+
+        # If exception, add debug info
+        if record.exc_info:
+            message.update(self.get_debug_fields(record))
+
+        return self.serialize(message)
+
+
+class StreamFormatter(logging.Formatter):
+    """
+    A custom formatter that adds extras.
+    Normally log messages are "level instance component msg extra: {}"
+    """
+
+    def_keys = [
+        "name",
+        "msg",
+        "args",
+        "levelname",
+        "levelno",
+        "pathname",
+        "filename",
+        "module",
+        "exc_info",
+        "exc_text",
+        "stack_info",
+        "lineno",
+        "funcName",
+        "created",
+        "msecs",
+        "relativeCreated",
+        "thread",
+        "threadName",
+        "processName",
+        "process",
+        "message",
+    ]
+
+    def format(self, record: logging.LogRecord) -> str:
+        """
+        :param record: LogRecord
+        :return: formatted log message
+        """
+
+        string = super().format(record)
+        extra = {k: v for k, v in record.__dict__.items() if k not in self.def_keys}
+        if len(extra) > 0:
+            string += " extra: " + str(extra)
+        return string
+
+
+def initialize_logging(log=None, **kwargs):
+    """
+    Get a logger and initialize it so that it outputs the correct format
+    :param request: Request id to insert into log messages
+    :param log: optional logger to initialize
+    :return: logger with correct formatting that outputs to console
+    """
+
+    logging.basicConfig(level=logging.INFO, force=True)
+
+    if log is None:
+        log = logging.getLogger()
+
+    log.setLevel(logging.INFO)
+    stream_handler = logging.StreamHandler()
+    stream_formatter = StreamFormatter(
+        "%(levelname)s " + f"{instance} OpenData DID Finder " + "%(message)s"
+    )
+    stream_handler.setFormatter(stream_formatter)
+    stream_handler.setLevel(log.level)
+    log.addHandler(stream_handler)
+
+    logstash_host = os.environ.get("LOGSTASH_HOST")
+
+    if logstash_host:
+        logstash_port = int(os.environ.get("LOGSTASH_PORT", 5959))
+        logstash_handler = logstash.TCPLogstashHandler(
+            logstash_host, logstash_port, version=1
+        )
+        logstash_formatter = LogstashFormatter("logstash", None, None)
+        logstash_handler.setFormatter(logstash_formatter)
+        logstash_handler.setLevel(log.level)
+        log.addHandler(logstash_handler)
+
+    log.debug("Initialized logging")
+
+    return log

did_finder_cernopendata/src/did_finder_cernopendata/celery.py

@@ -26,7 +26,7 @@
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 import os
-import logging
+from . import initialize_logging
 from subprocess import PIPE, Popen, STDOUT
 from typing import Any, Dict, Generator
 
@@ -36,7 +36,7 @@
     LookupFailureException,
 )
 
-__log = logging.getLogger(__name__)
+__log = initialize_logging()
 
 cache_prefix = os.environ.get("CACHE_PREFIX", "")
 

did_finder_rucio/pyproject.toml

@@ -13,6 +13,7 @@ xmltodict = "^0.13.0"
 servicex-did-finder-lib = "^3.1.1"
 geoip2 = "^4.7.0"
 requests = ">=2.25.0,<3.0.0"
+python-logstash = "^0.4.8"
 
 [tool.poetry.group.test]
 optional = true

did_finder_rucio/src/rucio_did_finder/__init__.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2019, IRIS-HEP
+# Copyright (c) 2019-25, IRIS-HEP
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -25,3 +25,120 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+import logging
+import os
+
+import logstash
+import functools
+
+
+instance = os.environ.get("INSTANCE_NAME", "Unknown")
+
+
+class LogstashFormatter(logstash.formatter.LogstashFormatterBase):
+
+    def format(self, record):
+        message = {
+            "@timestamp": self.format_timestamp(record.created),
+            "@version": "1",
+            "message": record.getMessage(),
+            "path": record.pathname,
+            "tags": self.tags,
+            "type": self.message_type,
+            "instance": instance,
+            "component": "transformer sidecar",
+            # Extra Fields
+            "level": record.levelname,
+        }
+
+        # Add extra fields
+        message.update(self.get_extra_fields(record))
+
+        # If exception, add debug info
+        if record.exc_info:
+            message.update(self.get_debug_fields(record))
+
+        return self.serialize(message)
+
+
+class StreamFormatter(logging.Formatter):
+    """
+    A custom formatter that adds extras.
+    Normally log messages are "level instance component msg extra: {}"
+    """
+
+    def_keys = [
+        "name",
+        "msg",
+        "args",
+        "levelname",
+        "levelno",
+        "pathname",
+        "filename",
+        "module",
+        "exc_info",
+        "exc_text",
+        "stack_info",
+        "lineno",
+        "funcName",
+        "created",
+        "msecs",
+        "relativeCreated",
+        "thread",
+        "threadName",
+        "processName",
+        "process",
+        "message",
+    ]
+
+    def format(self, record: logging.LogRecord) -> str:
+        """
+        :param record: LogRecord
+        :return: formatted log message
+        """
+
+        string = super().format(record)
+        extra = {k: v for k, v in record.__dict__.items() if k not in self.def_keys}
+        if len(extra) > 0:
+            string += " extra: " + str(extra)
+        return string
+
+
+@functools.lru_cache
+def initialize_logging(log=None, **kwargs):
+    """
+    Get a logger and initialize it so that it outputs the correct format
+    :param request: Request id to insert into log messages
+    :param log: optional logger to initialize
+    :return: logger with correct formatting that outputs to console
+    """
+
+    logging.basicConfig(level=logging.INFO, force=True)
+
+    if log is None:
+        log = logging.getLogger()
+
+    log.setLevel(logging.INFO)
+    stream_handler = logging.StreamHandler()
+    stream_formatter = StreamFormatter(
+        "%(levelname)s " + f"{instance} Rucio DID finder " + "%(message)s"
+    )
+    stream_handler.setFormatter(stream_formatter)
+    stream_handler.setLevel(log.level)
+    log.addHandler(stream_handler)
+
+    logstash_host = os.environ.get("LOGSTASH_HOST")
+
+    if logstash_host:
+        logstash_port = int(os.environ.get("LOGSTASH_PORT", 5959))
+        logstash_handler = logstash.TCPLogstashHandler(
+            logstash_host, logstash_port, version=1
+        )
+        logstash_formatter = LogstashFormatter("logstash", None, None)
+        logstash_handler.setFormatter(logstash_formatter)
+        logstash_handler.setLevel(log.level)
+        log.addHandler(logstash_handler)
+
+    log.debug("Initialized logging")
+
+    return log

did_finder_rucio/src/rucio_did_finder/celery.py

@@ -25,7 +25,7 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-import logging
+from . import initialize_logging
 import os
 
 from rucio.client.didclient import DIDClient
@@ -36,7 +36,7 @@
 from servicex_did_finder_lib import DIDFinderApp
 from .replica_distance import ReplicaSorter
 
-__log = logging.getLogger(__name__)
+initialize_logging()
 
 cache_prefix = os.environ.get("CACHE_PREFIX", "")
 # Initialize the finder