diff --git a/rust/.sqlx/query-14332a535d61ab0144d1f4dbb1cf68cd3acdf4dabc1eff87ae7bce49cee9328a.json b/rust/.sqlx/query-14332a535d61ab0144d1f4dbb1cf68cd3acdf4dabc1eff87ae7bce49cee9328a.json new file mode 100644 index 0000000000000..b489060b99660 --- /dev/null +++ b/rust/.sqlx/query-14332a535d61ab0144d1f4dbb1cf68cd3acdf4dabc1eff87ae7bce49cee9328a.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT id, team_id, issue_id, fingerprint, version FROM posthog_errortrackingissuefingerprintv2\n WHERE team_id = $1 AND fingerprint = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "team_id", + "type_info": "Int4" + }, + { + "ordinal": 2, + "name": "issue_id", + "type_info": "Uuid" + }, + { + "ordinal": 3, + "name": "fingerprint", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "version", + "type_info": "Int8" + } + ], + "parameters": { + "Left": ["Int4", "Text"] + }, + "nullable": [false, false, false, false, false] + }, + "hash": "14332a535d61ab0144d1f4dbb1cf68cd3acdf4dabc1eff87ae7bce49cee9328a" +} diff --git a/rust/.sqlx/query-44eb698252059821770eaaf5b8213132c3e7bf297834cd42228b35bf3e424dd7.json b/rust/.sqlx/query-44eb698252059821770eaaf5b8213132c3e7bf297834cd42228b35bf3e424dd7.json new file mode 100644 index 0000000000000..8d5f49a073584 --- /dev/null +++ b/rust/.sqlx/query-44eb698252059821770eaaf5b8213132c3e7bf297834cd42228b35bf3e424dd7.json @@ -0,0 +1,18 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO posthog_errortrackingissue (id, team_id, status, created_at)\n VALUES ($1, $2, $3, NOW())\n ON CONFLICT (id) DO NOTHING\n RETURNING (xmax = 0) AS was_inserted\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "was_inserted", + "type_info": "Bool" + } + ], + "parameters": { + "Left": ["Uuid", "Int4", "Text"] + }, + "nullable": [null] + }, + "hash": "44eb698252059821770eaaf5b8213132c3e7bf297834cd42228b35bf3e424dd7" +} diff --git a/rust/.sqlx/query-ad528f712bdaf75a82293018e3dec3544acd0228776b8535f22438db59002e1f.json b/rust/.sqlx/query-ad528f712bdaf75a82293018e3dec3544acd0228776b8535f22438db59002e1f.json new file mode 100644 index 0000000000000..1f2ff6b906965 --- /dev/null +++ b/rust/.sqlx/query-ad528f712bdaf75a82293018e3dec3544acd0228776b8535f22438db59002e1f.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT id, team_id, status FROM posthog_errortrackingissue\n WHERE team_id = $1 AND id = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "team_id", + "type_info": "Int4" + }, + { + "ordinal": 2, + "name": "status", + "type_info": "Text" + } + ], + "parameters": { + "Left": ["Int4", "Uuid"] + }, + "nullable": [false, false, false] + }, + "hash": "ad528f712bdaf75a82293018e3dec3544acd0228776b8535f22438db59002e1f" +} diff --git a/rust/.sqlx/query-fd6745e4ed7575699286d9828c9f6c959ba804dab9c1e5ba39979d19782582ea.json b/rust/.sqlx/query-fd6745e4ed7575699286d9828c9f6c959ba804dab9c1e5ba39979d19782582ea.json new file mode 100644 index 0000000000000..698578f72873a --- /dev/null +++ b/rust/.sqlx/query-fd6745e4ed7575699286d9828c9f6c959ba804dab9c1e5ba39979d19782582ea.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO posthog_errortrackingissuefingerprintv2 (id, team_id, issue_id, fingerprint, version, created_at)\n VALUES ($1, $2, $3, $4, 0, NOW())\n ON CONFLICT (team_id, fingerprint) DO NOTHING\n RETURNING id, team_id, issue_id, fingerprint, version\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "team_id", + "type_info": "Int4" + }, + { + "ordinal": 2, + "name": "issue_id", + "type_info": "Uuid" + }, + { + "ordinal": 3, + "name": "fingerprint", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "version", + "type_info": "Int8" + } + ], + "parameters": { + "Left": ["Uuid", "Int4", "Uuid", "Text"] + }, + "nullable": [false, false, false, false, false] + }, + "hash": "fd6745e4ed7575699286d9828c9f6c959ba804dab9c1e5ba39979d19782582ea" +} diff --git a/rust/cymbal/src/issue_resolution.rs b/rust/cymbal/src/issue_resolution.rs new file mode 100644 index 0000000000000..3d7ab3ddd6ac1 --- /dev/null +++ b/rust/cymbal/src/issue_resolution.rs @@ -0,0 +1,166 @@ +use sqlx::postgres::any::AnyConnectionBackend; +use uuid::Uuid; + +use crate::error::UnhandledError; + +pub struct IssueFingerprintOverride { + pub id: Uuid, + pub team_id: i32, + pub issue_id: Uuid, + pub fingerprint: String, + pub version: i64, +} + +pub struct Issue { + pub id: Uuid, + pub team_id: i32, + pub status: String, +} + +impl Issue { + pub fn new(team_id: i32) -> Self { + Self { + id: Uuid::new_v4(), + team_id, + status: "active".to_string(), // TODO - we should at some point use an enum here + } + } + + pub async fn load<'c, E>( + executor: E, + team_id: i32, + issue_id: Uuid, + ) -> Result, UnhandledError> + where + E: sqlx::Executor<'c, Database = sqlx::Postgres>, + { + let res = sqlx::query_as!( + Issue, + r#" + SELECT id, team_id, status FROM posthog_errortrackingissue + WHERE team_id = $1 AND id = $2 + "#, + team_id, + issue_id + ) + .fetch_optional(executor) + .await?; + + Ok(res) + } + + pub async fn insert<'c, E>(&self, executor: E) -> Result + where + E: sqlx::Executor<'c, Database = sqlx::Postgres>, + { + let did_insert = sqlx::query_scalar!( + r#" + INSERT INTO posthog_errortrackingissue (id, team_id, status, created_at) + VALUES ($1, $2, $3, NOW()) + ON CONFLICT (id) DO NOTHING + RETURNING (xmax = 0) AS was_inserted + "#, + self.id, + self.team_id, + self.status + ) + .fetch_one(executor) + .await?; + + // TODO - I'm fairly sure the Option here is a bug in sqlx, so the unwrap will + // never be hit, but nonetheless I'm not 100% sure the "no rows" case actually + // means the insert was not done. + Ok(did_insert.unwrap_or(false)) + } +} + +impl IssueFingerprintOverride { + pub async fn load<'c, E>( + executor: E, + team_id: i32, + fingerprint: &str, + ) -> Result, UnhandledError> + where + E: sqlx::Executor<'c, Database = sqlx::Postgres>, + { + let res = sqlx::query_as!( + IssueFingerprintOverride, + r#" + SELECT id, team_id, issue_id, fingerprint, version FROM posthog_errortrackingissuefingerprintv2 + WHERE team_id = $1 AND fingerprint = $2 + "#, + team_id, + fingerprint + ).fetch_optional(executor).await?; + + Ok(res) + } + + pub async fn create_or_load<'c, E>( + executor: E, + team_id: i32, + fingerprint: &str, + issue: &Issue, + ) -> Result + where + E: sqlx::Executor<'c, Database = sqlx::Postgres>, + { + // We do an "ON CONFLICT DO NOTHING" here because callers can compare the returned issue id + // to the passed Issue, to see if the issue was actually inserted or not. + let res = sqlx::query_as!( + IssueFingerprintOverride, + r#" + INSERT INTO posthog_errortrackingissuefingerprintv2 (id, team_id, issue_id, fingerprint, version, created_at) + VALUES ($1, $2, $3, $4, 0, NOW()) + ON CONFLICT (team_id, fingerprint) DO NOTHING + RETURNING id, team_id, issue_id, fingerprint, version + "#, + Uuid::new_v4(), + team_id, + issue.id, + fingerprint + ).fetch_one(executor).await?; + + Ok(res) + } +} + +pub async fn resolve_issue<'c, A>( + con: A, + fingerprint: &str, + team_id: i32, +) -> Result +where + A: sqlx::Acquire<'c, Database = sqlx::Postgres>, +{ + let mut conn = con.acquire().await?; + // If an override already exists, just fast-path, skipping the transaction + if let Some(issue_override) = + IssueFingerprintOverride::load(&mut *conn, team_id, fingerprint).await? + { + return Ok(issue_override); + } + + // Start a transaction, so we can roll it back on override insert failure + conn.begin().await?; + // Insert a new issue + let issue = Issue::new(team_id); + // We don't actually care if we insert the issue here or not - conflicts aren't possible at + // this stage. + issue.insert(&mut *conn).await?; + // Insert the fingerprint override + let issue_override = + IssueFingerprintOverride::create_or_load(&mut *conn, team_id, fingerprint, &issue).await?; + + // If we actually inserted a new row for the issue override, commit the transaction, + // saving both the issue and the override. Otherwise, rollback the transaction, and + // use the retrieved issue override. + let was_created = issue_override.issue_id == issue.id; + if !was_created { + conn.rollback().await?; + } else { + conn.commit().await?; + } + + Ok(issue_override) +} diff --git a/rust/cymbal/src/lib.rs b/rust/cymbal/src/lib.rs index 9ba7b6b8d635b..b88634c1828cd 100644 --- a/rust/cymbal/src/lib.rs +++ b/rust/cymbal/src/lib.rs @@ -4,6 +4,7 @@ use app_context::AppContext; use common_types::ClickHouseEvent; use error::{EventError, UnhandledError}; use fingerprinting::generate_fingerprint; +use issue_resolution::resolve_issue; use tracing::warn; use types::{ErrProps, Exception, Stacktrace}; use uuid::Uuid; @@ -13,6 +14,7 @@ pub mod config; pub mod error; pub mod fingerprinting; pub mod frames; +pub mod issue_resolution; pub mod langs; pub mod metric_consts; pub mod symbol_store; @@ -50,7 +52,12 @@ pub async fn handle_event( results.push(process_exception(context, event.team_id, exception).await?); } - props.fingerprint = Some(generate_fingerprint(&results)); + let fingerprint = generate_fingerprint(&results); + + let issue_override = resolve_issue(&context.pool, &fingerprint, event.team_id).await?; + + props.fingerprint = Some(fingerprint); + props.resolved_issue_id = Some(issue_override.issue_id); props.exception_list = Some(results); event.properties = Some(serde_json::to_string(&props).unwrap()); diff --git a/rust/cymbal/src/types/mod.rs b/rust/cymbal/src/types/mod.rs index c2b67c77aa4cc..3084297df89c1 100644 --- a/rust/cymbal/src/types/mod.rs +++ b/rust/cymbal/src/types/mod.rs @@ -3,6 +3,7 @@ use std::collections::HashMap; use serde::{Deserialize, Serialize}; use serde_json::Value; use sha2::{digest::Update, Sha512}; +use uuid::Uuid; use crate::frames::{Frame, RawFrame}; @@ -53,6 +54,11 @@ pub struct ErrProps { skip_serializing_if = "Option::is_none" )] pub fingerprint: Option, // We expect this not to exist when the event is received, and we populate it as part of processing + #[serde( + rename = "$exception_issue_id", + skip_serializing_if = "Option::is_none" + )] + pub resolved_issue_id: Option, // We populate the exception issue id as part of processing #[serde(flatten)] // A catch-all for all the properties we don't "care" about, so when we send back to kafka we don't lose any info pub other: HashMap, diff --git a/rust/cymbal/tests/test_migrations/20241101134611_test_migration_for_symbol_set_saving_tests.sql b/rust/cymbal/tests/test_migrations/20241101134611_test_migration_for_symbol_set_saving_tests.sql index 2484c6f429866..88029256999b0 100644 --- a/rust/cymbal/tests/test_migrations/20241101134611_test_migration_for_symbol_set_saving_tests.sql +++ b/rust/cymbal/tests/test_migrations/20241101134611_test_migration_for_symbol_set_saving_tests.sql @@ -11,7 +11,6 @@ CREATE TABLE posthog_errortrackingsymbolset ( -- Create index for team_id and ref combination CREATE INDEX idx_error_tracking_symbol_sets_team_ref ON posthog_errortrackingsymbolset(team_id, ref); --- Add migration script here CREATE TABLE IF NOT EXISTS posthog_errortrackingstackframe ( id UUID PRIMARY KEY, raw_id TEXT NOT NULL, @@ -23,3 +22,20 @@ CREATE TABLE IF NOT EXISTS posthog_errortrackingstackframe ( context TEXT, UNIQUE(raw_id, team_id) ); + +CREATE TABLE IF NOT EXISTS posthog_errortrackingissue ( + id UUID PRIMARY KEY, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + status TEXT, + team_id INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS posthog_errortrackingissuefingerprintv2 ( + id UUID PRIMARY KEY, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + fingerprint TEXT NOT NULL, + version BIGINT NOT NULL, + team_id INTEGER NOT NULL, + issue_id UUID NOT NULL, + UNIQUE(team_id, fingerprint) +);