diff --git a/src/sentry/features/temporary.py b/src/sentry/features/temporary.py index 11057cb61f09..6b382fdaac1b 100644 --- a/src/sentry/features/temporary.py +++ b/src/sentry/features/temporary.py @@ -465,4 +465,5 @@ def register_temporary_features(manager: FeatureManager) -> None: manager.add("projects:workflow-engine-performance-detectors", ProjectFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False) + manager.add("organizations:relay-generate-billing-outcome", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False) # fmt: on diff --git a/src/sentry/ingest/billing_metrics_consumer.py b/src/sentry/ingest/billing_metrics_consumer.py index b822a2d93964..c241b90a712a 100644 --- a/src/sentry/ingest/billing_metrics_consumer.py +++ b/src/sentry/ingest/billing_metrics_consumer.py @@ -42,6 +42,7 @@ class BillingTxCountMetricConsumerStrategy(ProcessingStrategy[KafkaPayload]): span_metric_id = SPAN_METRICS_NAMES["c:spans/usage@none"] span_is_segment_tag = str(SHARED_TAG_STRINGS["is_segment"]) + billing_outcome_accepted_tag = str(SHARED_TAG_STRINGS["billing_outcome_emitted"]) def __init__(self, next_step: ProcessingStrategy[Any]) -> None: self.__next_step = next_step @@ -79,6 +80,10 @@ def _count_processed_items(self, generic_metric: GenericMetric) -> Mapping[DataC if generic_metric["metric_id"] != self.span_metric_id: return {} + # If relay already produced an outcome for this item, ignore it. + if generic_metric["tags"].get(self.billing_outcome_accepted_tag) == "true": + return {} + value = generic_metric["value"] try: quantity = max(int(value), 0) # type: ignore[arg-type] diff --git a/src/sentry/relay/config/__init__.py b/src/sentry/relay/config/__init__.py index e375e68510fd..7c3c37f7edd7 100644 --- a/src/sentry/relay/config/__init__.py +++ b/src/sentry/relay/config/__init__.py @@ -56,6 +56,7 @@ "organizations:session-replay-recording-scrubbing", "organizations:session-replay-video-disabled", "organizations:session-replay", + "organizations:relay-generate-billing-outcome", "projects:discard-transaction", "projects:span-metrics-extraction", "projects:span-metrics-extraction-addons", diff --git a/src/sentry/sentry_metrics/indexer/strings.py b/src/sentry/sentry_metrics/indexer/strings.py index 187ce0990655..d25cce6584b1 100644 --- a/src/sentry/sentry_metrics/indexer/strings.py +++ b/src/sentry/sentry_metrics/indexer/strings.py @@ -202,6 +202,7 @@ "has_transaction": PREFIX + 281, "was_transaction": PREFIX + 282, "is_segment": PREFIX + 283, + "billing_outcome_emitted": PREFIX + 284, # GENERAL/MISC (don't have a category) "": PREFIX + 1000, } diff --git a/tests/sentry/ingest/billing_metrics_consumer/test_billing_metrics_consumer_kafka.py b/tests/sentry/ingest/billing_metrics_consumer/test_billing_metrics_consumer_kafka.py index e7ade8425e00..e8a3bd64ba84 100644 --- a/tests/sentry/ingest/billing_metrics_consumer/test_billing_metrics_consumer_kafka.py +++ b/tests/sentry/ingest/billing_metrics_consumer/test_billing_metrics_consumer_kafka.py @@ -141,3 +141,102 @@ def generate_kafka_message(generic_metric: GenericMetric) -> Message[KafkaPayloa strategy.join() assert next_step.join.call_count == 1 + + +@mock.patch("sentry.ingest.billing_metrics_consumer.track_outcome") +def test_no_double_billing_outcomes(track_outcome) -> None: + topic = Topic("snuba-generic-metrics") + + organization = 123 + project_1 = 56789 + + span_usage_mri = "c:spans/usage@none" + span_usage_id = SPAN_METRICS_NAMES[span_usage_mri] + + generic_metrics: list[GenericMetric] = [ + { + "mapping_meta": {"c": {str(span_usage_id): span_usage_mri}}, + "metric_id": span_usage_id, + "type": "d", + "org_id": organization, + "project_id": project_1, + "timestamp": 123456, + "value": 65.0, + "tags": {str(SHARED_TAG_STRINGS["billing_outcome_emitted"]): "true"}, + "use_case_id": "spans", + "retention_days": 90, + }, + { + "mapping_meta": {"c": {str(span_usage_id): span_usage_mri}}, + "metric_id": span_usage_id, + "type": "d", + "org_id": organization, + "project_id": project_1, + "timestamp": 123456, + "value": 12.0, + "tags": {str(SHARED_TAG_STRINGS["is_segment"]): "true"}, + "use_case_id": "spans", + "retention_days": 90, + }, + ] + + next_step = mock.MagicMock() + + strategy = BillingTxCountMetricConsumerStrategy( + next_step=next_step, + ) + + generate_kafka_message_counter = 0 + + def generate_kafka_message(generic_metric: GenericMetric) -> Message[KafkaPayload]: + nonlocal generate_kafka_message_counter + + encoded = orjson.dumps(generic_metric) + payload = KafkaPayload(key=None, value=encoded, headers=[]) + message = Message( + BrokerValue( + payload, + Partition(topic, index=0), + generate_kafka_message_counter, + datetime.now(timezone.utc), + ) + ) + generate_kafka_message_counter += 1 + return message + + # Mimick the behavior of StreamProcessor._run_once: Call poll repeatedly, + # then call submit when there is a message. + strategy.poll() + strategy.poll() + assert track_outcome.call_count == 0 + for generic_metric in generic_metrics: + strategy.poll() + strategy.submit(generate_kafka_message(generic_metric)) + + assert track_outcome.mock_calls == [ + mock.call( + org_id=organization, + project_id=project_1, + key_id=None, + outcome=Outcome.ACCEPTED, + reason=None, + timestamp=mock.ANY, + event_id=None, + category=DataCategory.SPAN, + quantity=12, + ), + mock.call( + org_id=organization, + project_id=project_1, + key_id=None, + outcome=Outcome.ACCEPTED, + reason=None, + timestamp=mock.ANY, + event_id=None, + category=DataCategory.TRANSACTION, + quantity=12, + ), + ] + + strategy.join() + assert next_step.join.call_count == 1 diff --git a/tests/sentry/sentry_metrics/test_prevent_modifications_to_indexer_strings.py b/tests/sentry/sentry_metrics/test_prevent_modifications_to_indexer_strings.py index d94c33ef2420..c9a7d9e74218 100644 --- a/tests/sentry/sentry_metrics/test_prevent_modifications_to_indexer_strings.py +++ b/tests/sentry/sentry_metrics/test_prevent_modifications_to_indexer_strings.py @@ -1,7 +1,7 @@ from hashlib import sha256 LOCKED_FILE = "src/sentry/sentry_metrics/indexer/strings.py" -LOCKED_DIGEST = "f984bf497f587f7d52665714cb9a049cf9f515f0bbb288f8782cc018484e65ea" +LOCKED_DIGEST = "9a2c7aeb2abf0c8ce704504201073df1d4776ff2d74d7a7af349bfc0198dedb7" MESSAGE = f"""{LOCKED_FILE} is locked. * We have detected you made changes to this file.