diff --git a/src/sentry/features/temporary.py b/src/sentry/features/temporary.py index 0c4e777a8b78..6386513e2a52 100644 --- a/src/sentry/features/temporary.py +++ b/src/sentry/features/temporary.py @@ -247,6 +247,8 @@ def register_temporary_features(manager: FeatureManager) -> None: manager.add("organizations:profiling-global-suspect-functions", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True) # Enable function trends widgets in profiling manager.add("organizations:profiling-function-trends", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True) + # Produce function regression occurrences to Kafka in Sentry instead of vroom. + manager.add("organizations:profiling-sentry-regression-occurrences", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False) # Prefer finalized release dates for issue regression and next-release resolution. manager.add("organizations:release-resolution-finalized-order", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False) # Enable replay AI summaries for mobile replays diff --git a/src/sentry/tasks/statistical_detectors.py b/src/sentry/tasks/statistical_detectors.py index f53ae4fd6d38..8e4b619a51c1 100644 --- a/src/sentry/tasks/statistical_detectors.py +++ b/src/sentry/tasks/statistical_detectors.py @@ -4,7 +4,7 @@ import logging from collections.abc import Generator, Iterable, Sequence from datetime import UTC, datetime, timedelta -from typing import Any +from typing import Any, cast from django.utils import timezone as django_timezone from snuba_sdk import ( @@ -23,6 +23,8 @@ from sentry import features, options, projectoptions from sentry.constants import ObjectStatus from sentry.issues.endpoints.project_performance_issue_settings import InternalProjectOptions +from sentry.issues.issue_occurrence import IssueOccurrence, IssueOccurrenceData +from sentry.issues.producer import produce_occurrence_to_kafka from sentry.models.options.project_option import ProjectOption from sentry.models.project import Project from sentry.models.statistical_detectors import RegressionType @@ -344,11 +346,16 @@ def detect_regressions_with_viewer_context() -> Generator[BreakpointData]: ) breakpoint_count = 0 - emitted_count = 0 + vroom_emitted_count = 0 + sentry_emitted_count = 0 for regression_chunk in chunked(regressions, 100): breakpoint_count += len(regression_chunk) - emitted_count += emit_function_regression_issue(projects_by_id, regression_chunk, start) + vroom_count, sentry_count = emit_function_regression_issue( + projects_by_id, regression_chunk, start + ) + vroom_emitted_count += vroom_count + sentry_emitted_count += sentry_count metrics.incr( "statistical_detectors.breakpoint.detected", @@ -359,8 +366,14 @@ def detect_regressions_with_viewer_context() -> Generator[BreakpointData]: metrics.incr( "statistical_detectors.breakpoint.emitted", - amount=emitted_count, - tags={"source": "profile", "kind": "function"}, + amount=vroom_emitted_count, + tags={"source": "profile", "kind": "function", "producer": "vroom"}, + sample_rate=1.0, + ) + metrics.incr( + "statistical_detectors.breakpoint.emitted", + amount=sentry_emitted_count, + tags={"source": "profile", "kind": "function", "producer": "sentry"}, sample_rate=1.0, ) @@ -369,7 +382,7 @@ def emit_function_regression_issue( projects_by_id: dict[int, Project], regressions: list[BreakpointData], start: datetime, -) -> int: +) -> tuple[int, int]: start = start - timedelta(hours=1) start = start.replace(minute=0, second=0, microsecond=0) @@ -424,7 +437,8 @@ def emit_function_regression_issue( continuous_examples = fetch_continuous_examples(raw_continuous_examples) - payloads = [] + payloads_by_producer: dict[bool, list[dict[str, Any]]] = {} + produce_in_sentry_by_org: dict[int, bool] = {} for regression in regressions: project_id = int(regression["project"]) @@ -440,7 +454,12 @@ def emit_function_regression_issue( if example is None: continue - payloads.append( + if project.organization_id not in produce_in_sentry_by_org: + produce_in_sentry_by_org[project.organization_id] = features.has( + "organizations:profiling-sentry-regression-occurrences", project.organization + ) + produce_in_sentry = produce_in_sentry_by_org[project.organization_id] + payloads_by_producer.setdefault(produce_in_sentry, []).append( { "organization_id": project.organization_id, "project_id": project_id, @@ -457,15 +476,37 @@ def emit_function_regression_issue( } ) - if not payloads: - return 0 + vroom_occurrences = 0 + sentry_occurrences = 0 + for produce_in_sentry, payloads in payloads_by_producer.items(): + response = get_from_profiling_service( + method="POST", + path="/regressed?produce=false" if produce_in_sentry else "/regressed", + json_data=payloads, + ) + if response.status != 200: + continue - response = get_from_profiling_service(method="POST", path="/regressed", json_data=payloads) - if response.status != 200: - return 0 + data = json.loads(response.data) + if produce_in_sentry: + for payload in data["occurrence_payloads"]: + event_data = payload["event"] + occurrence = IssueOccurrence.from_dict( + cast( + IssueOccurrenceData, + { + **payload, + "event_id": event_data["event_id"], + "resource_id": payload.get("resource_id"), + }, + ) + ) + produce_occurrence_to_kafka(occurrence=occurrence, event_data=event_data) + sentry_occurrences += len(data["occurrence_payloads"]) + else: + vroom_occurrences += data["occurrences"] - data = json.loads(response.data) - return data.get("occurrences") + return vroom_occurrences, sentry_occurrences def fetch_continuous_examples(raw_examples): diff --git a/tests/sentry/tasks/test_statistical_detectors.py b/tests/sentry/tasks/test_statistical_detectors.py index 6e679a2cfc92..f65043b2aece 100644 --- a/tests/sentry/tasks/test_statistical_detectors.py +++ b/tests/sentry/tasks/test_statistical_detectors.py @@ -1,19 +1,27 @@ import itertools import uuid +from copy import deepcopy from datetime import UTC, datetime, timedelta from unittest import mock import orjson import pytest +from arroyo.backends.kafka import KafkaPayload +from arroyo.backends.local.backend import LocalBroker +from arroyo.backends.local.storages.memory import MemoryMessageStorage +from arroyo.types import Partition +from arroyo.types import Topic as ArroyoTopic from django.db.models import F from django.test import override_settings from urllib3.response import HTTPResponse +from sentry.conf.types.kafka_definition import Topic from sentry.issues.endpoints.project_performance_issue_settings import InternalProjectOptions from sentry.issues.grouptype import ( ProfileFunctionRegressionType, ) -from sentry.issues.occurrence_consumer import _process_message +from sentry.issues.issue_occurrence import IssueOccurrence +from sentry.issues.occurrence_consumer import _get_kwargs, _process_message from sentry.issues.producer import PayloadType from sentry.issues.status_change_message import StatusChangeMessage from sentry.models.group import GroupStatus @@ -46,13 +54,14 @@ query_functions, run_detection, ) -from sentry.testutils.cases import ProfilesSnubaTestCase +from sentry.testutils.cases import ProfilesSnubaTestCase, TestCase from sentry.testutils.factories import Factories from sentry.testutils.helpers import override_options from sentry.testutils.helpers.datetime import before_now, freeze_time from sentry.testutils.pytest.fixtures import django_db_all from sentry.types.group import GroupSubStatus from sentry.utils.concurrent import ContextPropagatingThreadPoolExecutor +from sentry.utils.kafka_config import get_topic_definition from sentry.viewer_context import ( ActorType, ViewerContext, @@ -981,7 +990,10 @@ def test_process_function_change_points_groups_by_start(detect, timestamp): ] -@mock.patch("sentry.tasks.statistical_detectors.emit_function_regression_issue") +@mock.patch("sentry.tasks.statistical_detectors.metrics.incr") +@mock.patch( + "sentry.tasks.statistical_detectors.emit_function_regression_issue", return_value=(2, 3) +) @mock.patch("sentry.seer.breakpoints.seer_breakpoint_connection_pool.urlopen") @mock.patch("sentry.search.events.builder.base.raw_snql_query") @override_settings(SEER_API_SHARED_SECRET="viewer-context-test-secret") @@ -990,6 +1002,7 @@ def test_detect_function_change_points( mock_raw_snql_query, mock_urlopen, mock_emit_function_regression_issue, + mock_metrics_incr, timestamp, project, ): @@ -1047,6 +1060,22 @@ def test_detect_function_change_points( detect_function_change_points([(project.id, fingerprint)], timestamp.isoformat()) assert mock_emit_function_regression_issue.called + mock_metrics_incr.assert_has_calls( + [ + mock.call( + "statistical_detectors.breakpoint.emitted", + amount=2, + tags={"source": "profile", "kind": "function", "producer": "vroom"}, + sample_rate=1.0, + ), + mock.call( + "statistical_detectors.breakpoint.emitted", + amount=3, + tags={"source": "profile", "kind": "function", "producer": "sentry"}, + sample_rate=1.0, + ), + ] + ) viewer_context = decode_viewer_context( mock_urlopen.call_args.kwargs["headers"]["X-Viewer-Context"], key="viewer-context-test-secret", @@ -1597,6 +1626,209 @@ def get_trends(ts): ) +@override_settings(SENTRY_EVENTSTREAM="sentry.eventstream.kafka.KafkaEventStream") +class EmitFunctionRegressionOccurrencesTest(TestCase): + def setUp(self) -> None: + super().setUp() + self.now = datetime(2023, 8, 1, 12, tzinfo=UTC) + self.broker = LocalBroker(MemoryMessageStorage[KafkaPayload]()) + self.topic = ArroyoTopic(get_topic_definition(Topic.INGEST_OCCURRENCES)["real_topic_name"]) + self.broker.create_topic(self.topic, partitions=1) + producer_patch = mock.patch( + "sentry.issues.producer._occurrence_producer", self.broker.get_producer() + ) + producer_patch.start() + self.addCleanup(producer_patch.stop) + query_patch = mock.patch("sentry.tasks.statistical_detectors.functions.query") + self.query = query_patch.start() + self.addCleanup(query_patch.stop) + self.query.return_value = { + "data": [ + { + "project.id": self.project.id, + "fingerprint": 1176630088, + "all_examples()": [{"profile_id": "a" * 32}], + } + ] + } + vroom_patch = mock.patch("sentry.tasks.statistical_detectors.get_from_profiling_service") + self.vroom = vroom_patch.start() + self.addCleanup(vroom_patch.stop) + self.payload = { + "culprit": "work", + "detection_time": "2023-08-01T12:00:00.123456789Z", + "event": { + "debug_meta": {}, + "event_id": "a" * 32, + "platform": "node", + "project_id": self.project.id, + "received": "2023-08-01T12:00:00.123456789Z", + "stacktrace": {"frames": None}, + "tags": {}, + "timestamp": "2023-08-01T12:00:00.123456789Z", + }, + "evidence_data": { + "organization_id": self.organization.id, + "project_id": self.project.id, + "file": "", + "fingerprint": 1176630088, + "function": "work", + "module": "", + "package": "", + "path": "", + "symbol": "", + "absolute_percentage_change": 1, + "aggregate_range_1": 100_000_000, + "aggregate_range_2": 200_000_000, + "breakpoint": 1690804800, + "trend_difference": 100_000_000, + "trend_percentage": 2, + "unweighted_p_value": 0.01, + "unweighted_t_value": 1.23, + }, + "evidence_display": [ + { + "name": "Regression", + "value": "work duration increased from 100ms to 200ms (P95).", + "important": True, + }, + {"name": "Breakpoint", "value": "1690804800", "important": False}, + {"name": "Fully qualified name", "value": "work", "important": False}, + ], + "fingerprint": ["4621f348"], + "id": "b" * 32, + "issue_title": "Function Regression", + "level": "info", + "payload_type": "occurrence", + "project_id": self.project.id, + "subtitle": "Duration increased from 100ms to 200ms (P95).", + "type": 2011, + } + self.vroom.return_value = HTTPResponse( + status=200, + body=orjson.dumps( + {"occurrences": 1, "emitted": [], "occurrence_payloads": [self.payload]} + ), + ) + + def regression(self, project: Project) -> BreakpointData: + return { + "project": str(project.id), + "transaction": "1176630088", + "aggregate_range_1": 100_000_000, + "aggregate_range_2": 200_000_000, + "unweighted_t_value": 1.23, + "unweighted_p_value": 0.01, + "trend_percentage": 2, + "absolute_percentage_change": 1, + "trend_difference": 100_000_000, + "breakpoint": 1690804800, + } + + def emit(self) -> tuple[int, int]: + return emit_function_regression_issue( + {self.project.id: self.project}, [self.regression(self.project)], self.now + ) + + def test_enabled_payload_compatibility(self) -> None: + with self.feature("organizations:profiling-sentry-regression-occurrences"): + assert self.emit() == (0, 1) + assert self.vroom.call_args.kwargs["path"] == "/regressed?produce=false" + message = self.broker.consume(Partition(self.topic, 0), 0) + assert message is not None + payload = orjson.loads(message.payload.value) + assert message.payload.key == f"4621f348-{self.project.id}".encode() + assert payload["event_id"] == self.payload["event"]["event_id"] + assert payload["resource_id"] is None + assert payload["type"] == ProfileFunctionRegressionType.type_id + assert ( + payload["detection_time"] + == datetime(2023, 8, 1, 12, 0, 0, 123456, tzinfo=UTC).timestamp() + ) + assert payload["evidence_data"] == self.payload["evidence_data"] + assert payload["evidence_display"] == self.payload["evidence_display"] + assert payload["fingerprint"] == self.payload["fingerprint"] + assert payload["payload_type"] == "occurrence" + assert payload["event"].items() >= self.payload["event"].items() + original = _get_kwargs(deepcopy(self.payload)) + produced = _get_kwargs(payload) + assert ( + IssueOccurrence.from_dict(original["occurrence_data"]).to_dict() + == IssueOccurrence.from_dict(produced["occurrence_data"]).to_dict() + ) + original["event_data"]["contexts"]["trace"].pop("trace_id") + produced["event_data"]["contexts"]["trace"].pop("trace_id") + assert original["event_data"] == produced["event_data"] + assert self.broker.consume(Partition(self.topic, 0), 1) is None + + def test_disabled(self) -> None: + self.vroom.return_value = HTTPResponse(status=200, body=b'{"occurrences":1}') + with self.feature({"organizations:profiling-sentry-regression-occurrences": False}): + assert self.emit() == (1, 0) + assert self.vroom.call_args.kwargs["path"] == "/regressed" + assert self.broker.consume(Partition(self.topic, 0), 0) is None + + def test_mixed_organizations(self) -> None: + other_project = self.create_project(organization=self.create_organization()) + self.query.return_value["data"].append( + { + "project.id": other_project.id, + "fingerprint": 1176630088, + "all_examples()": [{"profile_id": "c" * 32}], + } + ) + self.vroom.side_effect = [ + self.vroom.return_value, + HTTPResponse(status=200, body=b'{"occurrences":1}'), + ] + with self.feature( + {"organizations:profiling-sentry-regression-occurrences": [self.organization.slug]} + ): + assert emit_function_regression_issue( + {self.project.id: self.project, other_project.id: other_project}, + [self.regression(self.project), self.regression(other_project)], + self.now, + ) == (1, 1) + assert self.vroom.call_count == 2 + inline, legacy = self.vroom.call_args_list + assert inline.kwargs["path"] == "/regressed?produce=false" + assert [p["project_id"] for p in inline.kwargs["json_data"]] == [self.project.id] + assert legacy.kwargs["path"] == "/regressed" + assert [p["project_id"] for p in legacy.kwargs["json_data"]] == [other_project.id] + assert self.broker.consume(Partition(self.topic, 0), 0) is not None + assert self.broker.consume(Partition(self.topic, 0), 1) is None + + def test_empty_response(self) -> None: + self.vroom.return_value = HTTPResponse( + status=200, body=b'{"occurrences":0,"emitted":[],"occurrence_payloads":[]}' + ) + with self.feature("organizations:profiling-sentry-regression-occurrences"): + assert self.emit() == (0, 0) + assert self.broker.consume(Partition(self.topic, 0), 0) is None + + def test_no_examples(self) -> None: + self.query.return_value = {"data": []} + assert self.emit() == (0, 0) + self.vroom.assert_not_called() + assert self.broker.consume(Partition(self.topic, 0), 0) is None + + def test_failed_response(self) -> None: + self.vroom.return_value = HTTPResponse(status=500) + with self.feature("organizations:profiling-sentry-regression-occurrences"): + assert self.emit() == (0, 0) + assert self.broker.consume(Partition(self.topic, 0), 0) is None + + def test_missing_payloads_does_not_fall_back(self) -> None: + self.vroom.return_value = HTTPResponse(status=200, body=b'{"occurrences":1}') + with ( + self.feature("organizations:profiling-sentry-regression-occurrences"), + pytest.raises(KeyError, match="occurrence_payloads"), + ): + self.emit() + assert self.vroom.call_count == 1 + assert self.broker.consume(Partition(self.topic, 0), 0) is None + + class FunctionsTasksTest(ProfilesSnubaTestCase): def setUp(self) -> None: super().setUp()