feat(loguru): Sentry logs for Loguru

CHANGELOG.md

@@ -122,7 +122,7 @@
   sentry_sdk.init(
     dsn="...",
     _experiments={
-        "enable_sentry_logs": True
+        "enable_logs": True
     }
     integrations=[
       LoggingIntegration(sentry_logs_level=logging.ERROR),

sentry_sdk/integrations/logging.py

@@ -363,10 +363,7 @@ def _capture_log_from_record(self, client, record):
                 for i, arg in enumerate(record.args):
                     attrs[f"sentry.message.parameter.{i}"] = (
                         arg
-                        if isinstance(arg, str)
-                        or isinstance(arg, float)
-                        or isinstance(arg, int)
-                        or isinstance(arg, bool)
+                        if isinstance(arg, (str, float, int, bool))
                         else safe_repr(arg)
                     )
         if record.lineno:

sentry_sdk/integrations/loguru.py

@@ -1,5 +1,6 @@
 import enum
 
+import sentry_sdk
 from sentry_sdk.integrations import Integration, DidNotEnable
 from sentry_sdk.integrations.logging import (
     BreadcrumbHandler,
@@ -11,12 +12,15 @@
 
 if TYPE_CHECKING:
     from logging import LogRecord
-    from typing import Optional, Any
+    from typing import Any, Optional, Tuple
 
 try:
     import loguru
     from loguru import logger
     from loguru._defaults import LOGURU_FORMAT as DEFAULT_FORMAT
+
+    if TYPE_CHECKING:
+        from loguru import Message
 except ImportError:
     raise DidNotEnable("LOGURU is not installed")
 
@@ -31,6 +35,10 @@ class LoggingLevels(enum.IntEnum):
     CRITICAL = 50
 
 
+DEFAULT_LEVEL = LoggingLevels.INFO.value
+DEFAULT_EVENT_LEVEL = LoggingLevels.ERROR.value
+
+
 SENTRY_LEVEL_FROM_LOGURU_LEVEL = {
     "TRACE": "DEBUG",
     "DEBUG": "DEBUG",
@@ -41,8 +49,22 @@ class LoggingLevels(enum.IntEnum):
     "CRITICAL": "CRITICAL",
 }
 
-DEFAULT_LEVEL = LoggingLevels.INFO.value
-DEFAULT_EVENT_LEVEL = LoggingLevels.ERROR.value
+
+def _loguru_level_to_otel(record_level):
+    # type: (int) -> Tuple[int, str]
+    for py_level, otel_severity_number, otel_severity_text in [
+        (LoggingLevels.CRITICAL, 21, "fatal"),
+        (LoggingLevels.ERROR, 17, "error"),
+        (LoggingLevels.WARNING, 13, "warn"),
+        (LoggingLevels.SUCCESS, 11, "info"),
+        (LoggingLevels.INFO, 9, "info"),
+        (LoggingLevels.DEBUG, 5, "debug"),
+        (LoggingLevels.TRACE, 1, "trace"),
+    ]:
+        if record_level >= py_level:
+            return otel_severity_number, otel_severity_text
+
+    return 0, "default"
 
 
 class LoguruIntegration(Integration):
@@ -52,19 +74,22 @@ class LoguruIntegration(Integration):
     event_level = DEFAULT_EVENT_LEVEL  # type: Optional[int]
     breadcrumb_format = DEFAULT_FORMAT
     event_format = DEFAULT_FORMAT
+    sentry_logs_level = DEFAULT_LEVEL  # type: Optional[int]
 
     def __init__(
         self,
         level=DEFAULT_LEVEL,
         event_level=DEFAULT_EVENT_LEVEL,
         breadcrumb_format=DEFAULT_FORMAT,
         event_format=DEFAULT_FORMAT,
+        sentry_logs_level=DEFAULT_LEVEL,
     ):
-        # type: (Optional[int], Optional[int], str | loguru.FormatFunction, str | loguru.FormatFunction) -> None
+        # type: (Optional[int], Optional[int], str | loguru.FormatFunction, str | loguru.FormatFunction, Optional[int]) -> None
         LoguruIntegration.level = level
         LoguruIntegration.event_level = event_level
         LoguruIntegration.breadcrumb_format = breadcrumb_format
         LoguruIntegration.event_format = event_format
+        LoguruIntegration.sentry_logs_level = sentry_logs_level
 
     @staticmethod
     def setup_once():
@@ -83,8 +108,23 @@ def setup_once():
                 format=LoguruIntegration.event_format,
             )
 
+        if LoguruIntegration.sentry_logs_level is not None:
+            logger.add(
+                loguru_sentry_logs_handler,
+                level=LoguruIntegration.sentry_logs_level,
+            )
+
 
 class _LoguruBaseHandler(_BaseHandler):
+    def __init__(self, *args, **kwargs):
+        # type: (*Any, **Any) -> None
+        if kwargs.get("level"):
+            kwargs["level"] = SENTRY_LEVEL_FROM_LOGURU_LEVEL.get(
+                kwargs.get("level", ""), DEFAULT_LEVEL
+            )
+
+        super().__init__(*args, **kwargs)
+
     def _logging_to_event_level(self, record):
         # type: (LogRecord) -> str
         try:
@@ -98,24 +138,70 @@ def _logging_to_event_level(self, record):
 class LoguruEventHandler(_LoguruBaseHandler, EventHandler):
     """Modified version of :class:`sentry_sdk.integrations.logging.EventHandler` to use loguru's level names."""
 
-    def __init__(self, *args, **kwargs):
-        # type: (*Any, **Any) -> None
-        if kwargs.get("level"):
-            kwargs["level"] = SENTRY_LEVEL_FROM_LOGURU_LEVEL.get(
-                kwargs.get("level", ""), DEFAULT_LEVEL
-            )
-
-        super().__init__(*args, **kwargs)
+    pass
 
 
 class LoguruBreadcrumbHandler(_LoguruBaseHandler, BreadcrumbHandler):
     """Modified version of :class:`sentry_sdk.integrations.logging.BreadcrumbHandler` to use loguru's level names."""
 
-    def __init__(self, *args, **kwargs):
-        # type: (*Any, **Any) -> None
-        if kwargs.get("level"):
-            kwargs["level"] = SENTRY_LEVEL_FROM_LOGURU_LEVEL.get(
-                kwargs.get("level", ""), DEFAULT_LEVEL
-            )
+    pass
 
-        super().__init__(*args, **kwargs)
+
+def loguru_sentry_logs_handler(message):
+    # type: (Message) -> None
+    # This is intentionally a callable sink instead of a standard logging handler
+    # since otherwise we wouldn't get direct access to message.record
+    client = sentry_sdk.get_client()
+
+    if not client.is_active():
+        return
+
+    if not client.options["_experiments"].get("enable_logs", False):
+        return
+
+    record = message.record
+
+    if (
+        LoguruIntegration.sentry_logs_level is None
+        or record["level"].no < LoguruIntegration.sentry_logs_level
+    ):
+        return
+
+    otel_severity_number, otel_severity_text = _loguru_level_to_otel(record["level"].no)
+
+    attrs = {"sentry.origin": "auto.logger.loguru"}  # type: dict[str, Any]
+
+    project_root = client.options["project_root"]
+    if record.get("file"):
+        if project_root is not None and record["file"].path.startswith(project_root):
+            attrs["code.file.path"] = record["file"].path[len(project_root) + 1 :]
+        else:
+            attrs["code.file.path"] = record["file"].path
+
+    if record.get("line") is not None:
+        attrs["code.line.number"] = record["line"]
+
+    if record.get("function"):
+        attrs["code.function.name"] = record["function"]
+
+    if record.get("thread"):
+        attrs["thread.name"] = record["thread"].name
+        attrs["thread.id"] = record["thread"].id
+
+    if record.get("process"):
+        attrs["process.pid"] = record["process"].id
+        attrs["process.executable.name"] = record["process"].name
+
+    if record.get("name"):
+        attrs["logger.name"] = record["name"]
+
+    client._capture_experimental_log(
+        {
+            "severity_text": otel_severity_text,
+            "severity_number": otel_severity_number,
+            "body": record["message"],
+            "attributes": attrs,
+            "time_unix_nano": int(record["time"].timestamp() * 1e9),
+            "trace_id": None,
+        }
+    )