feat(tracing): Enable Sentry span streaming mode

.pre-commit-config.yaml

@@ -37,7 +37,7 @@ repos:
         name: validate-configs-syntax
         entry: env SNUBA_SETTINGS=test python3 -m snuba.validate_configs
         language: python
-        additional_dependencies: [ 'fastjsonschema', 'pyyaml', 'sentry_sdk' ]
+        additional_dependencies: [ 'fastjsonschema', 'pyyaml', 'sentry_sdk>=2.66.1' ]
         pass_filenames: false
         files: 'snuba/datasets/configuration/.*'
       - id: cargo-fmt

docs/source/profiler.rst

@@ -1,41 +1,10 @@
 Profiling
 =========
 
-Snuba has two ways of using `Sentry's own profiling
-<https://docs.sentry.io/product/profiling/>`_: Profiles captured as part of
-regular transactions like API calls, and "on-demand" profiles.
-
-Regular transaction profiling
-=============================
+Snuba uses `Sentry's own profiling
+<https://docs.sentry.io/product/profiling/>`_ to capture profiles as part of
+regular tracing.
 
 Only enabled for a few deployments selectively via environment variables. Snuba
 admin is sampled at 100%, and some low-scale consumers also have a sample rate
-set.
-
-On-demand profiling
-===================
-
-For consumers who have ``SNUBA_PROFILES_SAMPLE_RATE`` set to a nonzero value,
-it is possible to capture a profile of the main thread for a fixed duration of
-time via runtime config.
-
-For example, if the deployment ``snuba-querylog`` misbehaves, you can pick a
-specific pod by ``pod_name`` like
-``snuba-querylog-consumer-production-6d95f9c8d9-4cqht`` and enable profiling
-just for that one pod.
-
-Since profiling has unknown amount of overhead, pick only a few pods at a time
-to minimize the risk and size of a backlog.
-
-Once you have that ``pod_name``, set the runtime config
-``ondemand_profiler_hostnames`` to that value, without quotes. Separate
-multiple values by comma.
-
-To stop profiling and send the profile to Sentry, unset the value.
-
-You should see a warning message in Sentry saying "Starting ondemand profile
-for <hostname>", and once you unset runtime config, a new profile with a
-transaction name of "ondemand profile: <hostname>"
-
-Profiles are capped to 30 second runtime, so if the runtime config is set
-forever, there will be profiles sent to Sentry every 30 seconds.
+set via ``SNUBA_PROFILES_SAMPLE_RATE``.

sentry-options/schemas/snuba/schema.json

@@ -571,11 +571,6 @@
       "default": false,
       "description": "When true, the ClickHouse read path uses the clickhouse-connect (HTTP) driver instead of the native protocol. Lets the driver migration roll out and roll back without a deploy. Migrated from the runtime config of the same name."
     },
