feat(flags): Update errors processor to index feature flag data

rust_snuba/src/processors/errors.rs

@@ -161,6 +161,8 @@ type GenericContext = BTreeMap<String, ContextStringify>;
 
 #[derive(Debug, Default, Deserialize, JsonSchema)]
 struct Contexts {
+    #[serde(default)]
+    flags: Option<FlagContext>,
     #[serde(default)]
     replay: Option<ReplayContext>,
     #[serde(default)]
@@ -183,6 +185,20 @@ struct TraceContext {
     other: GenericContext,
 }
 
+#[derive(Debug, Default, Deserialize, JsonSchema)]
+struct FlagContext {
+    #[serde(default)]
+    values: Option<Vec<Option<FlagContextItem>>>,
+}
+
+#[derive(Debug, Default, Deserialize, JsonSchema)]
+struct FlagContextItem {
+    #[serde(default)]
+    flag: Unicodify,
+    #[serde(default)]
+    result: Unicodify,
+}
+
 #[derive(Debug, Default, Deserialize, JsonSchema)]
 struct ReplayContext {
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -384,6 +400,10 @@ struct ErrorRow {
     tags_key: Vec<String>,
     #[serde(rename = "tags.value")]
     tags_value: Vec<String>,
+    #[serde(rename = "flags.key")]
+    flags_key: Vec<String>,
+    #[serde(rename = "flags.value")]
+    flags_value: Vec<String>,
     timestamp: u32,
     title: String,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -563,6 +583,21 @@ impl ErrorRow {
             replay_id = Some(rid)
         }
 
+        // Split feature keys and values into two vectors if the context could be parsed.
+        let mut flags_key = Vec::new();
+        let mut flags_value = Vec::new();
+
+        if let Some(ctx) = from_context.flags {
+            if let Some(values) = ctx.values {
+                for item in values.into_iter().flatten() {
+                    if let (Some(k), Some(v)) = (item.flag.0, item.result.0) {
+                        flags_key.push(k);
+                        flags_value.push(v);
+                    }
+                }
+            }
+        };
+
         // Stacktrace.
 
         let exceptions = from
@@ -668,6 +703,8 @@ impl ErrorRow {
             exception_stacks_mechanism_type: stack_mechanism_types,
             exception_stacks_type: stack_types,
             exception_stacks_value: stack_values,
+            flags_key,
+            flags_value,
             group_id: from.group_id,
             http_method: from_request.method.0,
             http_referer,

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__events-.snap

@@ -3,6 +3,13 @@ source: src/processors/mod.rs
 expression: diff
 ---
 [
+    Change {
+        path: ".<anyOf:0>.2.data.contexts.<anyOf:0>",
+        change: PropertyAdd {
+            lhs_additional_properties: true,
+            added: "flags",
+        },
+    },
     Change {
         path: ".<anyOf:0>.2.data.contexts.<anyOf:0>.trace.<anyOf:0>",
         change: PropertyAdd {

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@events-ErrorsProcessor-events__1__error-with-null-threads.json.snap

@@ -262,6 +262,8 @@ expression: snapshot_payload
     "exception_stacks.value": [
       "Some exception."
     ],
+    "flags.key": [],
+    "flags.value": [],
     "group_id": 124,
     "http_method": null,
     "http_referer": null,

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@events-ErrorsProcessor-events__1__error-with-null-values-threads.json.snap

@@ -262,6 +262,8 @@ expression: snapshot_payload
     "exception_stacks.value": [
       "Some exception."
     ],
+    "flags.key": [],
+    "flags.value": [],
     "group_id": 124,
     "http_method": null,
     "http_referer": null,

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@events-ErrorsProcessor-events__1__error-with-threads.json.snap

@@ -263,6 +263,8 @@ expression: snapshot_payload
     "exception_stacks.value": [
       "Some exception."
     ],
+    "flags.key": [],
+    "flags.value": [],
     "group_id": 124,
     "http_method": null,
     "http_referer": null,

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@events-ErrorsProcessor-events__1__errors1.json.snap

@@ -25,6 +25,8 @@ expression: snapshot_payload
     "exception_stacks.mechanism_type": [],
     "exception_stacks.type": [],
     "exception_stacks.value": [],
+    "flags.key": [],
+    "flags.value": [],
     "group_id": 124,
     "http_method": null,
     "http_referer": null,

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@events-ErrorsProcessor-events__1__null-tag-keys.json.snap

@@ -25,6 +25,8 @@ expression: snapshot_payload
     "exception_stacks.mechanism_type": [],
     "exception_stacks.type": [],
     "exception_stacks.value": [],
+    "flags.key": [],
+    "flags.value": [],
     "group_id": 123123123,
     "http_method": null,
     "http_referer": null,

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@events-ErrorsProcessor-events__1__null-values.json.snap

@@ -25,6 +25,8 @@ expression: snapshot_payload
     "exception_stacks.mechanism_type": [],
     "exception_stacks.type": [],
     "exception_stacks.value": [],
+    "flags.key": [],
+    "flags.value": [],
     "group_id": 124,
     "http_method": null,
     "http_referer": null,

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@events-ErrorsProcessor-events__1__sdk-info-java.json.snap

@@ -25,6 +25,8 @@ expression: snapshot_payload
     "exception_stacks.mechanism_type": [],
     "exception_stacks.type": [],
     "exception_stacks.value": [],
+    "flags.key": [],
+    "flags.value": [],
     "group_id": 123123,
     "http_method": null,
     "http_referer": null,

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@events-ErrorsProcessor-events__1__weird-transaction-source.json.snap

@@ -25,6 +25,8 @@ expression: snapshot_payload
     "exception_stacks.mechanism_type": [],
     "exception_stacks.type": [],
     "exception_stacks.value": [],
+    "flags.key": [],
+    "flags.value": [],
     "group_id": 123,
     "http_method": null,
     "http_referer": null,

tests/datasets/test_errors_processor.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import uuid
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
@@ -35,6 +36,7 @@ class ErrorEvent:
     trace_sampled: bool | None
     environment: str
     replay_id: uuid.UUID | None
+    flags: list[Mapping[str, Any]]
     received_timestamp: datetime
     errors: Sequence[Mapping[str, Any]] | None
 
@@ -135,6 +137,7 @@ def serialize(self) -> tuple[int, str, Mapping[str, Any]]:
                     ]
                 },
                 "contexts": {
+                    "flags": {"values": self.flags},
                     "runtime": {
                         "version": "3.7.6",
                         "type": "runtime",
@@ -377,8 +380,15 @@ def build_result(self, meta: KafkaMessageMetadata) -> Mapping[str, Any]:
             "modules.version": ["1.13.2", "0.2.0", "0.6.0"],
             "transaction_name": "",
             "num_processing_errors": len(self.errors) if self.errors is not None else 0,
+            "flags.key": [],
+            "flags.value": [],
         }
 
+        if self.flags:
+            for flag in self.flags:
+                expected_result["flags.key"].append(flag["flag"])
+                expected_result["flags.value"].append(json.dumps(flag["result"]))
+
         if self.replay_id:
             expected_result["replay_id"] = str(self.replay_id)
 
@@ -423,6 +433,7 @@ def __get_error_event(self, timestamp: datetime, recieved: datetime) -> ErrorEve
                 "subdivision": "fake_subdivision",
             },
             errors=None,
+            flags=[],
         )
 
     def test_errors_basic(self) -> None:
@@ -465,6 +476,7 @@ def test_errors_replayid_context(self) -> None:
             },
             replay_id=uuid.uuid4(),
             errors=None,
+            flags=[],
         )
 
         payload = message.serialize()
@@ -503,6 +515,7 @@ def test_errors_replayid_tag(self) -> None:
             },
             replay_id=None,
             errors=None,
+            flags=[],
         )
         replay_id = uuid.uuid4()
         payload = message.serialize()
