Add processor for new spans schema

rust_snuba/src/processors/mod.rs

@@ -1,3 +1,4 @@
+mod eap_spans;
 mod errors;
 mod functions;
 mod generic_metrics;
@@ -55,6 +56,7 @@ define_processing_functions! {
     ("QuerylogProcessor", "snuba-queries", ProcessingFunctionType::ProcessingFunction(querylog::process_message)),
     ("ReplaysProcessor", "ingest-replay-events", ProcessingFunctionType::ProcessingFunction(replays::process_message)),
     ("SpansMessageProcessor", "snuba-spans", ProcessingFunctionType::ProcessingFunction(spans::process_message)),
+    ("EAPSpansMessageProcessor", "snuba-spans", ProcessingFunctionType::ProcessingFunction(eap_spans::process_message)),
     ("MetricsSummariesMessageProcessor", "snuba-metrics-summaries", ProcessingFunctionType::ProcessingFunction(metrics_summaries::process_message)),
     ("OutcomesProcessor", "outcomes", ProcessingFunctionType::ProcessingFunction(outcomes::process_message)),
     ("GenericCountersMetricsProcessor", "snuba-generic-metrics", ProcessingFunctionType::ProcessingFunction(generic_metrics::process_counter_message)),

rust_snuba/src/processors/snapshots/rust_snuba__processors__eap_spans__tests__serialization.snap

@@ -0,0 +1,90 @@
+---
+source: src/processors/eap_spans.rs
+expression: span
+---
+{
+  "organization_id": 1,
+  "project_id": 1,
+  "trace_id": "d099bf9a-d5a1-43cf-8f83-a98081d0ed3b",
+  "span_id": 9832388815107059821,
+  "parent_span_id": 0,
+  "segment_id": 9832388815107059821,
+  "segment_name": "/api/0/relays/projectconfigs/",
+  "is_segment": true,
+  "_sort_timestamp": 1721319572,
+  "start_timestamp": 1721319572616,
+  "end_timestamp": 1721319572768,
+  "duration_ms": 152,
+  "exclusive_time_ms": 0.228,
+  "retention_days": 90,
+  "name": "/api/0/relays/projectconfigs/",
+  "sampling_factor": 1.0,
+  "sampling_weight": 1.0,
+  "sign": 1,
+  "attr_str_3": {
+    "server_name": "D23CXQ4GK2.local"
+  },
+  "attr_str_7": {
+    "relay_use_post_or_schedule_rejected": "version"
+  },
+  "attr_str_10": {
+    "transaction": "/api/0/relays/projectconfigs/"
+  },
+  "attr_str_11": {
+    "thread.id": "8522009600"
+  },
+  "attr_str_12": {
+    "relay_use_post_or_schedule": "True"
+  },
+  "attr_str_17": {
+    "release": "backend@24.7.0.dev0+c45b49caed1e5fcbf70097ab3f434b487c359b6b"
+  },
+  "attr_str_19": {
+    "environment": "development",
+    "platform": "python"
+  },
+  "attr_str_21": {
+    "relay_endpoint_version": "3"
+  },
+  "attr_str_23": {
+    "status_code": "200"
+  },
+  "attr_str_24": {
+    "http.status_code": "200"
+  },
+  "attr_str_25": {
+    "user": "ip:127.0.0.1"
+  },
+  "attr_str_30": {
+    "sdk.version": "2.7.0",
+    "trace.status": "ok"
+  },
+  "attr_str_31": {
+    "category": "http",
+    "thread.name": "uWSGIWorker1Core0"
+  },
+  "attr_str_33": {
+    "transaction.method": "POST"
+  },
+  "attr_str_34": {
+    "relay_id": "88888888-4444-4444-8444-cccccccccccc",
+    "spans_over_limit": "False",
+    "status": "ok"
+  },
+  "attr_str_38": {
+    "relay_no_cache": "False",
+    "sdk.name": "sentry.python.django"
+  },
+  "attr_str_40": {
+    "op": "http.server"
+  },
+  "attr_str_48": {
+    "transaction.op": "http.server"
+  },
+  "attr_str_49": {
+    "relay_protocol_version": "3"
+  },
+  "attr_num_17": {
+    "num_of_spans": 50.0
+  }
+}

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@snuba-spans-EAPSpansMessageProcessor-snuba-spans__1__basic_span.json.snap

@@ -0,0 +1,73 @@
+---
+source: src/processors/mod.rs
+description: "{\n  \"event_id\": \"dcc403b73ef548648188bbfa6012e9dc\",\n  \"organization_id\": 69,\n  \"project_id\": 1,\n  \"trace_id\": \"deadbeefdeadbeefdeadbeefdeadbeef\",\n  \"span_id\": \"deadbeefdeadbeef\",\n  \"parent_span_id\": \"deadbeefdeadbeef\",\n  \"segment_id\": \"deadbeefdeadbeef\",\n  \"duration_ms\": 1000,\n  \"exclusive_time_ms\": 1000,\n  \"is_segment\": false,\n  \"profile_id\": \"deadbeefdeadbeefdeadbeefdeadbeef\",\n  \"received\": 1715868485.381,\n  \"retention_days\": 90,\n  \"start_timestamp_ms\": 1715868485371,\n  \"start_timestamp_precise\": 1715868485.370551,\n  \"end_timestamp_precise\": 1715868486.370551,\n  \"tags\": {\n    \"tag1\": \"value1\",\n    \"tag2\": \"123\",\n    \"tag3\": \"True\"\n  },\n  \"sentry_tags\": {\n    \"http.method\": \"GET\",\n    \"action\": \"GET\",\n    \"domain\": \"targetdomain.tld:targetport\",\n    \"module\": \"http\",\n    \"group\": \"deadbeefdeadbeef\",\n    \"status\": \"ok\",\n    \"system\": \"python\",\n    \"status_code\": \"200\",\n    \"transaction\": \"/organizations/:orgId/issues/\",\n    \"transaction.op\": \"navigation\",\n    \"op\": \"http.client\",\n    \"transaction.method\": \"GET\"\n  },\n  \"measurements\": {\n    \"http.response_content_length\": {\n      \"value\": 100.0,\n      \"unit\": \"byte\"\n    }\n  },\n  \"_metrics_summary\": {\n    \"c:sentry.events.outcomes@none\": [\n      {\n        \"count\": 1,\n        \"max\": 1.0,\n        \"min\": 1.0,\n        \"sum\": 1.0,\n        \"tags\": {\n          \"category\": \"error\",\n          \"environment\": \"unknown\",\n          \"event_type\": \"error\",\n          \"outcome\": \"accepted\",\n          \"release\": \"backend@2af74c237fbd61489a1ccc46650f4f85befaf8b8\",\n          \"topic\": \"outcomes-billing\",\n          \"transaction\": \"sentry.tasks.store.save_event\"\n        }\n      }\n    ],\n    \"c:sentry.events.post_save.normalize.errors@none\": [\n      {\n        \"count\": 1,\n        \"max\": 0.0,\n        \"min\": 0.0,\n        \"sum\": 0.0,\n        \"tags\": {\n          \"environment\": \"unknown\",\n          \"event_type\": \"error\",\n          \"from_relay\": \"False\",\n          \"release\": \"backend@2af74c237fbd61489a1ccc46650f4f85befaf8b8\",\n          \"transaction\": \"sentry.tasks.store.save_event\"\n        }\n      }\n    ]\n  }\n}\n"
+expression: snapshot_payload
+---
+[
+  {
+    "_sort_timestamp": 1715868485,
+    "attr_num_12": {
+      "http.response_content_length": 100.0
+    },
+    "attr_str_1": {
+      "tag1": "value1",
+      "tag3": "True"
+    },
+    "attr_str_10": {
+      "transaction": "/organizations/:orgId/issues/"
+    },
+    "attr_str_12": {
+      "system": "python"
+    },
+    "attr_str_16": {
+      "domain": "targetdomain.tld:targetport"
+    },
+    "attr_str_17": {
+      "tag2": "123"
+    },
+    "attr_str_21": {
+      "http.method": "GET"
+    },
+    "attr_str_23": {
+      "status_code": "200"
+    },
+    "attr_str_30": {
+      "group": "deadbeefdeadbeef"
+    },
+    "attr_str_32": {
+      "module": "http"
+    },
+    "attr_str_33": {
+      "transaction.method": "GET"
+    },
+    "attr_str_34": {
+      "status": "ok"
+    },
+    "attr_str_40": {
+      "op": "http.client"
+    },
+    "attr_str_48": {
+      "transaction.op": "navigation"
+    },
+    "attr_str_7": {
+      "action": "GET"
+    },
+    "duration_ms": 1000,
+    "end_timestamp": 1715868486371,
+    "exclusive_time_ms": 1000.0,
+    "is_segment": false,
+    "name": "",
+    "organization_id": 69,
+    "parent_span_id": 16045690984833335023,
+    "project_id": 1,
+    "retention_days": 90,
+    "sampling_factor": 1.0,
+    "sampling_weight": 1.0,
+    "segment_id": 16045690984833335023,
+    "segment_name": "/organizations/:orgId/issues/",
+    "sign": 1,
+    "span_id": 16045690984833335023,
+    "start_timestamp": 1715868485371,
+    "trace_id": "deadbeef-dead-beef-dead-beefdeadbeef"
+  }
+]

rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__snuba-spans-.snap

@@ -0,0 +1,18 @@
+---
+source: src/processors/mod.rs
+expression: diff
+---
+[
+    Change {
+        path: "",
+        change: RequiredAdd {
+            property: "end_timestamp_precise",
+        },
+    },
+    Change {
+        path: "",
+        change: RequiredAdd {
+            property: "organization_id",
+        },
+    },
+]

rust_snuba/src/processors/spans.rs

@@ -32,33 +32,33 @@ pub fn process_message(
 }
 
 #[derive(Debug, Default, Deserialize, JsonSchema)]
-struct FromSpanMessage {
-    description: Option<String>,
-    duration_ms: u32,
-    #[serde(alias = "end_timestamp_micro")]
-    end_timestamp_precise: Option<f64>,
-    event_id: Option<Uuid>,
-    exclusive_time_ms: f64,
-    is_segment: bool,
-    measurements: Option<BTreeMap<String, FromMeasurementValue>>,
-    parent_span_id: Option<String>,
-    profile_id: Option<Uuid>,
-    project_id: u64,
-    received: f64,
-    retention_days: Option<u16>,
-    segment_id: Option<String>,
-    sentry_tags: Option<BTreeMap<String, String>>,
-    span_id: String,
+pub(crate) struct FromSpanMessage {
+    pub(crate) description: Option<String>,
+    pub(crate) duration_ms: u32,
+    pub(crate) end_timestamp_precise: f64,
+    pub(crate) event_id: Option<Uuid>,
+    pub(crate) exclusive_time_ms: f64,
+    pub(crate) is_segment: bool,
+    pub(crate) measurements: Option<BTreeMap<String, FromMeasurementValue>>,
+    pub(crate) parent_span_id: Option<String>,
+    pub(crate) profile_id: Option<Uuid>,
+    pub(crate) organization_id: u64,
+    pub(crate) project_id: u64,
+    pub(crate) received: f64,
+    pub(crate) retention_days: Option<u16>,
+    pub(crate) segment_id: Option<String>,
+    pub(crate) sentry_tags: Option<BTreeMap<String, String>>,
+    pub(crate) span_id: String,
     #[serde(alias = "start_timestamp_micro")]
-    start_timestamp_precise: Option<f64>,
-    start_timestamp_ms: u64,
-    tags: Option<BTreeMap<String, String>>,
-    trace_id: Uuid,
+    pub(crate) start_timestamp_precise: Option<f64>,
+    pub(crate) start_timestamp_ms: u64,
+    pub(crate) tags: Option<BTreeMap<String, String>>,
+    pub(crate) trace_id: Uuid,
 }
 
 #[derive(Debug, Default, Deserialize, JsonSchema)]
-struct FromMeasurementValue {
-    value: f64,
+pub(crate) struct FromMeasurementValue {
+    pub(crate) value: f64,
 }
 
 #[derive(Debug, Default, Serialize)]
@@ -153,7 +153,7 @@ impl TryFrom<FromSpanMessage> for Span {
             duration: from.duration_ms,
             end_ms: (end_timestamp_ms % 1000) as u16,
             end_timestamp: end_timestamp_ms / 1000,
-            end_timestamp_precise: (from.end_timestamp_precise.unwrap_or_default() * 1e6) as u64,
+            end_timestamp_precise: (from.end_timestamp_precise * 1e6) as u64,
             exclusive_time: from.exclusive_time_ms,
             group,
             is_segment: if from.is_segment { 1 } else { 0 },
@@ -344,6 +344,7 @@ mod tests {
         parent_span_id: Option<String>,
         profile_id: Option<Uuid>,
         project_id: Option<u64>,
+        organization_id: Option<u64>,
         received: Option<f64>,
         retention_days: Option<u16>,
         segment_id: Option<String>,
@@ -366,6 +367,7 @@ mod tests {
             parent_span_id: Some("deadbeefdeadbeef".into()),
             profile_id: Some(Uuid::new_v4()),
             project_id: Some(1),
+            organization_id: Some(1),
             retention_days: Some(90),
             received: Some(1691105878.720),
             segment_id: Some("deadbeefdeadbeef".into()),

snuba/cli/devserver.py

@@ -107,6 +107,17 @@ def devserver(*, bootstrap: bool, workers: bool) -> None:
                 *COMMON_RUST_CONSUMER_DEV_OPTIONS,
             ],
         ),
+        (
+            "eap-spans-consumer",
+            [
+                "snuba",
+                "rust-consumer",
+                "--storage=eap_spans",
+                "--consumer-group=eap_spans_group",
+                "--use-rust-processor",
+                *COMMON_RUST_CONSUMER_DEV_OPTIONS,
+            ],
+        ),
     ]
 
     if settings.SEPARATE_SCHEDULER_EXECUTOR_SUBSCRIPTIONS_DEV:

snuba/datasets/processors/spans_v2_processor.py

@@ -0,0 +1,6 @@
+from snuba.datasets.processors.rust_compat_processor import RustCompatProcessor
+
+
+class EAPSpansMessageProcessor(RustCompatProcessor):
+    def __init__(self) -> None:
+        super().__init__("EAPSpansMessageProcessor")

tests/datasets/test_spans_payloads.py

@@ -30,6 +30,7 @@
     "exclusive_time_ms": 1234567890123,
     "is_segment": True,
     "project_id": project_id,
+    "organization_id": 1,
     "received": received,
     "retention_days": 90,
     "segment_id": "1234567890123456",

tests/datasets/test_spans_processor.py

@@ -52,6 +52,7 @@ def serialize(self) -> SpanEvent:
                 "is_segment": False,
                 "parent_span_id": self.parent_span_id,
                 "project_id": self.project_id,
+                "organization_id": 1,
                 "received": self.received,
                 "retention_days": self.retention_days,
                 "segment_id": self.segment_id,

tests/test_spans_api.py

@@ -83,6 +83,7 @@ def generate_fizzbuzz_events(self) -> None:
                         .process_message(
                             {
                                 "project_id": p,
+                                "organization_id": 1,
                                 "event_id": uuid.uuid4().hex,
                                 "deleted": 0,
                                 "is_segment": False,