diff --git a/src/sentry/dynamic_sampling/per_org/gate.py b/src/sentry/dynamic_sampling/per_org/gate.py index 499bf370f51b..670d0961b498 100644 --- a/src/sentry/dynamic_sampling/per_org/gate.py +++ b/src/sentry/dynamic_sampling/per_org/gate.py @@ -17,6 +17,9 @@ SLIDING_WINDOW_COMPARISON_ORG_IDS_OPTION = ( "dynamic-sampling.per_org.sliding-window-comparison-org-ids" ) +CAP_DSC_TRANSACTION_LENGTH_ORG_IDS_OPTION = ( + "dynamic-sampling.per_org.cap-dsc-transaction-length-org-ids" +) SAMPLE_RATES_SUMMARY_LOG_ROLLOUT_RATE_OPTION = ( "dynamic-sampling.per_org.sample-rates-summary-log-rollout-rate" ) @@ -78,8 +81,16 @@ def transaction_volume_debug_project_ids() -> set[int]: def sliding_window_comparison_org_ids() -> set[int]: + return _org_ids_option(SLIDING_WINDOW_COMPARISON_ORG_IDS_OPTION) + + +def is_org_in_cap_dsc_transaction_length_rollout(org_id: int) -> bool: + return org_id in _org_ids_option(CAP_DSC_TRANSACTION_LENGTH_ORG_IDS_OPTION) + + +def _org_ids_option(option_name: str) -> set[int]: return { int(org_id) - for org_id in options.get(SLIDING_WINDOW_COMPARISON_ORG_IDS_OPTION) + for org_id in options.get(option_name) if isinstance(org_id, int) or (isinstance(org_id, str) and org_id.isdigit()) } diff --git a/src/sentry/dynamic_sampling/per_org/queries.py b/src/sentry/dynamic_sampling/per_org/queries.py index 49ba7ab3bd37..9e4819b5d49a 100644 --- a/src/sentry/dynamic_sampling/per_org/queries.py +++ b/src/sentry/dynamic_sampling/per_org/queries.py @@ -7,9 +7,11 @@ from enum import StrEnum from typing import Any, Protocol -from sentry_protos.snuba.v1.trace_item_attribute_pb2 import ExtrapolationMode +from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeValue, ExtrapolationMode +from sentry_protos.snuba.v1.trace_item_filter_pb2 import ComparisonFilter, TraceItemFilter from sentry import options +from sentry.dynamic_sampling.per_org.gate import is_org_in_cap_dsc_transaction_length_rollout from sentry.dynamic_sampling.rules.utils import ProjectId from sentry.dynamic_sampling.tasks.common import ( ACTIVE_ORGS_VOLUMES_DEFAULT_TIME_INTERVAL, @@ -20,6 +22,7 @@ from sentry.models.organization import Organization from sentry.models.project import Project from sentry.search.eap.constants import SAMPLING_MODE_HIGHEST_ACCURACY +from sentry.search.eap.resolver import SearchResolver from sentry.search.eap.types import SearchResolverConfig from sentry.search.events.types import SnubaParams from sentry.sentry_metrics import indexer @@ -49,6 +52,13 @@ class DynamicSamplingQueryFields(StrEnum): MAX_RECEIVED = "max(received)" +# Transaction names longer than this are not addressable by a sampling rule, so a per-name +# rate for them cannot be served. They all count as one class under the empty transaction +# name, which is the name the rule for that class carries. +MAX_DSC_TRANSACTION_LENGTH = 200 +LUMPED_DSC_TRANSACTION = "" + + @dataclass(order=True) class ProjectVolume: project_id: ProjectId @@ -330,10 +340,15 @@ def get_eap_transaction_volumes( root_projects: Sequence[Project] | None = None, ) -> list[ProjectTransactionCounts]: """ - Fetch the highest-volume transactions of every root project in a single - LIMIT BY query, mirroring the legacy pipeline's per-project top-N - (``LIMIT BY (org_id, project_id)`` in boost_low_volume_transactions) so the - transaction rebalancing model sees the same explicit transaction set. + Fetch the highest-volume transactions of every root project, mirroring the legacy + pipeline's per-project top-N (``LIMIT BY (org_id, project_id)`` in + boost_low_volume_transactions) so the transaction rebalancing model sees the same + explicit transaction set. + + For organizations in the cap-dsc-transaction-length rollout, transactions longer than + ``MAX_DSC_TRANSACTION_LENGTH`` cannot be named by a sampling rule, so they are counted + as one class under ``LUMPED_DSC_TRANSACTION`` rather than individually. That costs a + second query: the top-N one excludes them, and the second one sums them per project. """ # Spans rooted in one project can be owned by any project in the org, so the query # scope stays config.projects; root_projects only narrows which root projects @@ -355,56 +370,161 @@ def get_eap_transaction_volumes( end_time = datetime.now(UTC) start_time = end_time - time_interval - transaction_counts_by_project: defaultdict[int, list[tuple[str, float]]] = defaultdict(list) + params = SnubaParams( + start=start_time, + end=end_time, + projects=config.projects, + organization=config.organization, + ) + resolver_config = SearchResolverConfig( + auto_fields=True, + extrapolation_mode=ExtrapolationMode.EXTRAPOLATION_MODE_SERVER_ONLY, + ) + # Shared so both queries resolve their columns once and see the same attribute definitions. + resolver = Spans.get_resolver(params, resolver_config) + + root_project_filter = ",".join(str(project.id) for project in root_projects) + shared_query = { + "params": params, + "query_string": f"{DynamicSamplingQueryFilters.IS_SEGMENT} {DynamicSamplingQueryFields.DSC_PROJECT_ID}:[{root_project_filter}] has:{DynamicSamplingQueryFields.DSC_TRANSACTION}", + "referrer": Referrer.DYNAMIC_SAMPLING_PER_ORG_GET_EAP_TRANSACTION_VOLUMES.value, + "config": resolver_config, + "search_resolver": resolver, + "sampling_mode": SAMPLING_MODE_HIGHEST_ACCURACY, + } + + cap_transaction_length = is_org_in_cap_dsc_transaction_length_rollout(config.organization.id) + named_counts = _get_named_transaction_counts( + shared_query, + _dsc_transaction_length_filter(resolver, over_length=False) + if cap_transaction_length + else None, + max_transactions_per_project, + ) + over_length_counts = ( + _get_over_length_transaction_counts( + shared_query, _dsc_transaction_length_filter(resolver, over_length=True) + ) + if cap_transaction_length + else {} + ) - orderby = [ - DynamicSamplingQueryFields.DSC_PROJECT_ID, - f"-{DynamicSamplingQueryFields.COUNT}", - DynamicSamplingQueryFields.DSC_TRANSACTION, + return [ + ProjectTransactionCounts( + project_id=project_id, + org_id=config.organization.id, + transaction_counts=_merge_over_length_counts( + named_counts.get(project_id, []), + over_length_counts.get(project_id, 0.0), + max_transactions_per_project, + ), + ) + for project_id in sorted(named_counts.keys() | over_length_counts.keys()) ] - root_project_filter = ",".join(str(project.id) for project in root_projects) + +def _dsc_transaction_length_filter(resolver: SearchResolver, over_length: bool) -> TraceItemFilter: + """ + A filter on the length of ``dsc.transaction``, keeping either only the over-length names + or only the rest. + + The RPC has no length function, so the length test is a ClickHouse LIKE pattern of one + ``_`` (any single character) per allowed character, followed by ``%``. A name matches + that pattern only if it has more characters than the limit. + """ + attribute, _ = resolver.resolve_attribute(DynamicSamplingQueryFields.DSC_TRANSACTION) + return TraceItemFilter( + comparison_filter=ComparisonFilter( + key=attribute.proto_definition, + op=ComparisonFilter.OP_LIKE if over_length else ComparisonFilter.OP_NOT_LIKE, + value=AttributeValue(val_str="_" * (MAX_DSC_TRANSACTION_LENGTH + 1) + "%"), + ) + ) + + +def _get_named_transaction_counts( + shared_query: dict[str, Any], + length_filter: TraceItemFilter | None, + max_transactions_per_project: int, +) -> dict[int, list[tuple[str, float]]]: + """ + Per-project top-N transaction volumes. Without a ``length_filter`` every transaction is + counted by name; with one only the names a sampling rule can address are. + """ + counts_by_project: defaultdict[int, list[tuple[str, float]]] = defaultdict(list) for row in run_eap_spans_table_query_in_chunks( { - "params": SnubaParams( - start=start_time, - end=end_time, - projects=config.projects, - organization=config.organization, - ), - "query_string": f"{DynamicSamplingQueryFilters.IS_SEGMENT} {DynamicSamplingQueryFields.DSC_PROJECT_ID}:[{root_project_filter}] has:{DynamicSamplingQueryFields.DSC_TRANSACTION}", + **shared_query, "selected_columns": [ DynamicSamplingQueryFields.DSC_PROJECT_ID, DynamicSamplingQueryFields.DSC_TRANSACTION, DynamicSamplingQueryFields.COUNT, ], - "orderby": orderby, + "orderby": [ + DynamicSamplingQueryFields.DSC_PROJECT_ID, + f"-{DynamicSamplingQueryFields.COUNT}", + DynamicSamplingQueryFields.DSC_TRANSACTION, + ], "limit_by": LimitBy( columns=[DynamicSamplingQueryFields.DSC_PROJECT_ID], limit=max_transactions_per_project, ), - "referrer": Referrer.DYNAMIC_SAMPLING_PER_ORG_GET_EAP_TRANSACTION_VOLUMES.value, - "config": SearchResolverConfig( - auto_fields=True, - extrapolation_mode=ExtrapolationMode.EXTRAPOLATION_MODE_SERVER_ONLY, - ), - "sampling_mode": SAMPLING_MODE_HIGHEST_ACCURACY, + "extra_conditions": length_filter, } ): - transaction = row.get(DynamicSamplingQueryFields.DSC_TRANSACTION) total = _get_aggregate_float(row, DynamicSamplingQueryFields.COUNT) if total <= 0: continue project_id = _get_aggregate_int(row, DynamicSamplingQueryFields.DSC_PROJECT_ID) - transaction_counts = transaction_counts_by_project[project_id] - transaction_counts.append((str(transaction), total)) + transaction = str(row.get(DynamicSamplingQueryFields.DSC_TRANSACTION)) + counts_by_project[project_id].append((transaction, total)) - return [ - ProjectTransactionCounts( - project_id=project_id, - org_id=config.organization.id, - transaction_counts=transaction_counts, + return counts_by_project + + +def _get_over_length_transaction_counts( + shared_query: dict[str, Any], length_filter: TraceItemFilter +) -> dict[int, float]: + """ + Per-project volume of all over-length transactions together. They group by project only, + so every one of them is counted, however many distinct names a project has. + """ + counts_by_project: dict[int, float] = {} + for row in run_eap_spans_table_query_in_chunks( + { + **shared_query, + "selected_columns": [ + DynamicSamplingQueryFields.DSC_PROJECT_ID, + DynamicSamplingQueryFields.COUNT, + ], + "orderby": [DynamicSamplingQueryFields.DSC_PROJECT_ID], + "extra_conditions": length_filter, + } + ): + total = _get_aggregate_float(row, DynamicSamplingQueryFields.COUNT) + if total <= 0: + continue + + counts_by_project[_get_aggregate_int(row, DynamicSamplingQueryFields.DSC_PROJECT_ID)] = ( + total ) - for project_id, transaction_counts in sorted(transaction_counts_by_project.items()) - ] + + return counts_by_project + + +def _merge_over_length_counts( + named_counts: list[tuple[str, float]], + over_length_count: float, + max_transactions_per_project: int, +) -> list[tuple[str, float]]: + """ + Add the over-length volume as one more class and keep the per-project top-N. The lumped + class competes for a slot like any named one, so a project still gets at most N classes. + """ + if over_length_count <= 0: + return named_counts + + merged = named_counts + [(LUMPED_DSC_TRANSACTION, over_length_count)] + merged.sort(key=lambda entry: (-entry[1], entry[0])) + return merged[:max_transactions_per_project] diff --git a/src/sentry/options/defaults.py b/src/sentry/options/defaults.py index 465382b28fef..5408c16c1bf1 100644 --- a/src/sentry/options/defaults.py +++ b/src/sentry/options/defaults.py @@ -2319,6 +2319,17 @@ flags=FLAG_AUTOMATOR_MODIFIABLE, ) +# Organizations for which the per-org pipeline caps dsc.transaction at 200 characters, +# counting every longer transaction as one class under the empty transaction name. Longer +# names cannot be matched by a sampling rule, so a per-name rate for them cannot be served. +# This costs one extra EAP query per org. Empty keeps counting them individually by name. +register( + "dynamic-sampling.per_org.cap-dsc-transaction-length-org-ids", + type=Sequence, + default=[], + flags=FLAG_AUTOMATOR_MODIFIABLE, +) + # Per-project sample rate overrides for custom dynamic sampling. Maps a stringified # project id to a fixed sample rate (0.0-1.0) that hard-replaces the rate the custom # dynamic sampling path would otherwise compute for that project. Example: diff --git a/src/sentry/snuba/spans_rpc.py b/src/sentry/snuba/spans_rpc.py index 476bd99d5949..4b7d884313ef 100644 --- a/src/sentry/snuba/spans_rpc.py +++ b/src/sentry/snuba/spans_rpc.py @@ -13,6 +13,7 @@ ) from sentry_protos.snuba.v1.request_common_pb2 import PageToken, TraceItemType from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeKey +from sentry_protos.snuba.v1.trace_item_filter_pb2 import TraceItemFilter from sentry import options from sentry.models.project import Project @@ -62,6 +63,7 @@ def run_table_query( additional_queries: AdditionalQueries | None = None, max_string_length: int | None = None, limit_by: rpc_dataset_common.LimitBy | None = None, + extra_conditions: TraceItemFilter | None = None, ) -> EAPResponse: return cls._run_table_query( rpc_dataset_common.TableQuery( @@ -78,6 +80,7 @@ def run_table_query( additional_queries=additional_queries, max_string_length=max_string_length, limit_by=limit_by, + extra_conditions=extra_conditions, ), params.debug, ) diff --git a/tests/sentry/dynamic_sampling/per_org/test_queries.py b/tests/sentry/dynamic_sampling/per_org/test_queries.py index 7265cd82464f..ee3321108557 100644 --- a/tests/sentry/dynamic_sampling/per_org/test_queries.py +++ b/tests/sentry/dynamic_sampling/per_org/test_queries.py @@ -10,7 +10,10 @@ BaseDynamicSamplingConfiguration, get_configuration, ) +from sentry.dynamic_sampling.per_org.gate import CAP_DSC_TRANSACTION_LENGTH_ORG_IDS_OPTION from sentry.dynamic_sampling.per_org.queries import ( + LUMPED_DSC_TRANSACTION, + MAX_DSC_TRANSACTION_LENGTH, DynamicSamplingQueryFields, DynamicSamplingQueryFilters, ProjectTransactionCounts, @@ -23,6 +26,7 @@ ) from sentry.dynamic_sampling.tasks.common import OrganizationDataVolume from sentry.models.organization import Organization +from sentry.models.project import Project from sentry.search.eap.constants import SAMPLING_MODE_HIGHEST_ACCURACY from sentry.search.eap.types import SearchResolverConfig from sentry.search.events.types import SnubaParams @@ -715,3 +719,141 @@ def segment(transaction, project, offset): ("quiet-high", 3), ("quiet-low", 2), ] + + def store_over_length_transaction_spans( + self, organization: Organization, project: Project, at_limit: str + ) -> None: + timestamp = before_now(minutes=15) + + def segment(transaction, offset): + return self.create_span( + { + "is_segment": True, + "sentry_tags": { + "transaction": transaction, + "dsc.transaction": transaction, + "dsc.project_id": str(project.id), + }, + }, + organization=organization, + project=project, + start_ts=timestamp + timedelta(seconds=offset), + ) + + self.store_spans( + [ + segment(at_limit, 0), + segment(at_limit, 1), + segment(at_limit, 2), + segment(at_limit, 3), + segment("b" * (MAX_DSC_TRANSACTION_LENGTH + 1), 4), + segment("b" * (MAX_DSC_TRANSACTION_LENGTH + 1), 5), + segment("c" * (MAX_DSC_TRANSACTION_LENGTH + 10), 6), + ] + ) + + def test_get_eap_transaction_volumes_lumps_over_length_transactions(self) -> None: + organization = self.create_organization() + project = self.create_project(organization=organization) + at_limit = "a" * MAX_DSC_TRANSACTION_LENGTH + self.store_over_length_transaction_spans(organization, project, at_limit) + + with self.options({CAP_DSC_TRANSACTION_LENGTH_ORG_IDS_OPTION: [organization.id]}): + volumes = get_eap_transaction_volumes(self.get_config(organization)) + + # The two over-length names are one class, so their counts add up to 3. + assert volumes == [ + ProjectTransactionCounts( + org_id=organization.id, + project_id=project.id, + transaction_counts=[(at_limit, 4), (LUMPED_DSC_TRANSACTION, 3)], + ) + ] + + def test_get_eap_transaction_volumes_counts_over_length_transactions_by_name_when_not_capped( + self, + ) -> None: + organization = self.create_organization() + project = self.create_project(organization=organization) + at_limit = "a" * MAX_DSC_TRANSACTION_LENGTH + self.store_over_length_transaction_spans(organization, project, at_limit) + + volumes = get_eap_transaction_volumes(self.get_config(organization)) + + assert volumes == [ + ProjectTransactionCounts( + org_id=organization.id, + project_id=project.id, + transaction_counts=[ + (at_limit, 4), + ("b" * (MAX_DSC_TRANSACTION_LENGTH + 1), 2), + ("c" * (MAX_DSC_TRANSACTION_LENGTH + 10), 1), + ], + ) + ] + + def test_get_eap_transaction_volumes_counts_over_length_transactions_beyond_the_cap( + self, + ) -> None: + """ + The over-length names are one class no matter how many of them a project has, so + the per-project cap cannot hide any of their volume. + """ + organization = self.create_organization() + project = self.create_project(organization=organization) + timestamp = before_now(minutes=15) + + def segment(transaction, offset): + return self.create_span( + { + "is_segment": True, + "sentry_tags": { + "transaction": transaction, + "dsc.transaction": transaction, + "dsc.project_id": str(project.id), + }, + }, + organization=organization, + project=project, + start_ts=timestamp + timedelta(seconds=offset), + ) + + spans = [segment("named", 0), segment("named", 1), segment("named", 2)] + # Five distinct over-length names, well past the cap of two. + for index in range(5): + spans.append( + segment(chr(ord("a") + index) * (MAX_DSC_TRANSACTION_LENGTH + 1), 10 + index) + ) + self.store_spans(spans) + + with self.options({CAP_DSC_TRANSACTION_LENGTH_ORG_IDS_OPTION: [organization.id]}): + volumes = get_eap_transaction_volumes( + self.get_config(organization), + max_transactions_per_project=2, + ) + + assert volumes == [ + ProjectTransactionCounts( + org_id=organization.id, + project_id=project.id, + transaction_counts=[(LUMPED_DSC_TRANSACTION, 5), ("named", 3)], + ) + ] + + def test_get_eap_transaction_volumes_does_not_cap_organizations_outside_the_rollout( + self, + ) -> None: + organization = self.create_organization() + other_organization = self.create_organization() + project = self.create_project(organization=organization) + at_limit = "a" * MAX_DSC_TRANSACTION_LENGTH + self.store_over_length_transaction_spans(organization, project, at_limit) + + with self.options({CAP_DSC_TRANSACTION_LENGTH_ORG_IDS_OPTION: [other_organization.id]}): + volumes = get_eap_transaction_volumes(self.get_config(organization)) + + assert volumes[0].transaction_counts == [ + (at_limit, 4), + ("b" * (MAX_DSC_TRANSACTION_LENGTH + 1), 2), + ("c" * (MAX_DSC_TRANSACTION_LENGTH + 10), 1), + ]