@@ -548,6 +561,7 @@ def test_errors_replayid_tag_and_context(self) -> None:
             },
             replay_id=replay_id,
             errors=None,
+            flags=[],
         )
 
         payload = message.serialize()
@@ -591,6 +605,7 @@ def test_errors_replayid_invalid_tag(self) -> None:
             },
             replay_id=None,
             errors=None,
+            flags=[],
         )
         invalid_replay_id = "imnotavaliduuid"
         payload = message.serialize()
@@ -649,6 +664,7 @@ def test_exception_main_thread_true(self) -> None:
                 ]
             },
             errors=None,
+            flags=[],
         )
         payload = message.serialize()
         meta = KafkaMessageMetadata(offset=2, partition=2, timestamp=timestamp)
@@ -703,6 +719,7 @@ def test_exception_main_thread_false(self) -> None:
                 ]
             },
             errors=None,
+            flags=[],
         )
         payload = message.serialize()
         meta = KafkaMessageMetadata(offset=2, partition=2, timestamp=timestamp)
@@ -743,6 +760,7 @@ def test_trace_sampled(self) -> None:
             replay_id=None,
             threads=None,
             errors=None,
+            flags=[],
         )
         payload = message.serialize()
         meta = KafkaMessageMetadata(offset=2, partition=2, timestamp=timestamp)
@@ -794,6 +812,59 @@ def test_errors_processed(self) -> None:
             replay_id=None,
             threads=None,
             errors=[{"type": "one"}, {"type": "two"}, {"type": "three"}],
