diff --git a/rust/cymbal/src/core/metric_consts.rs b/rust/cymbal/src/core/metric_consts.rs index d238fdf1c956..d70a7a2c7142 100644 --- a/rust/cymbal/src/core/metric_consts.rs +++ b/rust/cymbal/src/core/metric_consts.rs @@ -90,6 +90,11 @@ pub const ASSIGNMENT_RULES_FOUND: &str = "cymbal_assignment_rules_found"; pub const ASSIGNMENT_RULES_TRIED: &str = "cymbal_assignment_rules_tried"; pub const AUTO_ASSIGNMENTS: &str = "cymbal_auto_assignments"; pub const ASSIGNMENT_RULES_DISABLED: &str = "cymbal_assignment_rules_disabled"; +pub const SEVERITY_RULES_PROCESSING_TIME: &str = "cymbal_severity_rules_processing_time"; +pub const SEVERITY_RULES_FOUND: &str = "cymbal_severity_rules_found"; +pub const SEVERITY_RULES_TRIED: &str = "cymbal_severity_rules_tried"; +pub const SEVERITY_RULES_MATCHED: &str = "cymbal_severity_rules_matched"; +pub const SEVERITY_RULES_DISABLED: &str = "cymbal_severity_rules_disabled"; pub const GROUPING_RULES_DISABLED: &str = "cymbal_grouping_rules_disabled"; pub const GROUPING_RULES_PROCESSING_TIME: &str = "cymbal_grouping_rules_processing_time"; pub const GROUPING_RULES_FOUND: &str = "cymbal_grouping_rules_found"; diff --git a/rust/cymbal/src/modes/processing/config.rs b/rust/cymbal/src/modes/processing/config.rs index 2332cc4c21fe..c7f329d2cf97 100644 --- a/rust/cymbal/src/modes/processing/config.rs +++ b/rust/cymbal/src/modes/processing/config.rs @@ -118,6 +118,12 @@ pub struct ProcessingConfig { // The maximum number of bytecode operations we'll store in the cache, across all rules, across all teams pub max_assignment_rule_cache_size: u64, + #[envconfig(default = "300")] + pub severity_rule_cache_ttl_secs: u64, + + #[envconfig(default = "100000")] + pub max_severity_rule_cache_size: u64, + #[envconfig(default = "300")] pub grouping_rule_cache_ttl_secs: u64, diff --git a/rust/cymbal/src/modes/processing/rules/mod.rs b/rust/cymbal/src/modes/processing/rules/mod.rs index 39eac14bb080..c1eec5db3fbf 100644 --- a/rust/cymbal/src/modes/processing/rules/mod.rs +++ b/rust/cymbal/src/modes/processing/rules/mod.rs @@ -6,5 +6,6 @@ pub mod assignment; pub mod bypass; pub mod grouping; pub mod rate_limit; +pub mod severity; pub mod spike; pub mod suppression; diff --git a/rust/cymbal/src/modes/processing/rules/severity.rs b/rust/cymbal/src/modes/processing/rules/severity.rs new file mode 100644 index 000000000000..5db076835669 --- /dev/null +++ b/rust/cymbal/src/modes/processing/rules/severity.rs @@ -0,0 +1,235 @@ +use std::fmt::Display; + +use chrono::{DateTime, Utc}; +use common_types::TeamId; +use hogvm::{ExecutionContext, Program, StepOutcome, VmError}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sqlx::{FromRow, PgConnection}; +use uuid::Uuid; + +use crate::{ + error::UnhandledError, + issue_resolution::Issue, + metric_consts::{ + SEVERITY_RULES_DISABLED, SEVERITY_RULES_FOUND, SEVERITY_RULES_MATCHED, + SEVERITY_RULES_PROCESSING_TIME, SEVERITY_RULES_TRIED, + }, + teams::TeamManager, + types::ProcessedExceptionProperties, +}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RuleSeverity { + Low, + Medium, + High, + Critical, +} + +impl Display for RuleSeverity { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RuleSeverity::Low => write!(f, "low"), + RuleSeverity::Medium => write!(f, "medium"), + RuleSeverity::High => write!(f, "high"), + RuleSeverity::Critical => write!(f, "critical"), + } + } +} + +#[derive(Debug, Clone, FromRow)] +pub struct SeverityRule { + pub id: Uuid, + pub team_id: TeamId, + pub severity: String, + pub order_key: i32, + pub bytecode: Value, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +impl SeverityRule { + pub async fn load_for_team<'c, E>( + executor: E, + team_id: TeamId, + ) -> Result, sqlx::Error> + where + E: sqlx::Executor<'c, Database = sqlx::Postgres>, + { + sqlx::query_as::<_, SeverityRule>( + r#" + SELECT id, team_id, severity, order_key, bytecode, created_at, updated_at + FROM posthog_errortrackingseverityrule + WHERE team_id = $1 AND disabled_data IS NULL + ORDER BY order_key, created_at, id + "#, + ) + .bind(team_id) + .fetch_all(executor) + .await + } + + pub async fn disable<'c, E>( + &self, + executor: E, + message: String, + issue: Value, + properties: Value, + ) -> Result<(), sqlx::Error> + where + E: sqlx::Executor<'c, Database = sqlx::Postgres>, + { + #[derive(Serialize)] + struct DisabledData { + message: String, + issue: Value, + properties: Value, + } + + let disabled_data = DisabledData { + message, + issue, + properties, + }; + sqlx::query( + r#" + UPDATE posthog_errortrackingseverityrule + SET disabled_data = $1, updated_at = NOW() + WHERE id = $2 + "#, + ) + .bind(serde_json::to_value(disabled_data).expect("Can serialize")) + .bind(self.id) + .execute(executor) + .await?; + + metrics::counter!(SEVERITY_RULES_DISABLED).increment(1); + Ok(()) + } + + fn parsed_severity(&self) -> Result { + match self.severity.as_str() { + "low" => Ok(RuleSeverity::Low), + "medium" => Ok(RuleSeverity::Medium), + "high" => Ok(RuleSeverity::High), + "critical" => Ok(RuleSeverity::Critical), + severity => Err(VmError::Other(format!( + "Invalid severity rule value {severity:?}" + ))), + } + } + + pub fn try_match( + &self, + issue: &Value, + properties: &Value, + ) -> Result, VmError> { + let rule_bytecode = match &self.bytecode { + Value::Array(ops) => ops, + _ => { + return Err(VmError::Other(format!( + "Invalid rule bytecode - expected array, got {:?}", + self.bytecode + ))) + } + }; + + let globals = Value::Object(serde_json::Map::from_iter([ + ("issue".to_string(), issue.clone()), + ("properties".to_string(), properties.clone()), + ])); + let program = Program::new(rule_bytecode.clone())?; + let context = ExecutionContext::with_defaults(program).with_globals(globals); + let mut vm = context.to_vm()?; + + metrics::counter!(SEVERITY_RULES_TRIED).increment(1); + + let mut steps = 0; + while steps < context.max_steps { + match vm.step()? { + StepOutcome::Finished(Value::Bool(true)) => { + return self.parsed_severity().map(Some) + } + StepOutcome::Finished(Value::Bool(false)) => return Ok(None), + StepOutcome::Finished(result) => { + return Err(VmError::Other(format!( + "Severity rule returned {result:?}, expected a boolean value" + ))) + } + StepOutcome::NativeCall(name, args) => { + context.execute_native_function_call(&mut vm, &name, args)? + } + StepOutcome::Continue => {} + } + steps += 1; + } + + Err(VmError::OutOfResource("steps".to_string())) + } +} + +pub async fn try_severity_rules( + connection: &mut PgConnection, + team_manager: &TeamManager, + issue: &Issue, + exception_properties: &ProcessedExceptionProperties, +) -> Result, UnhandledError> { + let timing = common_metrics::timing_guard(SEVERITY_RULES_PROCESSING_TIME, &[]); + let mut rules = team_manager + .get_severity_rules(&mut *connection, issue.team_id) + .await?; + + metrics::counter!(SEVERITY_RULES_FOUND).increment(rules.len() as u64); + if rules.is_empty() { + timing.label("outcome", "no_match").fin(); + return Ok(None); + } + + rules.sort_unstable_by_key(|rule| (rule.order_key, rule.created_at, rule.id)); + + #[derive(Serialize, Deserialize)] + struct IssueJson { + status: String, + name: Option, + description: Option, + } + + let issue_json = serde_json::to_value(IssueJson { + status: issue.status.to_string(), + name: issue.name.clone(), + description: issue.description.clone(), + })?; + let properties_json = serde_json::to_value(exception_properties)?; + + for rule in rules { + match rule.try_match(&issue_json, &properties_json) { + Ok(None) => continue, + Ok(Some(severity)) => { + metrics::counter!(SEVERITY_RULES_MATCHED).increment(1); + timing.label("outcome", "match").fin(); + return Ok(Some(severity)); + } + Err(VmError::OutOfResource(resource)) if resource == "steps" => { + tracing::warn!( + rule_id = %rule.id, + team_id = %rule.team_id, + "severity rule exceeded HogVM step budget for this event, skipping" + ); + } + Err(error) => { + rule.disable( + &mut *connection, + error.to_string(), + issue_json.clone(), + properties_json.clone(), + ) + .await?; + team_manager.severity_rules.invalidate(&issue.team_id); + } + } + } + + timing.label("outcome", "no_match").fin(); + Ok(None) +} diff --git a/rust/cymbal/src/modes/processing/teams.rs b/rust/cymbal/src/modes/processing/teams.rs index ca04986f0271..33d91a0c5e3b 100644 --- a/rust/cymbal/src/modes/processing/teams.rs +++ b/rust/cymbal/src/modes/processing/teams.rs @@ -10,6 +10,7 @@ use crate::{ modes::processing::rules::assignment::AssignmentRule, modes::processing::rules::bypass::BypassRule, modes::processing::rules::grouping::GroupingRule, modes::processing::rules::rate_limit::RateLimitSettings, + modes::processing::rules::severity::SeverityRule, modes::processing::rules::spike::SpikeDetectionConfig, modes::processing::rules::suppression::SuppressionRule, }; @@ -18,6 +19,7 @@ use crate::{ pub struct TeamManager { pub token_cache: Cache>, pub assignment_rules: Cache>, + pub severity_rules: Cache>, pub grouping_rules: Cache>, pub suppression_rules: Cache>, pub bypass_rules: Cache>, @@ -45,6 +47,16 @@ impl TeamManager { }) .build(); + let severity_rules = CacheBuilder::new(config.max_severity_rule_cache_size) + .time_to_live(Duration::from_secs(config.severity_rule_cache_ttl_secs)) + .weigher(|_, rules: &Vec| { + rules + .iter() + .map(|rule| rule.bytecode.as_array().map_or(0, Vec::len) as u32) + .sum() + }) + .build(); + let grouping_rules = CacheBuilder::new(config.max_grouping_rule_cache_size) .time_to_live(Duration::from_secs(config.grouping_rule_cache_ttl_secs)) .weigher(|_, v: &Vec| { @@ -83,6 +95,7 @@ impl TeamManager { Self { token_cache: cache, assignment_rules, + severity_rules, grouping_rules, suppression_rules, bypass_rules, @@ -159,6 +172,26 @@ impl TeamManager { Ok(rules) } + pub async fn get_severity_rules<'c, E>( + &self, + executor: E, + team_id: TeamId, + ) -> Result, UnhandledError> + where + E: sqlx::Executor<'c, Database = sqlx::Postgres>, + { + if let Some(rules) = self.severity_rules.get(&team_id) { + metrics::counter!(ANCILLARY_CACHE, "type" => "severity_rules", "outcome" => "hit") + .increment(1); + return Ok(rules.clone()); + } + metrics::counter!(ANCILLARY_CACHE, "type" => "severity_rules", "outcome" => "miss") + .increment(1); + let rules = SeverityRule::load_for_team(executor, team_id).await?; + self.severity_rules.insert(team_id, rules.clone()); + Ok(rules) + } + pub async fn get_suppression_rules<'c, E>( &self, e: E, diff --git a/rust/cymbal/tests/severity_rules.rs b/rust/cymbal/tests/severity_rules.rs new file mode 100644 index 000000000000..8725e88a3c01 --- /dev/null +++ b/rust/cymbal/tests/severity_rules.rs @@ -0,0 +1,222 @@ +use chrono::Utc; +use cymbal::{ + issue_resolution::{Issue, IssueStatus}, + modes::processing::{ + rules::severity::{try_severity_rules, SeverityRule}, + ProcessingConfig, + }, + teams::TeamManager, + types::ProcessedExceptionProperties, +}; +use serde_json::{json, Value}; +use sqlx::{PgPool, Row}; +use uuid::Uuid; + +fn bytecode(result: bool) -> Value { + json!(["_H", 1, if result { 29 } else { 30 }, 38]) +} + +fn rule(team_id: i32, severity: &str, order_key: i32, bytecode: Value) -> SeverityRule { + SeverityRule { + id: Uuid::now_v7(), + team_id, + severity: severity.to_string(), + order_key, + bytecode, + created_at: Utc::now(), + updated_at: Utc::now(), + } +} + +fn issue(team_id: i32) -> Issue { + Issue { + id: Uuid::now_v7(), + team_id, + status: IssueStatus::Active, + severity: Some("medium".to_string()), + name: Some("TypeError".to_string()), + description: Some("Example".to_string()), + created_at: Utc::now(), + } +} + +fn properties() -> ProcessedExceptionProperties { + serde_json::from_value(json!({ + "$exception_list": [{"type": "TypeError", "value": "Example"}], + "$exception_fingerprint": "fingerprint", + "$exception_fingerprint_record": [{"type": "manual"}], + "$exception_issue_id": Uuid::nil(), + "$exception_handled": false, + "$exception_types": ["TypeError"], + "$exception_values": ["Example"], + "$exception_sources": [], + "$exception_functions": [] + })) + .unwrap() +} + +async fn insert_rule(db: &PgPool, rule: &SeverityRule, disabled: bool) { + sqlx::query( + r#" + INSERT INTO posthog_errortrackingseverityrule + (id, team_id, filters, bytecode, severity, order_key, disabled_data, created_at, updated_at) + VALUES ($1, $2, '{}'::jsonb, $3, $4, $5, $6, $7, $8) + "#, + ) + .bind(rule.id) + .bind(rule.team_id) + .bind(&rule.bytecode) + .bind(&rule.severity) + .bind(rule.order_key) + .bind(disabled.then(|| json!({"message": "disabled"}))) + .bind(rule.created_at) + .bind(rule.updated_at) + .execute(db) + .await + .unwrap(); +} + +#[test] +fn supports_every_issue_severity() { + let issue = json!({"status": "active", "name": "TypeError", "description": "Example"}); + let properties = serde_json::to_value(properties()).unwrap(); + + for severity in ["low", "medium", "high", "critical"] { + let matched = rule(1, severity, 0, bytecode(true)) + .try_match(&issue, &properties) + .unwrap() + .unwrap(); + assert_eq!(matched.to_string(), severity); + } +} + +#[sqlx::test(migrations = "./tests/test_migrations")] +async fn loads_only_enabled_rules_for_the_requested_team(db: PgPool) { + let enabled = rule(1, "high", 0, bytecode(true)); + let disabled = rule(1, "critical", 1, bytecode(true)); + let other_team = rule(2, "low", 0, bytecode(true)); + insert_rule(&db, &enabled, false).await; + insert_rule(&db, &disabled, true).await; + insert_rule(&db, &other_team, false).await; + + let loaded = SeverityRule::load_for_team(&db, 1).await.unwrap(); + + assert_eq!(loaded.len(), 1); + assert_eq!(loaded[0].id, enabled.id); +} + +#[sqlx::test(migrations = "./tests/test_migrations")] +async fn first_matching_rule_wins_and_no_match_preserves_fallback(db: PgPool) { + let config = ProcessingConfig::init_with_defaults().unwrap(); + let manager = TeamManager::new(&config); + let team_id = 1; + manager.severity_rules.insert( + team_id, + vec![ + rule(team_id, "critical", 0, bytecode(false)), + rule(team_id, "medium", 2, bytecode(true)), + rule(team_id, "high", 1, bytecode(true)), + ], + ); + let mut conn = db.acquire().await.unwrap(); + + let matched = try_severity_rules(&mut conn, &manager, &issue(team_id), &properties()) + .await + .unwrap(); + assert_eq!(matched.unwrap().to_string(), "high"); + + manager + .severity_rules + .insert(team_id, vec![rule(team_id, "critical", 0, bytecode(false))]); + let unmatched = try_severity_rules(&mut conn, &manager, &issue(team_id), &properties()) + .await + .unwrap(); + assert!(unmatched.is_none()); +} + +#[sqlx::test(migrations = "./tests/test_migrations")] +async fn exact_order_ties_use_rule_id(db: PgPool) { + let config = ProcessingConfig::init_with_defaults().unwrap(); + let manager = TeamManager::new(&config); + let team_id = 1; + let created_at = Utc::now(); + let mut high_id = rule(team_id, "critical", 0, bytecode(true)); + high_id.id = Uuid::from_u128(u128::MAX); + high_id.created_at = created_at; + let mut low_id = rule(team_id, "low", 0, bytecode(true)); + low_id.id = Uuid::from_u128(1); + low_id.created_at = created_at; + manager + .severity_rules + .insert(team_id, vec![high_id, low_id]); + let mut conn = db.acquire().await.unwrap(); + + let matched = try_severity_rules(&mut conn, &manager, &issue(team_id), &properties()) + .await + .unwrap(); + + assert_eq!(matched.unwrap().to_string(), "low"); +} + +#[sqlx::test(migrations = "./tests/test_migrations")] +async fn invalid_rule_is_disabled_and_later_rule_still_matches(db: PgPool) { + let config = ProcessingConfig::init_with_defaults().unwrap(); + let manager = TeamManager::new(&config); + let team_id = 1; + let invalid = rule(team_id, "low", 0, json!("invalid")); + insert_rule(&db, &invalid, false).await; + manager.severity_rules.insert( + team_id, + vec![ + invalid.clone(), + rule(team_id, "critical", 1, bytecode(true)), + ], + ); + let mut conn = db.acquire().await.unwrap(); + + let matched = try_severity_rules(&mut conn, &manager, &issue(team_id), &properties()) + .await + .unwrap(); + + assert_eq!(matched.unwrap().to_string(), "critical"); + let disabled_data: Option = + sqlx::query("SELECT disabled_data FROM posthog_errortrackingseverityrule WHERE id = $1") + .bind(invalid.id) + .fetch_one(&db) + .await + .unwrap() + .get("disabled_data"); + assert!(disabled_data.is_some()); + assert!(manager.severity_rules.get(&team_id).is_none()); +} + +#[sqlx::test(migrations = "./tests/test_migrations")] +async fn step_exhaustion_skips_without_disabling_rule(db: PgPool) { + let config = ProcessingConfig::init_with_defaults().unwrap(); + let manager = TeamManager::new(&config); + let team_id = 1; + let looping = rule(team_id, "low", 0, json!(["_H", 1, 39, -2])); + insert_rule(&db, &looping, false).await; + manager.severity_rules.insert( + team_id, + vec![ + looping.clone(), + rule(team_id, "critical", 1, bytecode(true)), + ], + ); + let mut conn = db.acquire().await.unwrap(); + + let matched = try_severity_rules(&mut conn, &manager, &issue(team_id), &properties()) + .await + .unwrap(); + + assert_eq!(matched.unwrap().to_string(), "critical"); + let disabled_data: Option = + sqlx::query("SELECT disabled_data FROM posthog_errortrackingseverityrule WHERE id = $1") + .bind(looping.id) + .fetch_one(&db) + .await + .unwrap() + .get("disabled_data"); + assert!(disabled_data.is_none()); +} diff --git a/rust/cymbal/tests/test_migrations/20260813000000_add_severity_rules.sql b/rust/cymbal/tests/test_migrations/20260813000000_add_severity_rules.sql new file mode 100644 index 000000000000..7955963cfd39 --- /dev/null +++ b/rust/cymbal/tests/test_migrations/20260813000000_add_severity_rules.sql @@ -0,0 +1,12 @@ +CREATE TABLE IF NOT EXISTS posthog_errortrackingseverityrule +( + id uuid NOT NULL PRIMARY KEY, + team_id integer NOT NULL, + filters jsonb NOT NULL, + bytecode jsonb NOT NULL, + severity text NOT NULL, + order_key integer NOT NULL, + disabled_data jsonb, + created_at timestamp with time zone NOT NULL, + updated_at timestamp with time zone NOT NULL +);