feat(spans): Emitting outcomes when accepted_outcome_emitted is false (#109305)

Follow up to #109303 now that we
know that the value arrives correctly ([see
notebook](https://app.datadoghq.com/notebook/13946162/segment-outcome-emitted)).
This now adds the logic for writing the outcome to the EAP `TraceItem`.

Author: tobias-wilfert

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

@@ -9,11 +9,14 @@
 from sentry_protos.snuba.v1.trace_item_pb2 import (
     AnyValue,
     ArrayValue,
+    CategoryCount,
     KeyValue,
     KeyValueList,
+    Outcomes,
     TraceItem,
 )
 
+from sentry.constants import DataCategory
 from sentry.spans.consumers.process_segments.types import CompatibleSpan
 from sentry.utils import metrics
 from sentry.utils.eap import hex_to_item_id
@@ -111,6 +114,17 @@ def convert_span_to_item(span: CompatibleSpan) -> TraceItem:
         "spans.consumers.process_segments.outcome_emitted",
         tags={"already_emitted": str(span.get("accepted_outcome_emitted"))},
     )
+    outcomes = None
+    if span.get("accepted_outcome_emitted") is False:
+        outcomes = Outcomes(
+            key_id=int(span.get("key_id") or 0),
+            category_count=[
+                CategoryCount(
+                    data_category=int(DataCategory.SPAN_INDEXED),
+                    quantity=1,
+                ),
+            ],
+        )
 
     return TraceItem(
         organization_id=span["organization_id"],
@@ -125,6 +139,7 @@ def convert_span_to_item(span: CompatibleSpan) -> TraceItem:
         retention_days=span["retention_days"],
         downsampled_retention_days=span.get("downsampled_retention_days", 0),
         received=_timestamp(span["received"]),
+        outcomes=outcomes,
     )
 
 

tests/sentry/spans/consumers/process_segments/test_convert.py

@@ -2,11 +2,20 @@
 from typing import cast
 
 import orjson
+import pytest
 from google.protobuf.timestamp_pb2 import Timestamp
 from sentry_kafka_schemas.schema_types.ingest_spans_v1 import SpanEvent
 from sentry_protos.snuba.v1.request_common_pb2 import TraceItemType
-from sentry_protos.snuba.v1.trace_item_pb2 import AnyValue, ArrayValue, KeyValue, KeyValueList
+from sentry_protos.snuba.v1.trace_item_pb2 import (
+    AnyValue,
+    ArrayValue,
+    CategoryCount,
+    KeyValue,
+    KeyValueList,
+    Outcomes,
+)
 
+from sentry.constants import DataCategory
 from sentry.spans.consumers.process_segments.convert import RENAME_ATTRIBUTES, convert_span_to_item
 from sentry.spans.consumers.process_segments.types import CompatibleSpan
 
@@ -250,3 +259,58 @@ def test_convert_renamed_attribute_meta() -> None:
     assert item.attributes.get("sentry._meta.fields.attributes.sentry.raw_description") == AnyValue(
         string_value=orjson.dumps({"meta": description_meta}).decode()
     )
+
+
+@pytest.mark.parametrize(
+    "key_id, expected_key_id",
+    [(123, 123), ("123", 123), (None, 0)],
+    ids=["int_key_id", "str_key_id", "none_key_id"],
+)
+def test_convert_outcomes_when_not_emitted(key_id, expected_key_id) -> None:
+    message: SpanEvent = copy.deepcopy(SPAN_KAFKA_MESSAGE)
+    message["accepted_outcome_emitted"] = False
+    message["key_id"] = key_id
+
+    item = convert_span_to_item(cast(CompatibleSpan, message))
+
+    assert item.HasField("outcomes")
+    assert item.outcomes == Outcomes(
+        key_id=expected_key_id,
+        category_count=[
+            CategoryCount(
+                data_category=int(DataCategory.SPAN_INDEXED),
+                quantity=1,
+            ),
+        ],
+    )
+
+
+def test_convert_outcomes_when_not_emitted_missing_key_id() -> None:
+    message: SpanEvent = copy.deepcopy(SPAN_KAFKA_MESSAGE)
+    message["accepted_outcome_emitted"] = False
+
+    item = convert_span_to_item(cast(CompatibleSpan, message))
+
+    assert item.HasField("outcomes")
+    assert item.outcomes == Outcomes(
+        key_id=0,
+        category_count=[
+            CategoryCount(
+                data_category=int(DataCategory.SPAN_INDEXED),
+                quantity=1,
+            ),
+        ],
+    )
+
+
+def test_convert_outcomes_when_already_emitted() -> None:
+    message: SpanEvent = copy.deepcopy(SPAN_KAFKA_MESSAGE)
+    message["accepted_outcome_emitted"] = True
+    item = convert_span_to_item(cast(CompatibleSpan, message))
+    assert not item.HasField("outcomes")
+
+
+def test_convert_outcomes_when_field_missing() -> None:
+    message: SpanEvent = copy.deepcopy(SPAN_KAFKA_MESSAGE)
+    item = convert_span_to_item(cast(CompatibleSpan, message))
+    assert not item.HasField("outcomes")