-    "ondemand_profiler_hostnames": {
-      "type": "string",
-      "default": "",
-      "description": "Comma-separated list of hostnames for which the on-demand profiler should capture a profile."
-    },
     "max_spans_per_transaction": {
       "type": "integer",
       "default": 2000,

snuba/cli/__init__.py

@@ -6,13 +6,14 @@
 from typing import Any
 
 import click
-import sentry_sdk
 import structlog
+from sentry_sdk import traces
 
 from snuba.core import initialize
 from snuba.environment import metrics as environment_metrics
 from snuba.environment import setup_logging, setup_sentry
 from snuba.utils.metrics.wrapper import MetricsWrapper
+from snuba.utils.sentry import SENTRY_OP
 
 setup_sentry()
 
@@ -46,7 +47,13 @@ def get_command(self, ctx: Any, name: str) -> click.Command:
         # by default. To mimic this behavior we have to do that here
         # since all of our infrastructure depends on this being the
         # case
-        with sentry_sdk.start_transaction(op="snuba_init", name=f"[cli init] {name}", sampled=True):
+        # parent_span=None => service span. No stream-mode sampled=True; obeys
+        # SENTRY_TRACE_SAMPLE_RATE. snuba_init_time metric below is unaffected.
+        with traces.start_span(
+            name=f"[cli init] {name}",
+            attributes={SENTRY_OP: "snuba_init"},
+            parent_span=None,
+        ):
             actual_command_name = name.replace("-", "_")
             ns: dict[str, click.Command] = {}
             # NOTE: WE initialize snuba before we compile the command code

snuba/cli/consumer.py

@@ -204,6 +204,7 @@ def consumer(
     logger.info("Consumer Starting")
 
     storage_key = StorageKey(storage_name)
+    # Tag (not attribute): consumers have no active span; tags reach errors.
     sentry_sdk.set_tag("storage", storage_name)
 
     logger.info("Checking Clickhouse connections...")

snuba/clickhouse/connect.py

@@ -11,9 +11,11 @@
 import clickhouse_connect
 import sentry_sdk
 from clickhouse_connect import common as clickhouse_connect_common
+from clickhouse_connect.driver.binding import quote_identifier
 from clickhouse_connect.driver.client import Client
 from clickhouse_connect.driver.exceptions import ClickHouseError, OperationalError
 from clickhouse_connect.driver.httputil import get_pool_manager
+from sentry_sdk import traces
 
 from snuba import environment, settings, state
 from snuba.clickhouse.errors import ClickhouseError
@@ -25,11 +27,37 @@
 )
 from snuba.reader import unwrap_nullable_type
 from snuba.utils.metrics.wrapper import MetricsWrapper
+from snuba.utils.sentry import SENTRY_OP
 
 logger = logging.getLogger("snuba.clickhouse.connect")
 
 metrics = MetricsWrapper(environment.metrics, "clickhouse.connect")
 
+
+def _driver_params(params: Params) -> Sequence[Any] | dict[str, Any] | None:
+    """Narrow ``Params`` to clickhouse-connect's accepted forms.
+
+    Falsy => None. Mappings are copied to a plain ``dict`` (driver wants an
+    invariant dict); sequences pass through as positional binds.
+    """
+    if not params:
+        return None
+    if isinstance(params, Mapping):
+        return dict(params)
+    return params
+
+
+def _insert_statement(table: str, column_names: Sequence[str]) -> str:
+    """Rebuild the INSERT statement clickhouse-connect sends on the wire.
+
+    ``client.insert`` takes a row matrix, not SQL, but still emits
+    ``INSERT INTO <table> (<cols>) FORMAT Native``. Row data is excluded
+    (unbounded, may hold PII).
+    """
+    columns = ", ".join(quote_identifier(name) for name in column_names)
+    return f"INSERT INTO {table} ({columns}) FORMAT Native"
+
+
 # Stand-in for "no read timeout" on the HTTP path. The native driver maps a
 # profile with no timeout (``None``) to an unbounded socket, but clickhouse-connect
 # cannot safely take ``None`` (its progress-interval computation does arithmetic
@@ -178,8 +206,8 @@ def _build_query_settings(
         settings: Mapping[str, Any] | None,
         query_id: str | None,
         capture_trace: bool,
-    ) -> Mapping[str, Any] | None:
-        query_settings = dict(settings) if settings else {}
+    ) -> dict[str, Any] | None:
+        query_settings: dict[str, Any] = dict(settings) if settings else {}
         if query_id is not None:
             query_settings["query_id"] = query_id
         if capture_trace:
