ref(transactions): dont send transactions to post_process

src/sentry/event_manager.py

@@ -2655,6 +2655,12 @@ def save_transaction_events(jobs: Sequence[Job], projects: ProjectsMapping) -> S
         _nodestore_save_many(jobs=jobs, app_feature="transactions")
 
     with metrics.timer("save_transaction_events.eventstream_insert_many"):
+        for job in jobs:
+            if options.get("save_event_transactions.post_process_cleanup"):
+                # we don't need to send transactions to post process
+                # so set raw so we skip post_process
+                job["raw"] = True
+
         _eventstream_insert_many(jobs)
 
     with metrics.timer("save_transaction_events.track_outcome_accepted_many"):

src/sentry/options/defaults.py

@@ -2921,3 +2921,9 @@
     default=False,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+register(
+    "save_event_transactions.post_process_cleanup",
+    type=Bool,
+    default=False,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)

src/sentry/tasks/store.py

@@ -582,6 +582,14 @@ def _do_save_event(
             raise
 
         finally:
+            if consumer_type == ConsumerType.Transactions and options.get(
+                "save_event_transactions.post_process_cleanup"
+            ):
+                # 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

@@ -1603,10 +1603,12 @@ def test_transaction_sampler_and_recieve(self) -> None:
     @override_options({"save_event_transactions.sample_transactions_in_save": True})
     @patch("sentry.signals.transaction_processed.send_robust")
     @patch("sentry.ingest.transaction_clusterer.datasource.redis._record_sample")
+    @mock.patch("sentry.event_manager.eventstream.backend.insert")
     def test_transaction_sampler_and_recieve_mock_called(
         self,
-        mock_record_sample: mock.MagicMock,
         transaction_processed_signal_mock: mock.MagicMock,
+        mock_record_sample: mock.MagicMock,
+        eventstream_insert: mock.MagicMock,
     ) -> None:
         manager = EventManager(
             make_event(
@@ -1667,6 +1669,72 @@ def test_transaction_sampler_and_recieve_mock_called(
         assert mock_record_sample.mock_calls == [
             mock.call(ClustererNamespace.TRANSACTIONS, self.project, "wait")
         ]
+        eventstream_insert.assert_called_once()
+        assert "skip_consume" not in eventstream_insert.call_args.kwargs
+
+    @override_options({"save_event_transactions.post_process_cleanup": True})
+    @mock.patch("sentry.event_manager.eventstream.backend.insert")
+    def test_transaction_event_stream_insert_with_raw(
+        self, eventstream_insert: mock.MagicMock
+    ) -> 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)
+        eventstream_insert.assert_called_once()
+        assert eventstream_insert.call_args.kwargs["skip_consume"] is True
 
     def test_sdk(self) -> None:
         manager = EventManager(make_event(**{"sdk": {"name": "sentry-unity", "version": "1.0"}}))

tests/sentry/tasks/test_store.py

@@ -14,6 +14,7 @@
     save_event,
     save_event_transaction,
 )
+from sentry.testutils.helpers.options import override_options
 from sentry.testutils.pytest.fixtures import django_db_all
 
 EVENT_ID = "cc3e6c2bb6b6498097f336d1e6979f4b"
@@ -397,3 +398,64 @@ def test_store_consumer_type(
         )
 
     mock_transaction_processing_store.get.assert_called_once_with("tx:3")
+    mock_transaction_processing_store.delete_by_key.assert_not_called()
+
+
+@django_db_all
+@override_options({"save_event_transactions.post_process_cleanup": True})
+def test_store_consumer_type_transactions_cleanup(
+    default_project,
+    mock_save_event,
+    register_plugin,
+    mock_event_processing_store,
+    mock_transaction_processing_store,
+):
+    register_plugin(globals(), BasicPreprocessorPlugin)
+
+    data = {
+        "project": default_project.id,
+        "platform": "python",
+        "logentry": {"formatted": "test"},
+        "event_id": EVENT_ID,
+        "extra": {"foo": "bar"},
+        "timestamp": time(),
+    }
+
+    mock_event_processing_store.get.return_value = data
+    mock_event_processing_store.store.return_value = "e:2"
+
+    process_event(cache_key="e:2", start_time=1)
+
+    mock_event_processing_store.get.assert_called_once_with("e:2")
+
+    mock_save_event.delay.assert_called_once_with(
+        cache_key="e:2",
+        data=None,
+        start_time=1,
+        event_id=EVENT_ID,
+        project_id=default_project.id,
+    )
+
+    transaction_data = {
+        "project": default_project.id,
+        "platform": "transaction",
+        "event_id": EVENT_ID,
+        "extra": {"foo": "bar"},
+        "timestamp": time(),
+        "start_timestamp": time() - 1,
+    }
+
+    mock_transaction_processing_store.get.return_value = transaction_data
+    mock_transaction_processing_store.store.return_value = "tx:3"
+
+    with mock.patch("sentry.event_manager.EventManager.save", return_value=None):
+        save_event_transaction(
+            cache_key="tx:3",
+            data=None,
+            start_time=1,
+            event_id=EVENT_ID,
+            project_id=default_project.id,
+        )
+
+    mock_transaction_processing_store.get.assert_called_once_with("tx:3")
+    mock_transaction_processing_store.delete_by_key.assert_called_once_with("tx:3")