SNOW-2044497 Add a snowflake log formatter (#56)

Author: sfc-gh-bnandibhatla

src/snowflake/telemetry/_internal/exporter/otlp/proto/logs/__init__.py

@@ -25,6 +25,7 @@
     encode_logs,
 )
 from snowflake.telemetry._internal.opentelemetry.proto.logs.v1.logs_marshaler import LogsData
+from snowflake.telemetry.logs import SnowflakeLogFormatter
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk._logs import export
 from opentelemetry.sdk import _logs
@@ -96,20 +97,6 @@ def __init__(
         )
         super().__init__(logger_provider=provider)
 
-    @staticmethod
-    def _get_snowflake_log_level_name(py_level_name):
-        """
-        Adapted from the getSnowflakeLogLevelName method in XP's logger.py
-        """
-        level = py_level_name.upper()
-        if level == "WARNING":
-            return "WARN"
-        if level == "CRITICAL":
-            return "FATAL"
-        if level == "NOTSET":
-            return "TRACE"
-        return level
-
     @staticmethod
     def _get_attributes(record: logging.LogRecord) -> types.Attributes:
         attributes = _logs.LoggingHandler._get_attributes(record) # pylint: disable=protected-access
@@ -124,9 +111,7 @@ def _get_attributes(record: logging.LogRecord) -> types.Attributes:
 
     def _translate(self, record: logging.LogRecord) -> _logs.LogRecord:
         otel_record = super()._translate(record)
-        otel_record.severity_text = SnowflakeLoggingHandler._get_snowflake_log_level_name(
-            record.levelname
-        )
+        otel_record.severity_text = SnowflakeLogFormatter.get_severity_text(record.levelname)
         return otel_record
 
 

src/snowflake/telemetry/logs/__init__.py

@@ -0,0 +1,106 @@
+#
+# Copyright (c) 2012-2025 Snowflake Computing Inc. All rights reserved.
+#
+
+import json
+import logging
+import traceback
+
+# skip natural LogRecord attributes
+# http://docs.python.org/library/logging.html#logrecord-attributes
+_RESERVED_ATTRS = frozenset(
+    (
+        "asctime",
+        "args",
+        "created",
+        "exc_info",
+        "exc_text",
+        "filename",
+        "funcName",
+        "message",
+        "levelname",
+        "levelno",
+        "lineno",
+        "module",
+        "msecs",
+        "msg",
+        "name",
+        "pathname",
+        "process",
+        "processName",
+        "relativeCreated",
+        "stack_info",
+        "thread",
+        "threadName",
+        "taskName",
+        # Params that Snowflake will populate and don't want users of this API to overwrite.
+        "code.lineno",
+        "code.function",
+        "code.filepath",
+        "exception.type",
+        "exception.message",
+        "exception.stacktrace",
+    )
+)
+
+_PY_LOG_LEVELS = frozenset(
+    (
+        "CRITICAL",
+        "ERROR",
+        "WARNING",
+        "INFO",
+        "DEBUG"
+    )
+)
+
+
+class SnowflakeLogFormatter(logging.Formatter):
+    """
+    A formatter to emit logs in JSON format so that they can be parsed by Snowflake.
+    """
+
+    @staticmethod
+    def get_severity_text(py_level_name):
+        """
+        Maps from Python logging level to OpenTelemetry's Severity Text.
+        """
+        level = py_level_name.upper()
+        if level not in _PY_LOG_LEVELS:
+            return "TRACE"
+        if level == "WARNING":
+            return "WARN"
+        if level == "CRITICAL":
+            return "FATAL"
+        return level
+
+    def format(self, record: logging.LogRecord) -> str:
+        log_items = {
+            "body": record.getMessage(),
+            "severity_text": self.get_severity_text(record.levelname),
+            "code.lineno": record.lineno,
+            "code.function": record.funcName,
+            "code.filepath": record.pathname
+        }
+
+        if record.exc_info is not None:
+            exctype, value, tb = record.exc_info
+            if exctype is not None:
+                log_items["exception.type"] = exctype.__name__
+            if value is not None and value.args:
+                log_items["exception.message"] = value.args[0]
+            if tb is not None:
+                log_items["exception.stacktrace"] = "".join(traceback.format_exception(*record.exc_info))
+            # Remove traceback to avoid emitting logs in incorrect format
+            record.exc_info = None
+
+        for attr, value in record.__dict__.items():
+            if attr in _RESERVED_ATTRS:
+                continue
+            log_items[attr] = value
+
+        return json.dumps(log_items)
+
+
+__all__ = [
+    "SnowflakeLogFormatter",
+]

tests/test_snowflake_log_formatter.py

@@ -0,0 +1,64 @@
+import json
+import logging
+import unittest
+from io import StringIO
+
+from snowflake.telemetry.logs import SnowflakeLogFormatter
+
+
+class TestSnowflakeLogFormatter(unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+        self.stream = StringIO()
+        self.root_logger = logging.getLogger("test")
+        self.root_hdlr = logging.StreamHandler(self.stream)
+        formatter = SnowflakeLogFormatter()
+        self.root_hdlr.setFormatter(formatter)
+        self.root_logger.addHandler(self.root_hdlr)
+
+    def test_normal_log(self):
+        """ Does the logger produce the correct output? """
+        self.root_logger.warning('foo', extra={"test": 123, "code.lineno": 35})
+        expected_log_message = {
+            "body": "foo",
+            "severity_text": "WARN",
+            "code.lineno": 21,
+            "code.function": "test_normal_log",
+            "test": 123
+        }
+        actual_log_message = json.loads(self.stream.getvalue())
+        actual_filepath = actual_log_message.pop("code.filepath")
+        self.assertIn("snowflake-telemetry-python/tests/test_snowflake_log_formatter.py", actual_filepath)
+        self.assertEqual(expected_log_message, actual_log_message)
+
+    def test_exception_log(self):
+        """ Does the logger include exception details? """
+        try:
+            10 / 0
+        except ZeroDivisionError:
+            self.root_logger.exception("\"test exception\"")
+
+        expected_log_message = {
+            "body": "\"test exception\"",
+            "severity_text": "ERROR",
+            "code.lineno": 39,
+            "code.function": "test_exception_log",
+            "exception.type": "ZeroDivisionError",
+            "exception.message": "division by zero",
+        }
+        actual_log_message = json.loads(self.stream.getvalue(), strict=False)
+        actual_log_message.pop("code.filepath")
+        actual_stacktrace = actual_log_message.pop("exception.stacktrace")
+        self.assertIn("line 37, in test_exception_log\n", actual_stacktrace)
+        self.assertIn("ZeroDivisionError: division by zero\n", actual_stacktrace)
+        self.assertEqual(expected_log_message, actual_log_message)
+
+    def test_get_severity_text(self):
+        """ Does get_severity_text work correctly? """
+        self.assertEqual("INFO", SnowflakeLogFormatter.get_severity_text("info"))
+        self.assertEqual("FATAL", SnowflakeLogFormatter.get_severity_text("critical"))
+        self.assertEqual("WARN", SnowflakeLogFormatter.get_severity_text("warning"))
+        self.assertEqual("DEBUG", SnowflakeLogFormatter.get_severity_text("debug"))
+        self.assertEqual("ERROR", SnowflakeLogFormatter.get_severity_text("error"))
+        self.assertEqual("TRACE", SnowflakeLogFormatter.get_severity_text("notset"))
+        self.assertEqual("TRACE", SnowflakeLogFormatter.get_severity_text("random"))