+            flags=[],
+        )
+        payload = message.serialize()
+        meta = KafkaMessageMetadata(offset=2, partition=2, timestamp=timestamp)
+
+        result = message.build_result(meta)
+        result["num_processing_errors"] = 3
+
+        assert self.processor.process_message(payload, meta) == InsertBatch(
+            [result], ANY
+        )
+
+        # ensure old behavior where data.errors=None won't set 'num_processing_errors'
+        message.errors = None
+        payload = message.serialize()
+        meta = KafkaMessageMetadata(offset=2, partition=2, timestamp=timestamp)
+
+        result = message.build_result(meta)
+
+        assert self.processor.process_message(payload, meta) == InsertBatch(
+            [result], ANY
+        )
+
+    def test_errors_with_flags(self) -> None:
+        timestamp, recieved = self.__get_timestamps()
+        message = ErrorEvent(
+            event_id=str(uuid.UUID("dcb9d002cac548c795d1c9adbfc68040")),
+            organization_id=1,
+            project_id=2,
+            group_id=100,
+            platform="python",
+            message="",
+            trace_id=str(uuid.uuid4()),
+            trace_sampled=False,
+            timestamp=timestamp,
+            received_timestamp=recieved,
+            release="1.0.0",
+            dist="dist",
+            environment="prod",
+            email="[email protected]",
+            ip_address="127.0.0.1",
+            user_id="myself",
+            username="me",
+            geo={
+                "country_code": "XY",
+                "region": "fake_region",
+                "city": "fake_city",
+                "subdivision": "fake_subdivision",
+            },
+            replay_id=None,
+            threads=None,
+            errors=[{"type": "one"}, {"type": "two"}, {"type": "three"}],
+            flags=[{"flag": "abc", "result": True}],
         )
         payload = message.serialize()
         meta = KafkaMessageMetadata(offset=2, partition=2, timestamp=timestamp)
@@ -815,3 +886,73 @@ def test_errors_processed(self) -> None:
         assert self.processor.process_message(payload, meta) == InsertBatch(
             [result], ANY
         )
+
+    def test_errors_with_malformed_flags(self) -> None:
+        timestamp, recieved = self.__get_timestamps()
+        message = ErrorEvent(
+            event_id=str(uuid.UUID("dcb9d002cac548c795d1c9adbfc68040")),
+            organization_id=1,
+            project_id=2,
+            group_id=100,
+            platform="python",
+            message="",
+            trace_id=str(uuid.uuid4()),
+            trace_sampled=False,
+            timestamp=timestamp,
+            received_timestamp=recieved,
+            release="1.0.0",
+            dist="dist",
+            environment="prod",
+            email="[email protected]",
+            ip_address="127.0.0.1",
+            user_id="myself",
+            username="me",
+            geo={
+                "country_code": "XY",
+                "region": "fake_region",
+                "city": "fake_city",
+                "subdivision": "fake_subdivision",
+            },
+            replay_id=None,
+            threads=None,
+            errors=[{"type": "one"}, {"type": "two"}, {"type": "three"}],
+            flags=[],
+        )
+        payload = message.serialize()
+        meta = KafkaMessageMetadata(offset=2, partition=2, timestamp=timestamp)
+
+        # Assert malformed context type is ignored.
+        payload[2]["data"]["contexts"]["flags"] = {"key": "value"}
+        result = self.processor.process_message(payload, meta)
+        assert result.rows[0]["flags.key"] == []
+        assert result.rows[0]["flags.value"] == []
+
+        # Assert malformed values type is ignored.
+        payload[2]["data"]["contexts"]["flags"] = {"values": None}
+        result = self.processor.process_message(payload, meta)
+        assert result.rows[0]["flags.key"] == []
+        assert result.rows[0]["flags.value"] == []
+
+        # Assert malformed item type is ignored.
+        payload[2]["data"]["contexts"]["flags"] = {"values": [None]}
+        result = self.processor.process_message(payload, meta)
+        assert result.rows[0]["flags.key"] == []
+        assert result.rows[0]["flags.value"] == []
+
+        # Assert incorrect item contents is ignored.
+        payload[2]["data"]["contexts"]["flags"] = {"values": [{"key": "value"}]}
+        result = self.processor.process_message(payload, meta)
+        assert result.rows[0]["flags.key"] == []
+        assert result.rows[0]["flags.value"] == []
+
+        # Assert missing "result" key means the whole item is ignored.
+        payload[2]["data"]["contexts"]["flags"] = {"values": [{"flag": "value"}]}
+        result = self.processor.process_message(payload, meta)
+        assert result.rows[0]["flags.key"] == []
+        assert result.rows[0]["flags.value"] == []
+
+        # Assert missing "flag" key means the whole item is ignored.
+        payload[2]["data"]["contexts"]["flags"] = {"values": [{"result": "value"}]}
+        result = self.processor.process_message(payload, meta)
+        assert result.rows[0]["flags.key"] == []
+        assert result.rows[0]["flags.value"] == []