feat(err): parse stack traces (#25393)

Co-authored-by: David Newell <[email protected]>

rust/common/types/src/event.rs

@@ -23,3 +23,59 @@ impl CapturedEvent {
         format!("{}:{}", self.token, self.distinct_id)
     }
 }
+
+#[derive(Debug, Deserialize, Serialize, Clone, Copy)]
+#[serde(rename_all = "snake_case")]
+pub enum PersonMode {
+    Full,
+    Propertyless,
+    ForceUpgrade,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct ClickHouseEvent {
+    pub uuid: Uuid,
+    pub team_id: i32,
+    pub event: String,
+    pub distinct_id: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub properties: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub person_id: Option<String>,
+    // TODO: verify timestamp format
+    pub timestamp: String,
+    // TODO: verify timestamp format
+    pub created_at: String,
+    pub elements_chain: String,
+    // TODO: verify timestamp format
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub person_created_at: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub person_properties: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group0_properties: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group1_properties: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group2_properties: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group3_properties: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group4_properties: Option<String>,
+    // TODO: verify timestamp format
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group0_created_at: Option<String>,
+    // TODO: verify timestamp format
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group1_created_at: Option<String>,
+    // TODO: verify timestamp format
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group2_created_at: Option<String>,
+    // TODO: verify timestamp format
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group3_created_at: Option<String>,
+    // TODO: verify timestamp format
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub group4_created_at: Option<String>,
+    pub person_mode: PersonMode,
+}

rust/common/types/src/lib.rs

@@ -3,6 +3,7 @@ mod team;
 
 // Events
 pub use event::CapturedEvent;
+pub use event::ClickHouseEvent;
 
 // Teams
 pub use team::Team;

rust/cymbal/Cargo.toml

@@ -15,9 +15,11 @@ metrics = { workspace = true }
 common-metrics = { path = "../common/metrics" }
 common-alloc = { path = "../common/alloc" }
 common-kafka = { path = "../common/kafka" }
+common-types = { path = "../common/types" }
 thiserror = { workspace = true }
 sqlx = { workspace = true }
 serde_json = { workspace = true }
+serde = { workspace = true }
 
 [lints]
 workspace = true

rust/cymbal/src/lib.rs

@@ -1,3 +1,5 @@
 pub mod app_context;
 pub mod config;
 pub mod error;
+pub mod symbols;
+pub mod traits;

rust/cymbal/src/main.rs

@@ -3,9 +3,14 @@ use std::{future::ready, sync::Arc};
 use axum::{routing::get, Router};
 use common_kafka::kafka_consumer::RecvErr;
 use common_metrics::{serve, setup_metrics_routes};
-use cymbal::{app_context::AppContext, config::Config, error::Error};
+use common_types::ClickHouseEvent;
+use cymbal::{
+    app_context::AppContext,
+    config::Config,
+    error::Error,
+    symbols::types::{PropertyView, RawFrame},
+};
 use envconfig::Envconfig;
-use serde_json::Value;
 use tokio::task::JoinHandle;
 use tracing::{error, info};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer};
@@ -57,22 +62,54 @@ async fn main() -> Result<(), Error> {
         context.worker_liveness.report_healthy().await;
         // Just grab the event as a serde_json::Value and immediately drop it,
         // we can work out a real type for it later (once we're deployed etc)
-        let (_, offset): (Value, _) = match context.consumer.json_recv().await {
+        let (event, offset): (ClickHouseEvent, _) = match context.consumer.json_recv().await {
             Ok(r) => r,
             Err(RecvErr::Kafka(e)) => {
                 return Err(e.into()); // Just die if we recieve a Kafka error
             }
             Err(err) => {
                 // If we failed to parse the message, or it was empty, just log and continue, our
                 // consumer has already stored the offset for us.
-                metrics::counter!("error_tracking_errors", "cause" => "recv_err").increment(1);
+                metrics::counter!("cymbal_errors", "cause" => "recv_err").increment(1);
                 error!("Error receiving message: {:?}", err);
                 continue;
             }
         };
-        metrics::counter!("error_tracking_events_received").increment(1);
+        metrics::counter!("cymbal_events_received").increment(1);
 
-        // This is where the rest of the processing would go
         offset.store().unwrap();
+
+        if event.event != "$exception" {
+            error!("event of type {}", event.event);
+            continue;
+        }
+
+        let Some(properties) = &event.properties else {
+            metrics::counter!("cymbal_errors", "cause" => "no_properties").increment(1);
+            continue;
+        };
+
+        let properties: PropertyView = match serde_json::from_str(properties) {
+            Ok(r) => r,
+            Err(err) => {
+                metrics::counter!("cymbal_errors", "cause" => "invalid_exception_properties")
+                    .increment(1);
+                error!("Error parsing properties: {:?}", err);
+                continue;
+            }
+        };
+
+        let _stack_trace: Vec<RawFrame> =
+            match serde_json::from_str(&properties.exception_stack_trace_raw) {
+                Ok(r) => r,
+                Err(err) => {
+                    metrics::counter!("cymbal_errors", "cause" => "invalid_stack_trace")
+                        .increment(1);
+                    error!("Error parsing stack trace: {:?}", err);
+                    continue;
+                }
+            };
+
+        metrics::counter!("cymbal_stack_track_processed").increment(1);
     }
 }

