Repository: getsentry/sentry
Pull request: feat(issue-platform): Validate and save Occurrence + Event Messages
Processes Kafka payloads for incoming Occurrences and associated Events

Closes https://github.com/getsentry/sentry/issues/42053 and closes https://github.com/getsentry/sentry/issues/42036

Problem statement:
Ingestion: Validate issue occurrence messages
Parse and validate issue occurrence messages in issue occurrence consumer, and convert to the `IssueOccurrenceData` (draft forthcoming) typeddict to pass around to the persistence layer

Ingestion: When passed an event, save via event manager
If we had an `Event` passed to us in the `IssueOccurrenceData` then persist it via `EventManager`. We'll probably want a `save_platform_events` function that is cut down a fair bit. It shouldn't do:
 - Passing to eventstream
 - Saving the issue

Just should be concerned with saving the event.

Hints from issue discussion:

Commit message: added log statement

Patch to review:
diff --git a/src/sentry/issues/occurrence_consumer.py b/src/sentry/issues/occurrence_consumer.py
index 41c8d877f5c695..f85b2bf109a91b 100644
--- a/src/sentry/issues/occurrence_consumer.py
+++ b/src/sentry/issues/occurrence_consumer.py
@@ -1,6 +1,7 @@
 import logging
-from typing import Any, Callable, Mapping, Optional
+from typing import Any, Callable, Dict, Mapping, Optional
 
+import rapidjson
 from arroyo import Topic
 from arroyo.backends.kafka.configuration import build_kafka_consumer_configuration
 from arroyo.backends.kafka.consumer import KafkaConsumer, KafkaPayload
@@ -10,8 +11,16 @@
 from arroyo.types import Message, Partition, Position
 from django.conf import settings
 
+from sentry.eventstore.models import Event
+from sentry.issues.ingest import save_issue_occurrence
+from sentry.issues.issue_occurrence import IssueOccurrence, IssueOccurrenceData
+from sentry.utils import json, metrics
+from sentry.utils.canonical import CanonicalKeyDict
 from sentry.utils.kafka_config import get_kafka_consumer_cluster_options
 
+_DURATION_METRIC = "occurrence_ingest.duration"
+_MESSAGES_METRIC = "occurrence_ingest.messages"
+
 logger = logging.getLogger(__name__)
 
 
@@ -43,6 +52,92 @@ def create_ingest_occurences_consumer(
     )
 
 
