feat(ourlogs): Batch logs into a single envelope

Author: colin-sentry

sentry_sdk/_log_batcher.py

@@ -0,0 +1,141 @@
+import json
+import os
+import random
+import threading
+from datetime import datetime, timezone
+from typing import Optional, List, Callable, TYPE_CHECKING, Any
+
+from sentry_sdk.utils import format_timestamp
+from sentry_sdk.envelope import Envelope
+
+if TYPE_CHECKING:
+    from sentry_sdk._types import Log
+
+
+class LogBatcher:
+    MAX_LOGS_BEFORE_FLUSH = 100
+    FLUSH_WAIT_TIME = 5.0
+
+    def __init__(
+        self,
+        capture_func,  # type: Callable[[Envelope], None]
+    ):
+        # type: (...) -> None
+        self._log_buffer = []  # type: List[Log]
+        self._capture_func = capture_func
+        self._running = True
+        self._lock = threading.Lock()
+
+        self._flush_event = threading.Event()  # type: threading.Event
+
+        self._flusher = None  # type: Optional[threading.Thread]
+        self._flusher_pid = None  # type: Optional[int]
+
+    def _ensure_thread(self):
+        # type: (...) -> bool
+        """For forking processes we might need to restart this thread.
+        This ensures that our process actually has that thread running.
+        """
+        if not self._running:
+            return False
+
+        pid = os.getpid()
+        if self._flusher_pid == pid:
+            return True
+
+        with self._lock:
+            # Double-checked locking
+            if self._flusher_pid == pid:
+                return True
+
+            self._flusher_pid = pid
+
+            self._flusher = threading.Thread(target=self._flush_loop)
+            self._flusher.daemon = True
+
+            try:
+                self._flusher.start()
+            except RuntimeError:
+                # Unfortunately at this point the interpreter is in a state that no
+                # longer allows us to spawn a thread and we have to bail.
+                self._running = False
+                return False
+
+        return True
+
+    def _flush_loop(self):
+        # type: (...) -> None
+        while self._running:
+            self._flush_event.wait(self.FLUSH_WAIT_TIME + random.random())
+            self._flush()
+
+    def add(
+        self,
+        log,  # type: Log
+    ):
+        # type: (...) -> None
+        if not self._ensure_thread() or self._flusher is None:
+            return None
+
+        with self._lock:
+            self._log_buffer.append(log)
+            if len(self._log_buffer) >= self.MAX_LOGS_BEFORE_FLUSH:
+                self._flush_event.set()
+
+    def kill(self):
+        # type: (...) -> None
+        if self._flusher is None:
+            return
+
+        self._running = False
+        self._flush_event.set()
+        self._flusher = None
+
+    def flush(self):
+        # type: (...) -> None
+        self._flush()
+
+    @staticmethod
+    def _log_to_otel(log):
+        # type: (Log) -> Any
+        def format_attribute(key, val):
+            # type: (str, int | float | str | bool) -> Any
+            if isinstance(val, bool):
+                return {"key": key, "value": {"boolValue": val}}
+            if isinstance(val, int):
+                return {"key": key, "value": {"intValue": str(val)}}
+            if isinstance(val, float):
+                return {"key": key, "value": {"doubleValue": val}}
+            if isinstance(val, str):
+                return {"key": key, "value": {"stringValue": val}}
+            return {"key": key, "value": {"stringValue": json.dumps(val)}}
+
+        otel_log = {
+            "severityText": log["severity_text"],
+            "severityNumber": log["severity_number"],
+            "body": {"stringValue": log["body"]},
+            "timeUnixNano": str(log["time_unix_nano"]),
+            "attributes": [
+                format_attribute(k, v) for (k, v) in log["attributes"].items()
+            ],
+        }
+
+        if "trace_id" in log:
+            otel_log["traceId"] = log["trace_id"]
+
+        return otel_log
+
+    def _flush(self):
+        # type: (...) -> Optional[Envelope]
+
+        envelope = Envelope(
+            headers={"sent_at": format_timestamp(datetime.now(timezone.utc))}
+        )
+        with self._lock:
+            for log in self._log_buffer:
+                envelope.add_log(self._log_to_otel(log))
+            self._log_buffer.clear()
+        if envelope.items:
+            self._capture_func(envelope)
+            return envelope
+        return None

sentry_sdk/client.py

@@ -1,4 +1,3 @@
-import json
 import os
 import uuid
 import random
@@ -65,6 +64,7 @@
     from sentry_sdk.session import Session
     from sentry_sdk.spotlight import SpotlightClient
     from sentry_sdk.transport import Transport
+    from sentry_sdk._log_batcher import LogBatcher
 
     I = TypeVar("I", bound=Integration)  # noqa: E741
 
@@ -178,6 +178,7 @@ def __init__(self, options=None):
         self.transport = None  # type: Optional[Transport]
         self.monitor = None  # type: Optional[Monitor]
         self.metrics_aggregator = None  # type: Optional[MetricsAggregator]
+        self.log_batcher = None  # type: Optional[LogBatcher]
 
     def __getstate__(self, *args, **kwargs):
         # type: (*Any, **Any) -> Any
