From e2ff106f284eb6d60835af37fb02b8ccc5ba2a57 Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Mon, 22 Jun 2026 21:14:55 -0400 Subject: [PATCH] Filter noisy sentry traces/profiles (#834) * perf(reconciler): only trace Sentry transactions when work is found The reconciler enqueues sync_vectors every ~5 min per deriver instance. process_item wrapped every dequeued reconciler task in a single process_reconciler_task transaction, so idle cycles (the common case, where the cycle finds no rows and exits immediately) still created and sampled a transaction + profile, draining Sentry tracing/profiling quota. Remove the top-level transaction and push tracing into the sync batch helpers, starting a per-batch transaction only after rows are confirmed. Idle cycles now emit zero transactions; busy sweeps emit one smaller transaction per batch operation. Co-Authored-By: Claude Opus 4.8 (1M context) * perf(telemetry): drop infra/scrape transactions via a Sentry traces sampler Sentry was sampling every transaction at a flat traces_sample_rate with no sampler. The Prometheus /metrics scrape endpoint alone accounted for ~92% of all traced transactions (and their profiles), with /openapi.json and the deriver metrics server adding more pure noise. Add a traces_sampler that returns 0.0 for infra/scrape endpoints (/metrics, /health, /openapi.json, /docs, /redoc, and metrics/openapi transaction names) and the configured rate for real traffic. Sampling here (vs before_send_transaction) means dropped transactions are never recorded or profiled and the decision propagates to child spans. Shared init covers both the API server and the deriver worker. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- src/deriver/consumer.py | 25 +++--- src/reconciler/sync_vectors.py | 25 ++++-- src/telemetry/sentry.py | 56 ++++++++++++- tests/deriver/test_vector_reconciliation.py | 88 +++++++++++++++++++++ tests/telemetry/test_sentry_sampler.py | 60 ++++++++++++++ 5 files changed, 233 insertions(+), 21 deletions(-) create mode 100644 tests/telemetry/test_sentry_sampler.py diff --git a/src/deriver/consumer.py b/src/deriver/consumer.py index 8bbd9359..6e35d614 100644 --- a/src/deriver/consumer.py +++ b/src/deriver/consumer.py @@ -43,17 +43,20 @@ async def process_item(queue_item: models.QueueItem) -> None: # Handle reconciler first - it's the only task type that doesn't require workspace_name if task_type == "reconciler": - with sentry_sdk.start_transaction(name="process_reconciler_task", op="deriver"): - try: - validated = ReconcilerPayload(**queue_payload) - except ValidationError as e: - logger.error( - "Invalid reconciler payload received: %s. Payload: %s", - str(e), - queue_payload, - ) - raise ValueError(f"Invalid payload structure: {str(e)}") from e - await process_reconciler(validated) + # No top-level transaction here: reconciler tasks poll on a fixed + # interval and usually find no work. Tracing is started per-batch + # inside the reconciler only when actual work is found, so idle + # cycles don't consume Sentry tracing/profiling quota. + try: + validated = ReconcilerPayload(**queue_payload) + except ValidationError as e: + logger.error( + "Invalid reconciler payload received: %s. Payload: %s", + str(e), + queue_payload, + ) + raise ValueError(f"Invalid payload structure: {str(e)}") from e + await process_reconciler(validated) return # All other task types require a workspace_name diff --git a/src/reconciler/sync_vectors.py b/src/reconciler/sync_vectors.py index 00db6e3d..cdc18542 100644 --- a/src/reconciler/sync_vectors.py +++ b/src/reconciler/sync_vectors.py @@ -11,6 +11,7 @@ import time from dataclasses import dataclass from typing import Any, cast +import sentry_sdk from sqlalchemy import and_, delete, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import InstrumentedAttribute @@ -618,10 +619,13 @@ async def _reconcile_documents_batch( if not docs: return False - synced, failed = await _sync_documents(db, docs, external_vector_store) - metrics.documents_synced += synced - metrics.documents_failed += failed - await db.commit() + with sentry_sdk.start_transaction( + name="reconcile_documents_batch", op="reconciler" + ): + synced, failed = await _sync_documents(db, docs, external_vector_store) + metrics.documents_synced += synced + metrics.documents_failed += failed + await db.commit() return True @@ -639,10 +643,15 @@ async def _reconcile_message_embeddings_batch( if not embs: return False - synced, failed = await _sync_message_embeddings(db, embs, external_vector_store) - metrics.message_embeddings_synced += synced - metrics.message_embeddings_failed += failed - await db.commit() + with sentry_sdk.start_transaction( + name="reconcile_message_embeddings_batch", op="reconciler" + ): + synced, failed = await _sync_message_embeddings( + db, embs, external_vector_store + ) + metrics.message_embeddings_synced += synced + metrics.message_embeddings_failed += failed + await db.commit() return True diff --git a/src/telemetry/sentry.py b/src/telemetry/sentry.py index 1d16b1e7..4720c418 100644 --- a/src/telemetry/sentry.py +++ b/src/telemetry/sentry.py @@ -6,7 +6,7 @@ import inspect import logging from collections.abc import Callable, Sequence from functools import wraps -from typing import TYPE_CHECKING, ParamSpec, TypeVar, cast +from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast import sentry_sdk @@ -22,6 +22,56 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +# Paths whose transactions carry no debugging value but are hit constantly +# (health checks, Prometheus scrapes, OpenAPI schema, docs). Tracing them at the +# same rate as real traffic drowns the signal and burns tracing/profiling quota. +# Note: /docs and /redoc are disabled in production but are listed for safety. +_UNSAMPLED_PATHS = frozenset( + {"/metrics", "/health", "/openapi.json", "/docs", "/redoc"} +) + + +def _is_unsampled_transaction_name(name: str | None) -> bool: + """Match infra/scrape transactions by name. + + Fallback for transactions that don't expose an ASGI scope path (e.g. the + deriver's metrics server) or whose endpoint-style name encodes the route. + """ + if not name: + return False + return ( + name.endswith("openapi") + or name.endswith("metrics_endpoint") + or "prometheus.metrics" in name + ) + + +def traces_sampler(sampling_context: dict[str, Any]) -> float: + """Drop infra/scrape transactions; sample everything else at the default rate. + + Using a sampler (rather than ``before_send_transaction``) means dropped + transactions are never recorded or profiled, and the decision propagates to + child spans. ``SENTRY.TRACES_SAMPLE_RATE`` remains the rate for real traffic. + """ + asgi_scope = cast("dict[str, Any] | None", sampling_context.get("asgi_scope")) + if asgi_scope is not None and asgi_scope.get("path") in _UNSAMPLED_PATHS: + return 0.0 + + transaction_context = cast( + "dict[str, Any] | None", sampling_context.get("transaction_context") + ) + name = transaction_context.get("name") if transaction_context else None + if _is_unsampled_transaction_name(name if isinstance(name, str) else None): + return 0.0 + + # Respect an upstream sampling decision when continuing a distributed trace. + parent_sampled = sampling_context.get("parent_sampled") + if parent_sampled is not None: + return float(parent_sampled) + + return settings.SENTRY.TRACES_SAMPLE_RATE + + # Sentry SDK's default behavior: # - Captures INFO+ level logs as breadcrumbs # - Captures ERROR+ level logs as Sentry events @@ -44,7 +94,9 @@ def initialize_sentry( enable_tracing=True, release=settings.SENTRY.RELEASE, environment=settings.SENTRY.ENVIRONMENT, - traces_sample_rate=settings.SENTRY.TRACES_SAMPLE_RATE, + # traces_sampler supersedes traces_sample_rate; it returns the configured + # rate for real traffic and 0.0 for infra/scrape endpoints (see above). + traces_sampler=traces_sampler, profiles_sample_rate=settings.SENTRY.PROFILES_SAMPLE_RATE, before_send=before_send, integrations=integrations, diff --git a/tests/deriver/test_vector_reconciliation.py b/tests/deriver/test_vector_reconciliation.py index 1c004886..5e2e4712 100644 --- a/tests/deriver/test_vector_reconciliation.py +++ b/tests/deriver/test_vector_reconciliation.py @@ -6,6 +6,7 @@ message embeddings to the vector store, handling failures and retries. """ import datetime +from contextlib import asynccontextmanager from typing import cast from unittest.mock import AsyncMock, MagicMock, patch @@ -20,6 +21,7 @@ from src.reconciler.sync_vectors import ( ReconciliationMetrics, _get_documents_needing_sync, # pyright: ignore[reportPrivateUsage] _get_message_embeddings_needing_sync, # pyright: ignore[reportPrivateUsage] + _reconcile_documents_batch, # pyright: ignore[reportPrivateUsage] _reconcile_message_embeddings_batch, # pyright: ignore[reportPrivateUsage] _sync_documents, # pyright: ignore[reportPrivateUsage] _sync_message_embeddings, # pyright: ignore[reportPrivateUsage] @@ -993,6 +995,92 @@ class TestEndToEndReconciliation: mock_cleanup_docs.assert_awaited_once() +@pytest.mark.asyncio +class TestReconcilerTracing: + """A Sentry transaction is started only when a sync batch finds real work. + + Reconciler tasks poll on a fixed interval and usually find nothing; an idle + cycle must create zero transactions so it doesn't drain Sentry quota. + """ + + @staticmethod + def _fake_tracked_db(db: AsyncMock): + @asynccontextmanager + async def _cm(*_args: object, **_kwargs: object): + yield db + + return _cm + + async def test_no_transaction_when_no_embeddings_to_sync(self) -> None: + """The no-work path returns before starting a transaction.""" + metrics = ReconciliationMetrics() + with ( + patch( + "src.reconciler.sync_vectors.tracked_db", + self._fake_tracked_db(AsyncMock()), + ), + patch( + "src.reconciler.sync_vectors._get_message_embeddings_needing_sync", + new_callable=AsyncMock, + return_value=[], + ), + patch("src.reconciler.sync_vectors.sentry_sdk.start_transaction") as txn, + ): + worked = await _reconcile_message_embeddings_batch(None, metrics) + + assert worked is False + txn.assert_not_called() + + async def test_transaction_started_when_embeddings_present(self) -> None: + """A batch with real work starts its own named transaction.""" + metrics = ReconciliationMetrics() + with ( + patch( + "src.reconciler.sync_vectors.tracked_db", + self._fake_tracked_db(AsyncMock()), + ), + patch( + "src.reconciler.sync_vectors._get_message_embeddings_needing_sync", + new_callable=AsyncMock, + return_value=[MagicMock()], + ), + patch( + "src.reconciler.sync_vectors._sync_message_embeddings", + new_callable=AsyncMock, + return_value=(1, 0), + ), + patch("src.reconciler.sync_vectors.sentry_sdk.start_transaction") as txn, + ): + worked = await _reconcile_message_embeddings_batch(None, metrics) + + assert worked is True + assert metrics.message_embeddings_synced == 1 + txn.assert_called_once() + assert txn.call_args.kwargs.get("name") == "reconcile_message_embeddings_batch" + + async def test_no_transaction_when_no_documents_to_sync(self) -> None: + """The document batch also skips tracing when there is nothing to sync.""" + metrics = ReconciliationMetrics() + with ( + patch( + "src.reconciler.sync_vectors.tracked_db", + self._fake_tracked_db(AsyncMock()), + ), + patch( + "src.reconciler.sync_vectors._get_documents_needing_sync", + new_callable=AsyncMock, + return_value=[], + ), + patch("src.reconciler.sync_vectors.sentry_sdk.start_transaction") as txn, + ): + worked = await _reconcile_documents_batch( + MagicMock(spec=VectorStore), metrics + ) + + assert worked is False + txn.assert_not_called() + + def test_build_message_vector_record() -> None: """The shared vector-id/metadata builder: id is {message_id}_{position}, embeddings are coerced to float, metadata shape is fixed.""" diff --git a/tests/telemetry/test_sentry_sampler.py b/tests/telemetry/test_sentry_sampler.py new file mode 100644 index 00000000..c0d08a08 --- /dev/null +++ b/tests/telemetry/test_sentry_sampler.py @@ -0,0 +1,60 @@ +"""Tests for the Sentry traces sampler. + +The sampler must drop high-volume infra/scrape transactions (health checks, +Prometheus scrapes, OpenAPI schema, docs) while sampling real traffic at the +configured rate. These endpoints otherwise dominate transaction + profiling +volume and drown out useful traces. +""" + +import pytest + +from src.config import settings +from src.telemetry.sentry import traces_sampler + + +@pytest.mark.parametrize( + "path", + ["/metrics", "/health", "/openapi.json", "/docs", "/redoc"], +) +def test_infra_paths_are_dropped(path: str) -> None: + """ASGI requests to infra/scrape paths get a 0.0 sample rate.""" + assert traces_sampler({"asgi_scope": {"path": path}}) == 0.0 + + +@pytest.mark.parametrize( + "name", + [ + "src.telemetry.prometheus.metrics.metrics_endpoint", + "src.prometheus.metrics", + "fastapi.applications.FastAPI.setup..openapi", + ], +) +def test_infra_transaction_names_are_dropped(name: str) -> None: + """Transactions without an ASGI path still drop by their endpoint name.""" + assert traces_sampler({"transaction_context": {"name": name}}) == 0.0 + + +def test_real_route_uses_default_rate() -> None: + """A normal API route is sampled at the configured default rate.""" + ctx = { + "asgi_scope": {"path": "/v3/peers/alice/chat"}, + "transaction_context": {"name": "src.routers.peers.chat"}, + } + assert traces_sampler(ctx) == settings.SENTRY.TRACES_SAMPLE_RATE + + +def test_parent_sampling_decision_is_respected() -> None: + """When continuing a distributed trace, inherit the upstream decision.""" + assert traces_sampler({"parent_sampled": True}) == 1.0 + assert traces_sampler({"parent_sampled": False}) == 0.0 + + +def test_infra_path_overrides_parent_decision() -> None: + """Infra paths are dropped even if an upstream trace was sampled in.""" + ctx = {"asgi_scope": {"path": "/metrics"}, "parent_sampled": True} + assert traces_sampler(ctx) == 0.0 + + +def test_empty_context_falls_back_to_default_rate() -> None: + """A context with no scope, name, or parent uses the default rate.""" + assert traces_sampler({}) == settings.SENTRY.TRACES_SAMPLE_RATE