+def save_event_from_occurrence(
+    data: Dict[str, Any],
+    **kwargs: Any,
+) -> Event:
+
+    from sentry.event_manager import EventManager
+
+    event_type = "platform"
+
+    with metrics.global_tags(event_type=event_type):
+        project_id = data["data"].pop("project_id")
+        data = CanonicalKeyDict(data)
+
+        with metrics.timer("occurrence_consumer.save_event_occurrence.event_manager.save"):
+            manager = EventManager(data)
+            event = manager.save(project_id=project_id)
+
+            return event
+
+
+def process_event_and_issue_occurrence(
+    occurrence_data: IssueOccurrenceData, event_data: Dict[str, Any]
+) -> Optional[IssueOccurrence]:
+    try:
+        event = save_event_from_occurrence(event_data)
+    except Exception:
+        logger.error("error saving message")
+        return None
+
+    occurrence_data["event_id"] = event.event_id
+    return save_issue_occurrence(occurrence_data, event)
+
+
+def _get_kwargs(payload: Mapping[str, Any]) -> Optional[Mapping[str, Any]]:
+    try:
+        with metrics.timer(_DURATION_METRIC, instance="get_task_kwargs_for_message"):
+            metrics.timing("occurrence.ingest.size.data", len(payload))
+
+            kwargs = {
+                "occurrence_data": {
+                    "id": payload["id"],
+                    "event_id": None,
+                    "fingerprint": payload["fingerprint"],
+                    "issue_title": payload["issue_title"],
+                    "subtitle": payload["subtitle"],
+                    "resource_id": payload.get("resource_id"),
+                    "evidence_data": payload.get("evidence_data"),
+                    "evidence_display": payload.get("evidence_display"),
+                    "type": payload["type"],
+                    "detection_time": payload["detection_time"],
+                }
+            }
+
+            if "event" in payload:
+                kwargs["event_data"] = {"data": payload["event"]}
+            else:
+                kwargs["occurrence_data"]["event_id"] = payload.get("event_id")
+
+            return kwargs
+
+    except (KeyError, ValueError):
+        logger.exception("invalid payload data")
+        return None
+
+
+def _process_message(message: Mapping[str, Any]) -> Optional[IssueOccurrence]:
+    kwargs = _get_kwargs(message)
+    if not kwargs:
+        return None
+
+    metrics.incr(
+        _MESSAGES_METRIC,
+        amount=1,
+        tags={},  # TODO
+        sample_rate=1,
+    )
+
+    if "event_data" in kwargs:
+        return process_event_and_issue_occurrence(**kwargs)  # returning this now for easier testing
+    else:
+        # all occurrences will have Event data, for now
+        pass
+
+    return None
+
+
 class OccurrenceStrategy(ProcessingStrategy[KafkaPayload]):
     def __init__(
         self,
@@ -55,7 +150,11 @@ def poll(self) -> None:
         pass
 
     def submit(self, message: Message[KafkaPayload]) -> None:
-        logger.info(f"OCCURRENCE RECEIVED: {message.payload}")
+        try:
+            payload = json.loads(message.payload.value, use_rapid_json=True)
+            _process_message(payload)
+        except rapidjson.JSONDecodeError:
+            logger.exception("invalid json received")
 
     def close(self) -> None:
         pass
diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py
index 3bd1db436a36d4..22a7a140c85376 100644
--- a/src/sentry/runner/commands/run.py
+++ b/src/sentry/runner/commands/run.py
@@ -569,7 +569,7 @@ def occurrences_ingest_consumer():
     from sentry.issues.occurrence_consumer import get_occurrences_ingest_consumer
     from sentry.utils import metrics
 
-    consumer_type = "occurrences"
+    consumer_type = "ingest-occurrences"
 
     with metrics.global_tags(ingest_consumer_types=consumer_type, _all_threads=True):
         get_occurrences_ingest_consumer(consumer_type).run()
diff --git a/tests/sentry/issues/test_occurrence_consumer.py b/tests/sentry/issues/test_occurrence_consumer.py
new file mode 100644
index 00000000000000..7a501edabf3a2f
--- /dev/null
+++ b/tests/sentry/issues/test_occurrence_consumer.py
@@ -0,0 +1,67 @@
+import datetime
+import logging
+from typing import Any, Dict
+
+import pytest
+
+from sentry.issues.issue_occurrence import IssueOccurrence
+from sentry.issues.occurrence_consumer import _process_message
+from sentry.testutils import TestCase
+from sentry.types.issues import GroupType
+from tests.sentry.issues.test_utils import OccurrenceTestMixin
+
+logger = logging.getLogger(__name__)
+
+
+def get_test_message(
+    project_id: str, include_event: bool = True, **overrides: Any
+) -> Dict[str, Any]:
+    now = datetime.datetime.now()
+    payload = {
+        "id": "{}".format(now.strftime("000000000000%Y%m%d%H%M%S%f")),
+        "event_id": None,
+        "fingerprint": ["touch-id"],
+        "issue_title": "segfault",
+        "subtitle": "buffer overflow",
+        "resource_id": "12345",
+        "evidence_data": {"Line": 40},
+        "evidence_display": [
+            {"name": "Line", "value": "40", "important": True},
+            {"name": "Memory", "value": "breached", "important": False},
+        ],
+        "type": GroupType.PERFORMANCE_N_PLUS_ONE_DB_QUERIES,
+        "detection_time": now.isoformat(),
+    }
+
+    if include_event:
+        payload["event"] = {
+            "project_id": project_id,
+            "title": "code's broken",
+            "platform": "genesis",
+            "tags": {},
+            "timestamp": now.isoformat(),
+            "message_timestamp": now.isoformat(),
+        }
+
+    payload.update(overrides)
+    return payload
+
+
+class IssueOccurrenceTestMessage(OccurrenceTestMixin, TestCase):  # type: ignore
+    @pytest.mark.django_db
+    def test_occurrence_consumer_with_event(self) -> None:
+        message = get_test_message(self.project.id)
+        occurrence = _process_message(message)
+        assert occurrence is not None
+
+        fetched_occurrence = IssueOccurrence.fetch(occurrence.id, self.project.id)
+        assert fetched_occurrence is not None
+        self.assert_occurrences_identical(occurrence, fetched_occurrence)
+
+    @pytest.mark.django_db
+    def test_invalid_payload(self) -> None:
+        pass
+        # need to add better typing during message processing before this passes
+        # message = get_test_message(self.project.id, type=300)
+        # occurrence = _process_message(message)
+        # assert occurrence is None


Code review comments:
diff_hunk: @@ -43,6 +52,92 @@ def create_ingest_occurences_consumer(
     )
 
 
+def save_event_from_occurrence(
+    data: Dict[str, Any],
+    **kwargs: Any,
+) -> Event:
+
+    from sentry.event_manager import EventManager
+
+    event_type = "platform"
path: src/sentry/issues/occurrence_consumer.py
text: @user1:
You'll need to inject this into the event data so that the event type changes 

@author:
whoops, good catch

diff_hunk: @@ -0,0 +1,67 @@
+import datetime
+import logging
+from typing import Any, Dict
+
+import pytest
+
+from sentry.issues.issue_occurrence import IssueOccurrence
+from sentry.issues.occurrence_consumer import _process_message
+from sentry.testutils import TestCase
+from sentry.types.issues import GroupType
+from tests.sentry.issues.test_utils import OccurrenceTestMixin
+
+logger = logging.getLogger(__name__)
+
+
+def get_test_message(
+    project_id: str, include_event: bool = True, **overrides: Any
+) -> Dict[str, Any]:
+    now = datetime.datetime.now()
+    payload = {
+        "id": "{}".format(now.strftime("000000000000%Y%m%d%H%M%S%f")),
+        "event_id": None,
+        "fingerprint": ["touch-id"],
+        "issue_title": "segfault",
+        "subtitle": "buffer overflow",
+        "resource_id": "12345",
+        "evidence_data": {"Line": 40},
+        "evidence_display": [
+            {"name": "Line", "value": "40", "important": True},
+            {"name": "Memory", "value": "breached", "important": False},
+        ],
+        "type": GroupType.PERFORMANCE_N_PLUS_ONE_DB_QUERIES,
path: tests/sentry/issues/test_occurrence_consumer.py
text: @user1:
Let's use the profiling grouptype here.

@author:
got it

diff_hunk: @@ -43,6 +52,92 @@ def create_ingest_occurences_consumer(
     )
 
 
+def save_event_from_occurrence(
+    data: Dict[str, Any],
+    **kwargs: Any,
+) -> Event:
+
+    from sentry.event_manager import EventManager
+
+    event_type = "platform"
+
+    with metrics.global_tags(event_type=event_type):
+        project_id = data["data"].pop("project_id")
+        data = CanonicalKeyDict(data)
+
+        with metrics.timer("occurrence_consumer.save_event_occurrence.event_manager.save"):
+            manager = EventManager(data)
+            event = manager.save(project_id=project_id)
path: src/sentry/issues/occurrence_consumer.py
text: @user1:
At the moment this will run through saving the event either as an error or a transaction. We need another branch in eventmanager for our new events. I'm not sure if this also means you'll need to add a new event type to `eventtypes/`

@author:
ok. will investigate

diff_hunk: @@ -569,7 +569,7 @@ def occurrences_ingest_consumer():
     from sentry.issues.occurrence_consumer import get_occurrences_ingest_consumer
     from sentry.utils import metrics
 
-    consumer_type = "occurrences"
+    consumer_type = "ingest-occurrences"
path: src/sentry/runner/commands/run.py
text: @user1:
Should we just use `settings.KAFKA_INGEST_OCCURRENCES` here?