Merge branch 'master' into major/3.0

Author: sentrivana

scripts/populate_tox/package_dependencies.jsonl

@@ -21,16 +21,12 @@ class SpanBatcher(Batcher["SpanJSON"]):
     # MAX_BEFORE_FLUSH should be lower than MAX_BEFORE_DROP, so that there is
     # a bit of a buffer for spans that appear between the trigger to flush
     # and actually flushing the buffer.
+    #
+    # The max limits are all per trace (per bucket).
     MAX_ENVELOPE_SIZE = 1000  # spans
-
     MAX_BEFORE_FLUSH = 1000
-    GLOBAL_MAX_BEFORE_FLUSH = 5_000
-
     MAX_BEFORE_DROP = 2000
-    GLOBAL_MAX_BEFORE_DROP = 10_000
-
     MAX_BYTES_BEFORE_FLUSH = 5 * 1024 * 1024  # 5 MB
-    GLOBAL_MAX_BYTES_BEFORE_FLUSH = 25 * 1024 * 1024  # 25 MB
 
     FLUSH_WAIT_TIME = 5.0
 
@@ -48,11 +44,7 @@ def __init__(
         # envelope.
         # trace_id -> span buffer
         self._span_buffer: dict[str, list["SpanJSON"]] = defaultdict(list)
-        self._span_number: int = 0
-
         self._running_size: dict[str, int] = defaultdict(lambda: 0)
-        self._total_running_size: int = 0
-
         self._capture_func = capture_func
         self._record_lost_func = record_lost_func
         self._running = True
@@ -79,11 +71,7 @@ def _reset_in_child() -> None:
 
     def _reset_thread_state(self) -> None:
         self._span_buffer = defaultdict(list)
-        self._span_number = 0
-
         self._running_size = defaultdict(lambda: 0)
-        self._total_running_size = 0
-
         self._running = True
 
         self._lock = threading.Lock()
@@ -106,12 +94,8 @@ def _flush_loop(self) -> None:
             self._flush(only_pending=True)
 
             if (
-                self._span_number >= self.GLOBAL_MAX_BEFORE_FLUSH
-                or self._total_running_size >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH
-                or (
-                    time.monotonic() - self._last_full_flush
-                    >= self.FLUSH_WAIT_TIME + jitter
-                )
+                time.monotonic() - self._last_full_flush
+                >= self.FLUSH_WAIT_TIME + jitter
             ):
                 self._flush()
                 self._last_full_flush = time.monotonic()
@@ -132,10 +116,8 @@ def add(self, span: "SpanJSON") -> None:
                 return None
 
             with self._lock:
-                if (
-                    self._span_number >= self.GLOBAL_MAX_BEFORE_DROP
-                    or len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_DROP
-                ):
+                size = len(self._span_buffer[span["trace_id"]])
+                if size >= self.MAX_BEFORE_DROP:
                     self._record_lost_func(
                         reason="queue_overflow",
                         data_category="span",
@@ -144,25 +126,17 @@ def add(self, span: "SpanJSON") -> None:
                     return None
 
                 self._span_buffer[span["trace_id"]].append(span)
-                self._span_number += 1
-
-                estimated_size = self._estimate_size(span)
-                self._running_size[span["trace_id"]] += estimated_size
-                self._total_running_size += estimated_size
+                self._running_size[span["trace_id"]] += self._estimate_size(span)
 
                 if (
-                    len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_FLUSH
+                    size + 1 >= self.MAX_BEFORE_FLUSH
                     or self._running_size[span["trace_id"]]
                     >= self.MAX_BYTES_BEFORE_FLUSH
                 ):
                     self._pending_flush.add(span["trace_id"])
                     notify = True
                 else:
-                    notify = (
-                        self._span_number >= self.GLOBAL_MAX_BEFORE_FLUSH
-                        or self._total_running_size
-                        >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH
-                    )
+                    notify = False
 
             if notify:
                 self._flush_event.set()
@@ -253,10 +227,7 @@ def _flush(self, only_pending: bool = False) -> None:
 
                     envelopes.append(envelope)
 
-                self._span_number -= len(self._span_buffer[bucket_id])
                 del self._span_buffer[bucket_id]
-
-                self._total_running_size -= self._running_size[bucket_id]
                 del self._running_size[bucket_id]
 
         for envelope in envelopes:

sentry_sdk/_span_batcher.py

@@ -13,6 +13,7 @@
 from typing import TYPE_CHECKING, Dict, List, cast, overload
 
 from sentry_sdk._compat import check_uwsgi_thread_support
+from sentry_sdk._log_batcher import LogBatcher
 from sentry_sdk._metrics_batcher import MetricsBatcher
 from sentry_sdk._span_batcher import SpanBatcher
 from sentry_sdk.consts import (
@@ -58,8 +59,6 @@
     get_type_name,
     handle_in_app,
     has_data_collection_enabled,
-    has_logs_enabled,
-    has_metrics_enabled,
     logger,
 )
 
@@ -629,22 +628,27 @@ def _record_lost_event(
 
             self.session_flusher = SessionFlusher(capture_func=_capture_envelope)
 
-            self.log_batcher = None
+            if self.options.get("enable_logs", False) or self.options[
+                "_experiments"
+            ].get("enable_logs", False):
+                logger.warning(
+                    "The enable_logs option has no effect and will be removed in the next major."
+                )
 
-            if has_logs_enabled(self.options):
-                from sentry_sdk._log_batcher import LogBatcher
+            self.log_batcher = LogBatcher(
+                capture_func=_capture_envelope,
+                record_lost_func=_record_lost_event,
+            )
 
-                self.log_batcher = LogBatcher(
-                    capture_func=_capture_envelope,
-                    record_lost_func=_record_lost_event,
+            if self.options.get("enable_metrics", True) is False:
+                logger.warning(
+                    "The enable_metrics option has no effect and will be removed in the next major."
                 )
 
-            self.metrics_batcher = None
-            if has_metrics_enabled(self.options):
-                self.metrics_batcher = MetricsBatcher(
-                    capture_func=_capture_envelope,
-                    record_lost_func=_record_lost_event,
-                )
+            self.metrics_batcher = MetricsBatcher(
+                capture_func=_capture_envelope,
+                record_lost_func=_record_lost_event,
+            )
 
             self.span_batcher = None
             if has_span_streaming_enabled(self.options):

sentry_sdk/client.py

@@ -36,6 +36,7 @@
     TransactionSource,
 )
 from sentry_sdk.tracing_utils import (
+    add_http_breadcrumb,
     add_http_request_source,
     has_span_streaming_enabled,
     should_propagate_trace,
@@ -378,57 +379,64 @@ async def on_request_start(
         with capture_internal_exceptions():
             parsed_url = parse_url(str(params.url), sanitize=False)
 
+        breadcrumb = {}
+
         span_name = "%s %s" % (
             method,
             parsed_url.url if parsed_url else SENSITIVE_DATA_SUBSTITUTE,
         )
 
-        span: "Union[Span, StreamedSpan, None]"
+        span: "Union[Span, StreamedSpan, None]" = None
         if has_span_streaming_enabled(client.options):
-            if sentry_sdk.traces.get_current_span() is None:
-                span = None
-            else:
-                attributes: "Attributes" = {
-                    "sentry.op": OP.HTTP_CLIENT,
-                    "sentry.origin": AioHttpIntegration.origin,
-                    "http.request.method": method,
-                }
-                if parsed_url is not None:
-                    if has_data_collection_enabled(client.options):
-                        url_full = parsed_url.url
-                        attributes["url.path"] = params.url.path
-
-                        if parsed_url.query:
-                            filtered_query = (
-                                _apply_data_collection_filtering_to_query_string(
-                                    query_string=parsed_url.query,
-                                    behaviour=client.options["data_collection"][
-                                        "url_query_params"
-                                    ],
-                                )
+            attributes: "Attributes" = {
+                "sentry.op": OP.HTTP_CLIENT,
+                "sentry.origin": AioHttpIntegration.origin,
+                "http.request.method": method,
+            }
+            if parsed_url is not None:
+                if has_data_collection_enabled(client.options):
+                    url_full = parsed_url.url
+                    attributes["url.path"] = params.url.path
+
+                    if parsed_url.query:
+                        filtered_query = (
+                            _apply_data_collection_filtering_to_query_string(
+                                query_string=parsed_url.query,
+                                behaviour=client.options["data_collection"][
+                                    "url_query_params"
+                                ],
                             )
-                            if filtered_query:
-                                attributes["url.query"] = filtered_query
-                                url_full += "?" + filtered_query
-
-                        if parsed_url.fragment:
-                            attributes["url.fragment"] = parsed_url.fragment
-                            url_full += "#" + parsed_url.fragment
-
-                        attributes["url.full"] = url_full
-                    elif should_send_default_pii():
-                        url_full = parsed_url.url
-                        attributes["url.path"] = params.url.path
-
-                        if parsed_url.query:
-                            url_full += "?" + parsed_url.query
-                            attributes["url.query"] = parsed_url.query
-                        if parsed_url.fragment:
-                            url_full += "#" + parsed_url.fragment
-                            attributes["url.fragment"] = parsed_url.fragment
-
-                        attributes["url.full"] = url_full
-
+                        )
+                        if filtered_query:
+                            attributes["url.query"] = filtered_query
+                            url_full += "?" + filtered_query
+                            breadcrumb[SPANDATA.HTTP_QUERY] = filtered_query
+
+                    if parsed_url.fragment:
+                        attributes["url.fragment"] = parsed_url.fragment
+                        url_full += "#" + parsed_url.fragment
+                        breadcrumb[SPANDATA.HTTP_FRAGMENT] = parsed_url.fragment
+
+                    attributes["url.full"] = url_full
+                    breadcrumb["url"] = url_full
+
+                elif should_send_default_pii():
+                    url_full = parsed_url.url
+                    attributes["url.path"] = params.url.path
+
+                    if parsed_url.query:
+                        url_full += "?" + parsed_url.query
+                        attributes["url.query"] = parsed_url.query
+                        breadcrumb[SPANDATA.HTTP_QUERY] = parsed_url.query
+                    if parsed_url.fragment:
+                        url_full += "#" + parsed_url.fragment
+                        attributes["url.fragment"] = parsed_url.fragment
+                        breadcrumb[SPANDATA.HTTP_FRAGMENT] = parsed_url.fragment
+
+                    attributes["url.full"] = url_full
+                    breadcrumb["url"] = url_full
+
+            if sentry_sdk.traces.get_current_span() is not None:
                 span = sentry_sdk.traces.start_span(
                     name=span_name, attributes=attributes
                 )
@@ -443,6 +451,13 @@ async def on_request_start(
                 legacy_span.set_data("url", parsed_url.url)
                 legacy_span.set_data(SPANDATA.HTTP_QUERY, parsed_url.query)
                 legacy_span.set_data(SPANDATA.HTTP_FRAGMENT, parsed_url.fragment)
+                breadcrumb.update(
+                    {
+                        SPANDATA.HTTP_QUERY: parsed_url.query,
+                        SPANDATA.HTTP_FRAGMENT: parsed_url.fragment,
+                        "url": parsed_url.url,
+                    }
+                )
             span = legacy_span
 
         if should_propagate_trace(client, str(params.url)):
@@ -465,19 +480,35 @@ async def on_request_start(
                 else:
                     params.headers[key] = value
 
-        trace_config_ctx.span = span
+        trace_config_ctx._sentry_span = span
+        trace_config_ctx._sentry_breadcrumb = breadcrumb
 
     async def on_request_end(
         session: "ClientSession",
         trace_config_ctx: "SimpleNamespace",
         params: "TraceRequestEndParams",
     ) -> None:
-        if trace_config_ctx.span is None:
-            return
-
-        span = trace_config_ctx.span
         status = int(params.response.status)
 
+        breadcrumb = getattr(trace_config_ctx, "_sentry_breadcrumb", None)
+        if breadcrumb is not None:
+            breadcrumb.update(
+                {
+                    SPANDATA.HTTP_METHOD: params.method.upper(),
+                    SPANDATA.HTTP_STATUS_CODE: status,
+                    "reason": params.response.reason,
+                }
+            )
+
+            add_http_breadcrumb(
+                status,
+                breadcrumb,
+            )
+
+        span = getattr(trace_config_ctx, "_sentry_span", None)
+        if span is None:
+            return
+
         if isinstance(span, StreamedSpan):
             span.set_attribute("http.response.status_code", status)
             span.status = (

sentry_sdk/integrations/aiohttp.py

@@ -609,14 +609,16 @@ def _set_output_data(
         set_on_span(SPANDATA.GEN_AI_RESPONSE_FINISH_REASONS, [finish_reason])
 
     client = sentry_sdk.get_client()
+    record_inputs = False
     record_outputs = False
     if has_data_collection_enabled(client.options):
-        if client.options["data_collection"]["gen_ai"]["outputs"]:
-            record_outputs = True
+        record_inputs = client.options["data_collection"]["gen_ai"]["inputs"]
+        record_outputs = client.options["data_collection"]["gen_ai"]["outputs"]
     elif should_send_default_pii() and integration.include_prompts:
+        record_inputs = True
         record_outputs = True
 
-    if record_outputs:
+    if record_inputs or record_outputs:
         output_messages: "dict[str, list[Any]]" = {
             "response": [],
             "tool": [],
@@ -628,15 +630,15 @@ def _set_output_data(
             elif output["type"] == "tool_use":
                 output_messages["tool"].append(output)
 
-        if len(output_messages["tool"]) > 0:
+        if record_inputs and len(output_messages["tool"]) > 0:
             set_data_normalized(
                 span,
                 SPANDATA.GEN_AI_RESPONSE_TOOL_CALLS,
                 output_messages["tool"],
                 unpack=False,
             )
 
-        if len(output_messages["response"]) > 0:
+        if record_outputs and len(output_messages["response"]) > 0:
             set_data_normalized(
                 span, SPANDATA.GEN_AI_RESPONSE_TEXT, output_messages["response"]
             )