ref(event_manager): Inline signals into _record_transaction_info (#86604)

Centralizes all transaction-related side effects in
`_record_transaction_info` so it can be moved into the new spans
consumer.

- Two signals were emitted directly in `event_manager.save`. The
  underlying functions are now called directly and the calls have been
  moved in.
- We no longer connect the `record_*` functions that were called
  directly to their old signals.

Author: jan-auer

src/sentry/event_manager.py

@@ -103,14 +103,16 @@
 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
+from sentry.receivers.onboarding import (
+    record_first_insight_span,
+    record_first_transaction,
+    record_release_received,
+)
 from sentry.reprocessing2 import is_reprocessed_event
 from sentry.seer.signed_seer_api import make_signed_seer_api_request
 from sentry.signals import (
     first_event_received,
     first_event_with_minified_stack_trace_received,
-    first_insight_span_received,
-    first_transaction_received,
     issue_unresolved,
 )
 from sentry.tasks.process_buffer import buffer_incr
@@ -454,23 +456,11 @@ def save(
         event_type = self._data.get("type")
         if event_type == "transaction":
             job["data"]["project"] = project.id
-            jobs = save_transaction_events([job], projects)
-
-            if not project.flags.has_transactions and not skip_send_first_transaction:
-                first_transaction_received.send_robust(
-                    project=project, event=jobs[0]["event"], sender=Project
-                )
-
-            for module, is_module in INSIGHT_MODULE_FILTERS.items():
-                if not get_project_insight_flag(project, module) and is_module(job["data"]):
-                    first_insight_span_received.send_robust(
-                        project=project, module=module, sender=Project
-                    )
+            jobs = save_transaction_events([job], projects, skip_send_first_transaction)
             return jobs[0]["event"]
         elif event_type == "generic":
             job["data"]["project"] = project.id
             jobs = save_generic_events([job], projects)
-
             return jobs[0]["event"]
         else:
             project = job["event"].project
@@ -2459,7 +2449,9 @@ def _detect_performance_problems(jobs: Sequence[Job], projects: ProjectsMapping)
 
 
 @sentry_sdk.tracing.trace
-def _record_transaction_info(jobs: Sequence[Job], projects: ProjectsMapping) -> None:
+def _record_transaction_info(
+    jobs: Sequence[Job], projects: ProjectsMapping, skip_send_first_transaction: bool
+) -> None:
     for job in jobs:
         try:
             event = job["event"]
@@ -2468,16 +2460,19 @@ def _record_transaction_info(jobs: Sequence[Job], projects: ProjectsMapping) ->
             with sentry_sdk.start_span(op="event_manager.record_transaction_name_for_clustering"):
                 record_transaction_name_for_clustering(project, event.data)
 
+            record_event_processed(project, event)
+
+            if not skip_send_first_transaction:
+                record_first_transaction(project, event.datetime)
+
+            for module, is_module in INSIGHT_MODULE_FILTERS.items():
+                if not get_project_insight_flag(project, module) and is_module(job["data"]):
+                    record_first_insight_span(project, module)
+
             if job["release"]:
                 environment = job["data"].get("environment") or None  # coorce "" to None
                 record_latest_release(project, job["release"], environment)
-
-            # 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_release_received(project, event)
+                record_release_received(project, job["release"].version)
         except Exception:
             sentry_sdk.capture_exception()
 
@@ -2546,7 +2541,11 @@ def _send_occurrence_to_platform(jobs: Sequence[Job], projects: ProjectsMapping)
 
 
 @sentry_sdk.tracing.trace
-def save_transaction_events(jobs: Sequence[Job], projects: ProjectsMapping) -> Sequence[Job]:
+def save_transaction_events(
+    jobs: Sequence[Job],
+    projects: ProjectsMapping,
+    skip_send_first_transaction: bool = False,
+) -> Sequence[Job]:
     from .ingest.types import ConsumerType
 
     organization_ids = {project.organization_id for project in projects.values()}
@@ -2589,7 +2588,7 @@ def save_transaction_events(jobs: Sequence[Job], projects: ProjectsMapping) -> S
     _track_outcome_accepted_many(jobs)
     _detect_performance_problems(jobs, projects)
     _send_occurrence_to_platform(jobs, projects)
-    _record_transaction_info(jobs, projects)
+    _record_transaction_info(jobs, projects, skip_send_first_transaction)
 
     return jobs
 

src/sentry/receivers/features.py

@@ -96,42 +96,62 @@ def record_first_event(project, **kwargs):
 
 
 def record_event_processed(project, event, **kwargs):
-    feature_slugs = []
+    return record_generic_event_processed(
+        project,
+        platform=event.group.platform if event.group else event.platform,
+        release=event.get_tag("sentry:release"),
+        environment=event.get_tag("environment"),
+        user_keys=event.data.get("user", {}).keys(),
+        tag_keys={tag[0] for tag in event.tags},
+        has_sourcemap=has_sourcemap(event),
+        has_breadcrumbs=event.data.get("breadcrumbs"),
+    )
 
-    platform = event.group.platform if event.group else event.platform
+
+def record_generic_event_processed(
+    project,
+    platform=None,
+    release=None,
+    environment=None,
+    user_keys=None,
+    tag_keys=None,
+    has_sourcemap=False,
+    has_breadcrumbs=False,
+    **kwargs,
+):
+    feature_slugs = []
 
     # Platform
     if platform in manager.location_slugs("language"):
         feature_slugs.append(platform)
 
     # Release Tracking
-    if event.get_tag("sentry:release"):
+    if release:
         feature_slugs.append("release_tracking")
 
     # Environment Tracking
-    if event.get_tag("environment"):
+    if environment:
         feature_slugs.append("environment_tracking")
 
     # User Tracking
-    user_context = event.data.get("user")
     # We'd like them to tag with id or email.
     # Certain SDKs automatically tag with ip address.
     # Check to make sure more the ip address is being sent.
     # testing for this in test_no_user_tracking_for_ip_address_only
     # list(d.keys()) pattern is to make this python3 safe
-    if user_context and len(user_context.keys() - {"ip_address", "sentry_user"}) > 0:
+    if user_keys and len(set(user_keys) - {"ip_address", "sentry_user"}) > 0:
         feature_slugs.append("user_tracking")
 
     # Custom Tags
-    if {tag[0] for tag in event.tags} - DEFAULT_TAGS:
+    if tag_keys and set(tag_keys) - DEFAULT_TAGS:
         feature_slugs.append("custom_tags")
 
     # Sourcemaps
-    if has_sourcemap(event):
+    if has_sourcemap:
         feature_slugs.append("source_maps")
 
     # Breadcrumbs
-    if event.data.get("breadcrumbs"):
+    if has_breadcrumbs:
         feature_slugs.append("breadcrumbs")
 
     if not feature_slugs:

src/sentry/receivers/onboarding.py

@@ -224,14 +224,21 @@ def record_first_event(project, event, **kwargs):
 
 
 @first_transaction_received.connect(weak=False)
-def record_first_transaction(project, event, **kwargs):
+def _record_first_transaction(project, event, **kwargs):
+    return record_first_transaction(project, event.datetime, **kwargs)
+
+
+def record_first_transaction(project, datetime, **kwargs):
+    if project.flags.has_transactions:
+        return
+
     project.update(flags=F("flags").bitor(Project.flags.has_transactions))
 
     OrganizationOnboardingTask.objects.record(
         organization_id=project.organization_id,
         task=OnboardingTask.FIRST_TRANSACTION,
         status=OnboardingTaskStatus.COMPLETE,
-        date_completed=event.datetime,
+        date_completed=datetime,
     )
 
     try:
@@ -430,8 +437,12 @@ def record_member_joined(organization_id: int, organization_member_id: int, **kw
     )
 
 
-def record_release_received(project, event, **kwargs):
-    if not event.data.get("release"):
+def _record_release_received(project, event, **kwargs):
+    return record_release_received(project, event.data.get("release"), **kwargs)
+
+
+def record_release_received(project, release, **kwargs):
+    if not release:
         return
 
     success = OrganizationOnboardingTask.objects.record(
@@ -459,8 +470,8 @@ def record_release_received(project, event, **kwargs):
         )
 
 
-event_processed.connect(record_release_received, weak=False)
-transaction_processed.connect(record_release_received, weak=False)
+event_processed.connect(_record_release_received, weak=False)
+transaction_processed.connect(_record_release_received, weak=False)
 
 
 @first_event_with_minified_stack_trace_received.connect(weak=False)

src/sentry/spans/consumers/process_segments/message.py

@@ -237,6 +237,6 @@ def process_segment(spans: list[Span]) -> list[Span]:
     job: Job = {"data": event, "project_id": project.id, "raw": False, "start_time": None}
 
     _pull_out_data([job], projects)
-    _record_transaction_info([job], projects)
+    _record_transaction_info([job], projects, skip_send_first_transaction=False)
 
     return spans

tests/sentry/event_manager/test_event_manager.py

@@ -16,11 +16,12 @@
 from arroyo.types import Partition, Topic
 from django.conf import settings
 from django.core.cache import cache
+from django.db.models import F
 from django.utils import timezone
 
 from sentry import eventstore, nodestore, tsdb
 from sentry.attachments import CachedAttachment, attachment_cache
-from sentry.constants import MAX_VERSION_LENGTH, DataCategory
+from sentry.constants import MAX_VERSION_LENGTH, DataCategory, InsightModules
 from sentry.dynamic_sampling import (
     ExtendedBoostedRelease,
     Platform,
@@ -59,6 +60,7 @@
 from sentry.models.grouprelease import GroupRelease
 from sentry.models.groupresolution import GroupResolution
 from sentry.models.grouptombstone import GroupTombstone
+from sentry.models.project import Project
 from sentry.models.pullrequest import PullRequest, PullRequestCommit
 from sentry.models.release import Release
 from sentry.models.releasecommit import ReleaseCommit
@@ -1577,15 +1579,19 @@ def test_transaction_sampler_and_receive(self) -> None:
         manager.normalize()
         manager.save(self.project.id)
 
-    @patch("sentry.event_manager.record_event_processed")
+    @patch("sentry.event_manager.record_first_transaction")
+    @patch("sentry.event_manager.record_first_insight_span")
     @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_event: mock.MagicMock,
+        mock_record_insight: mock.MagicMock,
+        mock_record_transaction: mock.MagicMock,
     ) -> None:
+        self.project.update(flags=F("flags").bitand(~Project.flags.has_transactions))
+
         manager = EventManager(
             make_event(
                 **{
@@ -1606,8 +1612,14 @@ def test_transaction_sampler_and_receive_mock_called(
                             "start_timestamp": 0,
                             "timestamp": 1,
                             "same_process_as_parent": True,
-                            "op": "default",
-                            "description": "span a",
+                            "op": "db.redis",
+                            "description": "EXEC *",
+                            "sentry_tags": {
+                                "description": "EXEC *",
+                                "category": "db",
+                                "op": "db.redis",
+                                "transaction": "/app/index",
+                            },
                         },
                         {
                             "trace_id": "a0fa8803753e40fd8124b21eeb2986b5",
@@ -1633,6 +1645,7 @@ def test_transaction_sampler_and_receive_mock_called(
                     "timestamp": "2019-06-14T14:01:40Z",
                     "start_timestamp": "2019-06-14T14:01:40Z",
                     "type": "transaction",
+                    "release": "foo@1.0.0",
                     "transaction_info": {
                         "source": "url",
                     },
@@ -1642,8 +1655,9 @@ def test_transaction_sampler_and_receive_mock_called(
         manager.normalize()
         event = manager.save(self.project.id)
 
-        mock_record_event.assert_called_once_with(self.project, event)
-        mock_record_release.assert_called_once_with(self.project, event)
+        mock_record_release.assert_called_once_with(self.project, "foo@1.0.0")
+        mock_record_insight.assert_called_once_with(self.project, InsightModules.DB)
+        mock_record_transaction.assert_called_once_with(self.project, event.datetime)
         assert mock_record_sample.mock_calls == [
             mock.call(ClustererNamespace.TRANSACTIONS, self.project, "wait")
         ]