rust/cymbal/src/symbols/js.rs

@@ -0,0 +1,25 @@
+use serde::{Deserialize, Serialize};
+
+// A minifed JS stack frame. Just the minimal information needed to lookup some
+// sourcemap for it and produce a "real" stack frame.
+#[derive(Debug, Clone, Deserialize)]
+pub struct RawJSFrame {
+    #[serde(rename = "lineno")]
+    pub line: u32,
+    #[serde(rename = "colno")]
+    pub column: u32,
+    #[serde(rename = "filename")]
+    pub uri: String,
+    pub in_app: bool,
+    #[serde(rename = "function")]
+    pub fn_name: String,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct ProcessedFrame {
+    pub line: u32,
+    pub column: u32,
+    pub uri: String,
+    pub in_app: bool,
+    pub fn_name: String,
+}

rust/cymbal/src/symbols/mod.rs

@@ -0,0 +1,3 @@
+pub mod js;
+pub mod symbolifier;
+pub mod types;

rust/cymbal/src/symbols/symbolifier.rs

@@ -0,0 +1,9 @@
+use super::types::{ProcessedStack, RawStack};
+
+pub struct Symbolifier;
+
+impl Symbolifier {
+    pub fn symbolify(_stack: RawStack) -> ProcessedStack {
+        unimplemented!()
+    }
+}

rust/cymbal/src/symbols/types.rs

@@ -0,0 +1,79 @@
+use std::collections::HashMap;
+
+use serde::Deserialize;
+use serde::Serialize;
+use serde_json::Value;
+
+use super::{js::RawJSFrame, symbolifier::Symbolifier};
+
+#[derive(Debug, Deserialize)]
+#[serde(untagged)]
+pub enum RawFrame {
+    JavaScript(RawJSFrame),
+}
+
+// Stacks don't care what the "type" of the frames they contain are, and even permit
+// frames of different types to be mixed together, because we're going to end up "exploding"
+// them into their frame-set anyway, and dispatching a task per frame in a language-agnostic
+// way. Supporting mixed-type stacks is a side benefit of this - I don't know that we'll ever
+// see them, but we get the flexibility "for free"
+#[derive(Debug, Deserialize)]
+pub struct RawStack {
+    pub frames: Vec<RawFrame>,
+}
+
+pub enum ProcessedFrame {
+    JavaScript(),
+}
+
+pub struct ProcessedStack {
+    pub frames: Vec<ProcessedFrame>,
+}
+
+impl RawStack {
+    pub async fn process(self, _sym: &Symbolifier) -> ProcessedStack {
+        unimplemented!()
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct PropertyView {
+    #[serde(rename = "$exception_type")]
+    pub exception_type: String,
+    #[serde(rename = "$exception_message")]
+    pub exception_message: String,
+    #[serde(rename = "$exception_stack_trace_raw")]
+    pub exception_stack_trace_raw: String,
+    #[serde(rename = "$exception_level")]
+    pub exception_level: String,
+    #[serde(rename = "$exception_source")]
+    pub exception_source: String,
+    #[serde(rename = "$exception_lineno")]
+    pub exception_line: u32,
+    #[serde(rename = "$exception_colno")]
+    pub exception_col: u32,
+    #[serde(flatten)]
+    other: HashMap<String, Value>,
+}
+
+#[cfg(test)]
+mod test {
+    use common_types::ClickHouseEvent;
+
+    use crate::symbols::types::{PropertyView, RawFrame};
+
+    #[test]
+    fn it_symbolifies() {
+        let raw: &'static str = include_str!("../../tests/static/raw_js_stack.json");
+
+        let raw: ClickHouseEvent = serde_json::from_str(raw).unwrap();
+
+        let exception_properties: PropertyView =
+            serde_json::from_str(&raw.properties.unwrap()).unwrap();
+
+        let stack_trace: Vec<RawFrame> =
+            serde_json::from_str(&exception_properties.exception_stack_trace_raw).unwrap();
+
+        println!("{:?}", stack_trace);
+    }
+}

rust/cymbal/src/traits.rs

@@ -0,0 +1,13 @@
+use crate::symbols::types::RawStack;
+
+// An "exception" is anything that can self-identify with a "fingerprint"
+pub trait Exception {
+    fn id(&self) -> String;
+    fn to_raw(self) -> serde_json::Value;
+}
+
+// Some excpetions have a raw stack trace we can process. If they do,
+
+pub trait Stacked: Exception {
+    fn stack(&self) -> RawStack;
+}