@@ -207,13 +235,20 @@ def _execute_once(
         client = self._get_client()
         query_settings = self._build_query_settings(settings, query_id, capture_trace)
 
-        with sentry_sdk.start_span(description=query, op="db.clickhouse") as span:
-            span.set_data(sentry_sdk.consts.SPANDATA.DB_SYSTEM, "clickhouse")
-            span.set_data("query_id", query_id)
-            span.set_data("settings", query_settings)
+        with traces.start_span(
+            name="clickhouse query",
+            attributes={
+                SENTRY_OP: "db.clickhouse",
+                sentry_sdk.consts.SPANDATA.DB_SYSTEM: "clickhouse",
+                sentry_sdk.consts.SPANDATA.DB_QUERY_TEXT: query,
+            },
+        ) as span:
+            if query_id is not None:
+                span.set_attribute("query_id", query_id)
+            span.set_attribute("settings", json.dumps(query_settings, default=repr))
             query_result = client.query(
                 query,
-                parameters=params if params else None,
+                parameters=_driver_params(params),
                 settings=query_settings,
                 column_oriented=columnar,
             )
@@ -292,12 +327,19 @@ def execute_with_totals(
             if query_id is not None:
                 json_settings["query_id"] = query_id
 
-            with sentry_sdk.start_span(description=query, op="db.clickhouse") as span:
-                span.set_data(sentry_sdk.consts.SPANDATA.DB_SYSTEM, "clickhouse")
-                span.set_data("query_id", query_id)
+            with traces.start_span(
+                name="clickhouse query",
+                attributes={
+                    SENTRY_OP: "db.clickhouse",
+                    sentry_sdk.consts.SPANDATA.DB_SYSTEM: "clickhouse",
+                    sentry_sdk.consts.SPANDATA.DB_QUERY_TEXT: query,
+                },
+            ) as span:
+                if query_id is not None:
+                    span.set_attribute("query_id", query_id)
                 raw = client.raw_query(
                     query,
-                    parameters=params if params else None,
+                    parameters=_driver_params(params),
                     settings=json_settings,
                     fmt="JSONCompact",
                 )
@@ -445,11 +487,20 @@ def insert(
 
         with self._translate_clickhouse_errors():
             client = self._get_client()
-            with sentry_sdk.start_span(
-                description=f"INSERT INTO {table}", op="db.clickhouse"
+            with traces.start_span(
+                name=f"INSERT INTO {table}",
+                attributes={
+                    SENTRY_OP: "db.clickhouse",
+                    sentry_sdk.consts.SPANDATA.DB_SYSTEM: "clickhouse",
+                    sentry_sdk.consts.SPANDATA.DB_QUERY_TEXT: _insert_statement(
+                        table, column_names
+                    ),
+                },
             ) as span:
-                span.set_data(sentry_sdk.consts.SPANDATA.DB_SYSTEM, "clickhouse")
-                span.set_data("query_id", query_id)
+                span.set_attribute(
+                    "query_id",
+                    query_id if query_id is not None else "unknown-query-id",
+                )
                 client.insert(
                     table,
                     matrix,
@@ -506,8 +557,14 @@ def execute_explain(self, query: str) -> ClickhouseResult:
         """
         with self._translate_clickhouse_errors():
             client = self._get_client()
-            with sentry_sdk.start_span(description=query, op="db.clickhouse") as span:
-                span.set_data(sentry_sdk.consts.SPANDATA.DB_SYSTEM, "clickhouse")
+            with traces.start_span(
+                name="clickhouse query",
+                attributes={
+                    SENTRY_OP: "db.clickhouse",
+                    sentry_sdk.consts.SPANDATA.DB_SYSTEM: "clickhouse",
+                    sentry_sdk.consts.SPANDATA.DB_QUERY_TEXT: query,
+                },
+            ):
                 output = client.command(query)
             return self._explain_result(output)
 

snuba/clickhouse/native.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import logging
 import queue
 import re
@@ -20,6 +21,7 @@
 import sentry_sdk
 from clickhouse_driver import Client, errors
 from dateutil.tz import tz
+from sentry_sdk import traces
 from sentry_sdk.integrations.logging import ignore_logger
 
 from snuba import environment, settings
@@ -29,6 +31,7 @@
 from snuba.state.sentry_options import get_option
 from snuba.utils.metrics.gauge import ThreadSafeGauge
 from snuba.utils.metrics.wrapper import MetricsWrapper
+from snuba.utils.sentry import SENTRY_OP
 
 ignore_logger("clickhouse_driver.connection")
 
@@ -279,10 +282,17 @@ def execute(
                         )
 
                     def query_execute(conn: Client = conn, settings: Any = settings) -> Any:
-                        with sentry_sdk.start_span(description=query, op="db.clickhouse") as span:
-                            span.set_data(sentry_sdk.consts.SPANDATA.DB_SYSTEM, "clickhouse")
-                            span.set_data("query_id", query_id)
-                            span.set_data("settings", settings)
+                        with traces.start_span(
+                            name="clickhouse query",
+                            attributes={
+                                SENTRY_OP: "db.clickhouse",
+                                sentry_sdk.consts.SPANDATA.DB_SYSTEM: "clickhouse",
+                                sentry_sdk.consts.SPANDATA.DB_QUERY_TEXT: query,
+                            },
+                        ) as span:
+                            if query_id is not None:
+                                span.set_attribute("query_id", query_id)
+                            span.set_attribute("settings", json.dumps(settings, default=repr))
                             return conn.execute(
                                 query,
                                 params=params,

snuba/datasets/configuration/json_schema.py

@@ -4,9 +4,10 @@
 from typing import Any
 
 import fastjsonschema
-import sentry_sdk
+from sentry_sdk import traces
 
 from snuba import settings
+from snuba.utils.sentry import SENTRY_OP
 
 # Snubadocs are automatically generated from this file. When adding new schemas or individual keys,
 # please ensure you add a description key in the same level and succinctly describe the property.
@@ -864,17 +865,17 @@ def registered_class_array_schema(
 }
 
 if settings.VALIDATE_DATASET_YAMLS_ON_STARTUP:
-    with sentry_sdk.start_span(op="compile", description="Storage Validators"):
+    with traces.start_span(name="Storage Validators", attributes={SENTRY_OP: "compile"}):
         STORAGE_VALIDATORS = {
             "readable_storage": fastjsonschema.compile(V1_READABLE_STORAGE_SCHEMA),
             "writable_storage": fastjsonschema.compile(V1_WRITABLE_STORAGE_SCHEMA),
             "cdc_storage": fastjsonschema.compile(V1_CDC_STORAGE_SCHEMA),
         }
 
-    with sentry_sdk.start_span(op="compile", description="Entity Validators"):
+    with traces.start_span(name="Entity Validators", attributes={SENTRY_OP: "compile"}):
         ENTITY_VALIDATORS = {"entity": fastjsonschema.compile(V1_ENTITY_SCHEMA)}
 
-    with sentry_sdk.start_span(op="compile", description="Dataset Validators"):
+    with traces.start_span(name="Dataset Validators", attributes={SENTRY_OP: "compile"}):
         DATASET_VALIDATORS = {"dataset": fastjsonschema.compile(V1_DATASET_SCHEMA)}
 else:
     STORAGE_VALIDATORS = {}

snuba/datasets/configuration/loader.py

@@ -2,23 +2,26 @@
 
 from typing import Any
 
-import sentry_sdk
+from sentry_sdk import traces
 from yaml import safe_load
 
 from snuba import settings
+from snuba.utils.sentry import SENTRY_OP
 
 
 def load_configuration_data(path: str, validators: dict[str, Any]) -> dict[str, Any]:
     """
     Loads a configuration file from the given path
     Returns an untyped dict of dicts
     """
-    with sentry_sdk.start_span(op="load_and_validate") as span:
-        span.set_tag("file", path)
+    with traces.start_span(
+        name="load_and_validate",
+        attributes={SENTRY_OP: "load_and_validate", "file": path},
+    ) as span:
         with open(path) as file:
             config = safe_load(file)
         assert isinstance(config, dict)
         if settings.VALIDATE_DATASET_YAMLS_ON_STARTUP:
             validators[config["kind"]](config)
-        span.description = config["name"]
+        span.name = config["name"]
         return config

snuba/datasets/entities/factory.py

@@ -3,7 +3,7 @@
 from collections.abc import Sequence
 from glob import glob
 
-import sentry_sdk
+from sentry_sdk import traces
 
 from snuba import settings
 from snuba.datasets.configuration.entity_builder import build_entity_from_config
@@ -13,12 +13,13 @@
 from snuba.datasets.storages.factory import initialize_storage_factory
 from snuba.datasets.table_storage import TableWriter
 from snuba.utils.config_component_factory import ConfigComponentFactory
+from snuba.utils.sentry import SENTRY_OP
 from snuba.utils.serializable_exception import SerializableException
 
 
 class _EntityFactory(ConfigComponentFactory[Entity, EntityKey]):
     def __init__(self) -> None:
-        with sentry_sdk.start_span(op="initialize", description="Entity Factory"):
+        with traces.start_span(name="Entity Factory", attributes={SENTRY_OP: "initialize"}):
             initialize_storage_factory()
             self._entity_map: dict[EntityKey, PluggableEntity] = {}
             self._name_map: dict[type[Entity], EntityKey] = {}