@@ -375,6 +376,12 @@ def _capture_envelope(envelope):
                         "Metrics not supported on Python 3.6 and lower with gevent."
                     )
 
+            self.log_batcher = None
+            if experiments.get("enable_logs", False):
+                from sentry_sdk._log_batcher import LogBatcher
+
+                self.log_batcher = LogBatcher(capture_func=_capture_envelope)
+
             max_request_body_size = ("always", "never", "small", "medium")
             if self.options["max_request_body_size"] not in max_request_body_size:
                 raise ValueError(
@@ -451,6 +458,7 @@ def _capture_envelope(envelope):
         if (
             self.monitor
             or self.metrics_aggregator
+            or self.log_batcher
             or has_profiling_enabled(self.options)
             or isinstance(self.transport, BaseHttpTransport)
         ):
@@ -868,15 +876,11 @@ def capture_event(
 
     def _capture_experimental_log(self, current_scope, log):
         # type: (Scope, Log) -> None
-        logs_enabled = self.options["_experiments"].get("enable_sentry_logs", False)
+        logs_enabled = self.options["_experiments"].get("enable_logs", False)
         if not logs_enabled:
             return
         isolation_scope = current_scope.get_isolation_scope()
 
-        headers = {
-            "sent_at": format_timestamp(datetime.now(timezone.utc)),
-        }  # type: dict[str, object]
-
         environment = self.options.get("environment")
         if environment is not None and "sentry.environment" not in log["attributes"]:
             log["attributes"]["sentry.environment"] = environment
@@ -913,46 +917,13 @@ def _capture_experimental_log(self, current_scope, log):
                 f'[Sentry Logs] {log["body"]}',
             )
 
-        envelope = Envelope(headers=headers)
-
         before_emit_log = self.options["_experiments"].get("before_emit_log")
         if before_emit_log is not None:
             log = before_emit_log(log, {})
         if log is None:
             return
 
-        def format_attribute(key, val):
-            # type: (str, int | float | str | bool) -> Any
-            if isinstance(val, bool):
-                return {"key": key, "value": {"boolValue": val}}
-            if isinstance(val, int):
-                return {"key": key, "value": {"intValue": str(val)}}
-            if isinstance(val, float):
-                return {"key": key, "value": {"doubleValue": val}}
-            if isinstance(val, str):
-                return {"key": key, "value": {"stringValue": val}}
-            return {"key": key, "value": {"stringValue": json.dumps(val)}}
-
-        otel_log = {
-            "severityText": log["severity_text"],
-            "severityNumber": log["severity_number"],
-            "body": {"stringValue": log["body"]},
-            "timeUnixNano": str(log["time_unix_nano"]),
-            "attributes": [
-                format_attribute(k, v) for (k, v) in log["attributes"].items()
-            ],
-        }
-
-        if "trace_id" in log:
-            otel_log["traceId"] = log["trace_id"]
-
-        envelope.add_log(otel_log)  # TODO: batch these
-
-        if self.spotlight:
-            self.spotlight.capture_envelope(envelope)
-
-        if self.transport is not None:
-            self.transport.capture_envelope(envelope)
+        self.log_batcher.add(log)
 
     def capture_session(
         self, session  # type: Session
@@ -1006,6 +977,8 @@ def close(
             self.session_flusher.kill()
             if self.metrics_aggregator is not None:
                 self.metrics_aggregator.kill()
+            if self.log_batcher is not None:
+                self.log_batcher.kill()
             if self.monitor:
                 self.monitor.kill()
             self.transport.kill()
@@ -1030,6 +1003,8 @@ def flush(
             self.session_flusher.flush()
             if self.metrics_aggregator is not None:
                 self.metrics_aggregator.flush()
+            if self.log_batcher is not None:
+                self.log_batcher.flush()
             self.transport.flush(timeout=timeout, callback=callback)
 
     def __enter__(self):

sentry_sdk/consts.py

@@ -78,7 +78,7 @@ class CompressionAlgo(Enum):
                 Callable[[str, MetricValue, MeasurementUnit, MetricTags], bool]
             ],
             "metric_code_locations": Optional[bool],
-            "enable_sentry_logs": Optional[bool],
+            "enable_logs": Optional[bool],
         },
         total=False,
     )

sentry_sdk/integrations/logging.py

@@ -348,7 +348,7 @@ def emit(self, record):
             if not client.is_active():
                 return
 
-            if not client.options["_experiments"].get("enable_sentry_logs", False):
+            if not client.options["_experiments"].get("enable_logs", False):
                 return
 
             SentryLogsHandler._capture_log_from_record(client, record)

sentry_sdk/logger.py

@@ -17,7 +17,11 @@ def _capture_log(severity_text, severity_number, template, **kwargs):
     if "attributes" in kwargs:
         attrs.update(kwargs.pop("attributes"))
     for k, v in kwargs.items():
-        attrs[f"sentry.message.parameters.{k}"] = v if isinstance(v, str) else repr(v)
+        attrs[f"sentry.message.parameters.{k}"] = (
+            v
+            if (isinstance(v, str) or isinstance(v, int) or isinstance(v, bool))
+            else repr(v)
+        )
 
     # noinspection PyProtectedMember
     client._capture_experimental_log(