Revert "Add processor for new spans schema (#6123)"

This reverts commit 794fec5.

Co-authored-by: colin-sentry <161344340+colin-sentry@users.noreply.github.com>

rust_snuba/src/processors/mod.rs

@@ -1,4 +1,3 @@
-mod eap_spans;
 mod errors;
 mod functions;
 mod generic_metrics;
@@ -56,7 +55,6 @@ 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/spans.rs

@@ -32,33 +32,33 @@ pub fn process_message(
 }
 
 #[derive(Debug, Default, Deserialize, JsonSchema)]
-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,
+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,
     #[serde(alias = "start_timestamp_micro")]
-    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,
+    start_timestamp_precise: Option<f64>,
+    start_timestamp_ms: u64,
+    tags: Option<BTreeMap<String, String>>,
+    trace_id: Uuid,
 }
 
 #[derive(Debug, Default, Deserialize, JsonSchema)]
-pub(crate) struct FromMeasurementValue {
-    pub(crate) value: f64,
+struct FromMeasurementValue {
+    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 * 1e6) as u64,
+            end_timestamp_precise: (from.end_timestamp_precise.unwrap_or_default() * 1e6) as u64,
             exclusive_time: from.exclusive_time_ms,
             group,
             is_segment: if from.is_segment { 1 } else { 0 },
@@ -344,7 +344,6 @@ 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>,
@@ -367,7 +366,6 @@ 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,17 +107,6 @@ 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:

tests/datasets/test_spans_payloads.py

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