ref(transactions): add functionality to sample transactions in save_event_transaction (#81077)

first PR of #81065

no behavior changes until option is turned on. need to change
post_process as well before we can turn the option on. equivalent of
https://github.com/getsentry/sentry/blob/b84d3d9238ab75822b076d16de1111c3035fe073/src/sentry/tasks/post_process.py#L614
in post_process

Author: JoshFerge

src/sentry/event_manager.py

@@ -47,6 +47,7 @@
 from sentry.eventtypes import EventType
 from sentry.eventtypes.transaction import TransactionEvent
 from sentry.exceptions import HashDiscarded
+from sentry.features.rollout import in_rollout_group
 from sentry.grouping.api import (
     NULL_GROUPHASH_INFO,
     GroupHashInfo,
@@ -70,6 +71,9 @@
 )
 from sentry.grouping.variants import BaseVariant
 from sentry.ingest.inbound_filters import FilterStatKeys
+from sentry.ingest.transaction_clusterer.datasource.redis import (
+    record_transaction_name as record_transaction_name_for_clustering,
+)
 from sentry.integrations.tasks.kick_off_status_syncs import kick_off_status_syncs
 from sentry.issues.grouptype import ErrorGroupType
 from sentry.issues.issue_occurrence import IssueOccurrence
@@ -99,6 +103,8 @@
 from sentry.net.http import connection_from_url
 from sentry.plugins.base import plugins
 from sentry.quotas.base import index_data_category
+from sentry.receivers.features import record_event_processed
+from sentry.receivers.onboarding import record_release_received, record_user_context_received
 from sentry.reprocessing2 import is_reprocessed_event
 from sentry.seer.signed_seer_api import make_signed_seer_api_request
 from sentry.signals import (
@@ -2512,6 +2518,34 @@ def _detect_performance_problems(
         )
 
 
+@sentry_sdk.tracing.trace
+def _record_transaction_info(jobs: Sequence[Job], projects: ProjectsMapping) -> None:
+    """
+    this function does what we do in post_process for transactions. if this option is
+    turned on, we do the actions here instead of in post_process, with the goal
+    eventually being to not run transactions through post_process
+    """
+    for job in jobs:
+        try:
+            event = job["event"]
+            if not in_rollout_group("transactions.do_post_process_in_save", event.event_id):
+                continue
+
+            project = event.project
+            with sentry_sdk.start_span(op="event_manager.record_transaction_name_for_clustering"):
+                record_transaction_name_for_clustering(project, event.data)
+
+            # these are what the "transaction_processed" signal hooked into
+            # we should not use signals here, so call the recievers directly
+            # instead of sending a signal. we should consider potentially
+            # deleting these
+            record_event_processed(project, event)
+            record_user_context_received(project, event)
+            record_release_received(project, event)
+        except Exception:
+            sentry_sdk.capture_exception()
+
+
 class PerformanceJob(TypedDict, total=False):
     performance_problems: Sequence[PerformanceProblem]
     event: Event
@@ -2637,6 +2671,9 @@ def save_transaction_events(jobs: Sequence[Job], projects: ProjectsMapping) -> S
     with metrics.timer("save_transaction_events.send_occurrence_to_platform"):
         _send_occurrence_to_platform(jobs, projects)
 
+    with metrics.timer("save_transaction_events.record_transaction_info"):
+        _record_transaction_info(jobs, projects)
+
     return jobs
 
 

src/sentry/options/defaults.py

@@ -2920,3 +2920,9 @@
     default=0.0,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+
+register(
+    "transactions.do_post_process_in_save",
+    default=0.0,
+    flags=FLAG_AUTOMATOR_MODIFIABLE | FLAG_RATE,
+)

src/sentry/tasks/post_process.py

@@ -16,6 +16,7 @@
 from sentry import features, projectoptions
 from sentry.eventstream.types import EventStreamEventType
 from sentry.exceptions import PluginError
+from sentry.features.rollout import in_rollout_group
 from sentry.issues.grouptype import GroupCategory
 from sentry.issues.issue_occurrence import IssueOccurrence
 from sentry.killswitches import killswitch_matches_context
@@ -480,6 +481,17 @@ def should_update_escalating_metrics(event: Event, is_transaction_event: bool) -
     )
 
 
+def _get_event_id_from_cache_key(cache_key: str) -> str | None:
+    """
+    format is "e:{}:{}",event_id,project_id
+    """
+
+    try:
+        return cache_key.split(":")[1]
+    except IndexError:
+        return None
+
+
 @instrumented_task(
     name="sentry.tasks.post_process.post_process_group",
     time_limit=120,
@@ -527,6 +539,18 @@ def post_process_group(
             # need to rewind history.
             data = processing_store.get(cache_key)
             if not data:
+                event_id = _get_event_id_from_cache_key(cache_key)
+                if event_id:
+                    if in_rollout_group(
+                        "transactions.do_post_process_in_save",
+                        event_id,
+                    ):
+                        # if we're doing the work for transactions in save_event_transaction
+                        # instead of here, this is expected, so simply increment a metric
+                        # instead of logging
+                        metrics.incr("post_process.skipped_do_post_process_in_save")
+                        return
+
                 logger.info(
                     "post_process.skipped",
                     extra={"cache_key": cache_key, "reason": "missing_cache"},

src/sentry/tasks/store.py

@@ -16,6 +16,7 @@
 from sentry.constants import DEFAULT_STORE_NORMALIZER_ARGS
 from sentry.datascrubbing import scrub_data
 from sentry.eventstore import processing
+from sentry.features.rollout import in_rollout_group
 from sentry.feedback.usecases.create_feedback import FeedbackCreationSource, create_feedback_issue
 from sentry.ingest.types import ConsumerType
 from sentry.killswitches import killswitch_matches_context
@@ -582,6 +583,16 @@ def _do_save_event(
             raise
 
         finally:
+            if (
+                consumer_type == ConsumerType.Transactions
+                and event_id
+                and in_rollout_group("transactions.do_post_process_in_save", event_id)
+            ):
+                # we won't use the transaction data in post_process
+                # so we can delete it from the cache now.
+                if cache_key:
+                    processing_store.delete_by_key(cache_key)
+
             reprocessing2.mark_event_reprocessed(data)
             if cache_key and has_attachments:
                 attachment_cache.delete(cache_key)

tests/sentry/event_manager/test_event_manager.py

@@ -49,6 +49,7 @@
 from sentry.grouping.api import load_grouping_config
 from sentry.grouping.utils import hash_from_values
 from sentry.ingest.inbound_filters import FilterStatKeys
+from sentry.ingest.transaction_clusterer import ClustererNamespace
 from sentry.integrations.models.external_issue import ExternalIssue
 from sentry.integrations.models.integration import Integration
 from sentry.issues.grouptype import (
@@ -1540,6 +1541,140 @@ def test_transaction_event_span_grouping(self) -> None:
         # the basic strategy is to simply use the description
         assert spans == [{"hash": hash_values([span["description"]])} for span in data["spans"]]
 
+    @override_options({"transactions.do_post_process_in_save": 1.0})
+    def test_transaction_sampler_and_receive(self) -> None:
+        # make sure with the option on we don't get any errors
+        manager = EventManager(
+            make_event(
+                **{
+                    "transaction": "wait",
+                    "contexts": {
+                        "trace": {
+                            "parent_span_id": "bce14471e0e9654d",
+                            "op": "foobar",
+                            "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                            "span_id": "bf5be759039ede9a",
+                        }
+                    },
+                    "spans": [
+                        {
+                            "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                            "parent_span_id": "bf5be759039ede9a",
+                            "span_id": "a" * 16,
+                            "start_timestamp": 0,
+                            "timestamp": 1,
+                            "same_process_as_parent": True,
+                            "op": "default",
+                            "description": "span a",
+                        },
+                        {
+                            "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                            "parent_span_id": "bf5be759039ede9a",
+                            "span_id": "b" * 16,
+                            "start_timestamp": 0,
+                            "timestamp": 1,
+                            "same_process_as_parent": True,
+                            "op": "default",
+                            "description": "span a",
+                        },
+                        {
+                            "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                            "parent_span_id": "bf5be759039ede9a",
+                            "span_id": "c" * 16,
+                            "start_timestamp": 0,
+                            "timestamp": 1,
+                            "same_process_as_parent": True,
+                            "op": "default",
+                            "description": "span b",
+                        },
+                    ],
+                    "timestamp": "2019-06-14T14:01:40Z",
+                    "start_timestamp": "2019-06-14T14:01:40Z",
+                    "type": "transaction",
+                    "transaction_info": {
+                        "source": "url",
+                    },
+                }
+            )
+        )
+        manager.normalize()
+        manager.save(self.project.id)
+
+    @override_options({"transactions.do_post_process_in_save": 1.0})
+    @patch("sentry.event_manager.record_event_processed")
+    @patch("sentry.event_manager.record_user_context_received")
+    @patch("sentry.event_manager.record_release_received")
+    @patch("sentry.ingest.transaction_clusterer.datasource.redis._record_sample")
+    def test_transaction_sampler_and_receive_mock_called(
+        self,
+        mock_record_sample: mock.MagicMock,
+        mock_record_release: mock.MagicMock,
+        mock_record_user: mock.MagicMock,
+        mock_record_event: mock.MagicMock,
+    ) -> None:
+        manager = EventManager(
+            make_event(
+                **{
+                    "transaction": "wait",
+                    "contexts": {
+                        "trace": {
+                            "parent_span_id": "bce14471e0e9654d",
+                            "op": "foobar",
+                            "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                            "span_id": "bf5be759039ede9a",
+                        }
+                    },
+                    "spans": [
+                        {
+                            "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                            "parent_span_id": "bf5be759039ede9a",
+                            "span_id": "a" * 16,
+                            "start_timestamp": 0,
+                            "timestamp": 1,
+                            "same_process_as_parent": True,
+                            "op": "default",
+                            "description": "span a",
+                        },
+                        {
+                            "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                            "parent_span_id": "bf5be759039ede9a",
+                            "span_id": "b" * 16,
+                            "start_timestamp": 0,
+                            "timestamp": 1,
+                            "same_process_as_parent": True,
+                            "op": "default",
+                            "description": "span a",
+                        },
+                        {
+                            "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
+                            "parent_span_id": "bf5be759039ede9a",
+                            "span_id": "c" * 16,
+                            "start_timestamp": 0,
+                            "timestamp": 1,
+                            "same_process_as_parent": True,
+                            "op": "default",
+                            "description": "span b",
+                        },
+                    ],
+                    "timestamp": "2019-06-14T14:01:40Z",
+                    "start_timestamp": "2019-06-14T14:01:40Z",
+                    "type": "transaction",
+                    "transaction_info": {
+                        "source": "url",
+                    },
+                }
+            )
+        )
+        manager.normalize()
+        event = manager.save(self.project.id)
+
+        mock_record_event.assert_called_once_with(self.project, event)
+        mock_record_user.assert_called_once_with(self.project, event)
+        mock_record_release.assert_called_once_with(self.project, event)
+        assert mock_record_sample.mock_calls == [
+            mock.call(ClustererNamespace.TRANSACTIONS, self.project, "wait")
+        ]
+
     def test_sdk(self) -> None:
         manager = EventManager(make_event(**{"sdk": {"name": "sentry-unity", "version": "1.0"}}))
         manager.normalize()

tests/sentry/tasks/test_post_process.py

@@ -60,6 +60,7 @@
 from sentry.tasks.post_process import (
     HIGHER_ISSUE_OWNERS_PER_PROJECT_PER_MIN_RATELIMIT,
     ISSUE_OWNERS_PER_PROJECT_PER_MIN_RATELIMIT,
+    _get_event_id_from_cache_key,
     feedback_filter_decorator,
     locks,
     post_process_group,
@@ -2735,6 +2736,60 @@ def test_process_transaction_event_clusterer(
         ]
 
 
+class ProcessingStoreTransactionEmptyTestcase(TestCase):
+    @patch("sentry.tasks.post_process.logger")
+    def test_logger_called_when_empty(self, mock_logger):
+        post_process_group(
+            is_new=False,
+            is_regression=False,
+            is_new_group_environment=False,
+            cache_key="e:1:2",
+            group_id=None,
+            project_id=self.project.id,
+            eventstream_type=EventStreamEventType.Transaction,
+        )
+        assert mock_logger.info.called
+        mock_logger.info.assert_called_with(
+            "post_process.skipped", extra={"cache_key": "e:1:2", "reason": "missing_cache"}
+        )
+
+    @patch("sentry.tasks.post_process.logger")
+    @patch("sentry.utils.metrics.incr")
+    @override_options({"transactions.do_post_process_in_save": 1.0})
+    def test_logger_called_when_empty_option_on(self, mock_metric_incr, mock_logger):
+        post_process_group(
+            is_new=False,
+            is_regression=False,
+            is_new_group_environment=False,
+            cache_key="e:1:2",
+            group_id=None,
+            project_id=self.project.id,
+            eventstream_type=EventStreamEventType.Transaction,
+        )
+        assert not mock_logger.info.called
+        mock_metric_incr.assert_called_with("post_process.skipped_do_post_process_in_save")
+
+    @patch("sentry.tasks.post_process.logger")
+    @override_options({"transactions.do_post_process_in_save": 1.0})
+    def test_logger_called_when_empty_option_on_invalid_cache_key(self, mock_logger):
+        post_process_group(
+            is_new=False,
+            is_regression=False,
+            is_new_group_environment=False,
+            cache_key="invalidhehe",
+            group_id=None,
+            project_id=self.project.id,
+            eventstream_type=EventStreamEventType.Transaction,
+        )
+        mock_logger.info.assert_called_with(
+            "post_process.skipped", extra={"cache_key": "invalidhehe", "reason": "missing_cache"}
+        )
+
+    def test_get_event_id_from_cache_key(self):
+        assert _get_event_id_from_cache_key("e:1:2") == "1"
+        assert _get_event_id_from_cache_key("invalid") is None
+
+
 class PostProcessGroupGenericTest(
     TestCase,
     SnubaTestCase,