feat(cymbal): evaluate error tracking severity rules

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";

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,
 

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;

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<Utc>,
+    pub updated_at: DateTime<Utc>,
+}
+
+impl SeverityRule {
+    pub async fn load_for_team<'c, E>(
+        executor: E,
+        team_id: TeamId,
+    ) -> Result<Vec<Self>, 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<RuleSeverity, VmError> {
+        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<Option<RuleSeverity>, 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<Option<RuleSeverity>, 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<String>,
+        description: Option<String>,
+    }
+
+    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)
+}

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<String, Option<Team>>,
     pub assignment_rules: Cache<TeamId, Vec<AssignmentRule>>,
+    pub severity_rules: Cache<TeamId, Vec<SeverityRule>>,
     pub grouping_rules: Cache<TeamId, Vec<GroupingRule>>,
     pub suppression_rules: Cache<TeamId, Vec<SuppressionRule>>,
     pub bypass_rules: Cache<TeamId, Vec<BypassRule>>,
@@ -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<SeverityRule>| {
+                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<GroupingRule>| {
@@ -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<Vec<SeverityRule>, 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,