Merge branch 'master' into antonpirker/fix/modified-decimal-context

Author: antonpirker

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## 2.25.1
+
+### Various fixes & improvements
+
+- fix(logs): Add a class which batches groups of logs together. (#4229) by @colin-sentry
+- fix(logs): Use repr instead of json for message and arguments (#4227) by @colin-sentry
+- fix(logs): Debug output from Sentry logs should always be `debug` level. (#4224) by @antonpirker
+- fix(ai): Do not consume anthropic streaming stop (#4232) by @colin-sentry
+- fix(spotlight): Do not spam sentry_sdk.warnings logger w/ Spotlight (#4219) by @BYK
+- fix(docs): fixed code snippet (#4218) by @antonpirker
+- build(deps): bump actions/create-github-app-token from 1.11.7 to 1.12.0 (#4214) by @dependabot
+
 ## 2.25.0
 
 ### Various fixes & improvements

docs/conf.py

@@ -31,7 +31,7 @@
 copyright = "2019-{}, Sentry Team and Contributors".format(datetime.now().year)
 author = "Sentry Team and Contributors"
 
-release = "2.25.0"
+release = "2.25.1"
 version = ".".join(release.split(".")[:2])  # The short X.Y version.
 
 

sentry_sdk/__init__.py

@@ -45,7 +45,7 @@
     "start_transaction",
     "trace",
     "monitor",
-    "_experimental_logger",
+    "logger",
 ]
 
 # Initialize the debug support after everything is loaded

sentry_sdk/_log_batcher.py

@@ -0,0 +1,142 @@
+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, safe_repr
+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:
+            # Recheck to make sure another thread didn't get here and start the
+            # the flusher in the meantime
+            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_event.clear()
+            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": safe_repr(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,9 +1,7 @@
-import json
 import os
 import uuid
 import random
 import socket
-import logging
 from collections.abc import Mapping
 from datetime import datetime, timezone
 from importlib import import_module
@@ -65,6 +63,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 +177,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 +375,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 +457,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 +875,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
@@ -900,59 +903,18 @@ def _capture_experimental_log(self, current_scope, log):
         # If debug is enabled, log the log to the console
         debug = self.options.get("debug", False)
         if debug:
-            severity_text_to_logging_level = {
-                "trace": logging.DEBUG,
-                "debug": logging.DEBUG,
-                "info": logging.INFO,
-                "warn": logging.WARNING,
-                "error": logging.ERROR,
-                "fatal": logging.CRITICAL,
-            }
-            logger.log(
-                severity_text_to_logging_level.get(log["severity_text"], logging.DEBUG),
-                f'[Sentry Logs] {log["body"]}',
+            logger.debug(
+                f'[Sentry Logs] [{log.get("severity_text")}] {log.get("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, {})
+        before_send_log = self.options["_experiments"].get("before_send_log")
+        if before_send_log is not None:
+            log = before_send_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)
+        if self.log_batcher:
+            self.log_batcher.add(log)
 
     def capture_session(
         self, session  # type: Session
@@ -1006,6 +968,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 +994,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,
     )
@@ -966,4 +966,4 @@ def _get_default_options():
 del _get_default_options
 
 
-VERSION = "2.25.0"
+VERSION = "2.25.1"

sentry_sdk/integrations/anthropic.py

@@ -184,8 +184,7 @@ def new_iterator():
                     input_tokens, output_tokens, content_blocks = _collect_ai_data(
                         event, input_tokens, output_tokens, content_blocks
                     )
-                    if event.type != "message_stop":
-                        yield event
+                    yield event
 
                 _add_ai_data_to_span(
                     span, integration, input_tokens, output_tokens, content_blocks
@@ -202,8 +201,7 @@ async def new_iterator_async():
                     input_tokens, output_tokens, content_blocks = _collect_ai_data(
                         event, input_tokens, output_tokens, content_blocks
                     )
-                    if event.type != "message_stop":
-                        yield event
+                    yield event
 
                 _add_ai_data_to_span(
                     span, integration, input_tokens, output_tokens, content_blocks