From 9eefded4dfe2e5448e4b2fc3af99c60391ec3384 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 23 Jul 2026 21:14:41 +0000 Subject: [PATCH 1/5] feat(search_issues): rewrite the search issues consumer in Rust Reimplement `SearchIssuesMessageProcessor` as a native Rust processor (`rust_snuba/src/processors/search_issues.rs`) so the search issues / issue-occurrence consumer runs in Rust instead of falling back to the Python processor over the multiprocessing bridge. The new processor mirrors the Python implementation exactly: - Parses the `[version, "insert", event, group_state?]` payload, ignoring the trailing post-process group state. - Promotes environment/release/dist/user tags, supports tags supplied either as an object or as a list of pairs, and sorts them like `_as_dict_safe` + `sorted(...)`. - Preserves payload key order for `contexts.key` / `contexts.value` (Python does not sort contexts) using order-preserving deserialization, since serde_json's `Map` is a `BTreeMap` without `preserve_order`. - Promotes `trace_id` / `profile_id` / `replay_id` from contexts and coerces them to UUIDs, extracts user/ip, sdk, http referer, and transaction duration, and derives `client_timestamp` / `timestamp_ms` from either `data.client_timestamp` or the `datetime` field. The processor is registered in `processors/mod.rs` against the `generic-events` topic, and the Python `SearchIssuesMessageProcessor` becomes a thin `RustCompatProcessor` shim (the `SearchIssueEvent` TypedDicts are retained for other importers). Testing: - 25 Rust unit tests ported from the Python processor tests. - Snapshot tests over the two `generic-events` schema examples. - The Python processor tests now exercise the Rust implementation through the compat shim. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WaxqQJj1RHRonojiRC3puS --- rust_snuba/src/processors/mod.rs | 6 + rust_snuba/src/processors/search_issues.rs | 1147 +++++++++++++++++ ...essor-generic-events__1__generic.json.snap | 65 + ...cessor-generic-events__1__insert.json.snap | 51 + .../processors/search_issues_processor.py | 296 +---- .../datasets/test_search_issues_processor.py | 108 +- 6 files changed, 1306 insertions(+), 367 deletions(-) create mode 100644 rust_snuba/src/processors/search_issues.rs create mode 100644 rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@generic-events-SearchIssuesMessageProcessor-generic-events__1__generic.json.snap create mode 100644 rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@generic-events-SearchIssuesMessageProcessor-generic-events__1__insert.json.snap diff --git a/rust_snuba/src/processors/mod.rs b/rust_snuba/src/processors/mod.rs index 30a9694841a..9deaa046092 100644 --- a/rust_snuba/src/processors/mod.rs +++ b/rust_snuba/src/processors/mod.rs @@ -11,6 +11,7 @@ mod profiles; mod querylog; mod release_health_metrics; mod replays; +mod search_issues; pub mod utils; use crate::config::ProcessorConfig; @@ -64,6 +65,7 @@ define_processing_functions! { ("PolymorphicMetricsProcessor", "snuba-metrics", ProcessingFunctionType::ProcessingFunction(release_health_metrics::process_metrics_message)), ("ErrorsProcessor", "events", ProcessingFunctionType::ProcessingFunctionWithReplacements(errors::process_message_with_replacement)), ("ProfileChunksProcessor", "snuba-profile-chunks", ProcessingFunctionType::ProcessingFunction(profile_chunks::process_message)), + ("SearchIssuesMessageProcessor", "generic-events", ProcessingFunctionType::ProcessingFunction(search_issues::process_message)), ("EAPItemsProcessor", "snuba-items", ProcessingFunctionType::ProcessingFunction(eap_items::process_message)), ("LlmProxyCostProcessor", "snuba-llm-proxy-cost", ProcessingFunctionType::ProcessingFunction(llm_proxy_cost::process_message)), } @@ -158,6 +160,10 @@ mod tests { settings.add_redaction(".*.message_timestamp", ""); } + if *topic_name == "generic-events" { + settings.add_redaction(".*.message_timestamp", ""); + } + if *topic_name == "snuba-items" { settings.add_redaction( ".*.*[\"sentry._internal.ingested_at\"]", diff --git a/rust_snuba/src/processors/search_issues.rs b/rust_snuba/src/processors/search_issues.rs new file mode 100644 index 00000000000..621d04cba29 --- /dev/null +++ b/rust_snuba/src/processors/search_issues.rs @@ -0,0 +1,1147 @@ +use std::collections::BTreeMap; +use std::fmt; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + +use anyhow::{anyhow, bail, Context}; +use chrono::NaiveDateTime; +use serde::de::{self, IgnoredAny, MapAccess, SeqAccess, Visitor}; +use serde::{Deserialize, Deserializer, Serialize}; +use serde_json::Value; +use uuid::Uuid; + +use sentry_arroyo::backends::kafka::types::KafkaPayload; + +use crate::config::{EnvConfig, ProcessorConfig}; +use crate::processors::utils::enforce_retention; +use crate::types::{InsertBatch, KafkaMessageMetadata}; + +/// Format used for the `datetime` / `group_first_seen` fields. Mirrors +/// `snuba.settings.PAYLOAD_DATETIME_FORMAT` ("%Y-%m-%dT%H:%M:%S.%fZ"). chrono's +/// `%.f` consumes the leading dot, so the literal dot is folded into it. +const PAYLOAD_DATETIME_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%.fZ"; + +/// Matches `SearchIssuesMessageProcessor.FINGERPRINTS_HARD_LIMIT_SIZE`. Only the +/// first `LIMIT - 1` fingerprints are kept. +const FINGERPRINTS_HARD_LIMIT_SIZE: usize = 100; + +pub fn process_message( + payload: KafkaPayload, + metadata: KafkaMessageMetadata, + config: &ProcessorConfig, +) -> anyhow::Result { + let payload_bytes = payload.payload().context("Expected payload")?; + let msg: Message = serde_json::from_slice(payload_bytes).with_context(|| { + format!( + "payload start: {}", + String::from_utf8_lossy(&payload_bytes[..payload_bytes.len().min(200)]) + ) + })?; + + if msg.version != 2 { + bail!("Unsupported message version: {}", msg.version); + } + if msg.operation != "insert" { + bail!("Invalid message type: {}", msg.operation); + } + + let row = SearchIssuesRow::parse(msg.event, &metadata, &config.env_config)?; + InsertBatch::from_rows([row], None) +} + +/// The kafka payload is a JSON array `[version, "insert", event, group_state?]`. +/// We only need the first three elements; any trailing element (the post-process +/// group state) is ignored, matching the Python processor which reads +/// `message[0]`, `message[1]` and `message[2]`. +#[derive(Debug)] +struct Message { + version: u8, + operation: String, + event: InsertEvent, +} + +impl<'de> Deserialize<'de> for Message { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + struct MessageVisitor; + + impl<'de> Visitor<'de> for MessageVisitor { + type Value = Message; + + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.write_str("a sequence of [version, operation, event, ...]") + } + + fn visit_seq(self, mut seq: A) -> Result + where + A: SeqAccess<'de>, + { + let version = seq + .next_element()? + .ok_or_else(|| de::Error::invalid_length(0, &self))?; + let operation = seq + .next_element()? + .ok_or_else(|| de::Error::invalid_length(1, &self))?; + let event = seq + .next_element()? + .ok_or_else(|| de::Error::invalid_length(2, &self))?; + // Drain any trailing elements (e.g. the group-state object) so + // serde does not error on the extra items. + while seq.next_element::()?.is_some() {} + Ok(Message { + version, + operation, + event, + }) + } + } + + deserializer.deserialize_seq(MessageVisitor) + } +} + +#[derive(Debug, Deserialize)] +struct InsertEvent { + organization_id: u64, + project_id: u64, + group_id: u64, + #[serde(default)] + group_first_seen: Option, + event_id: String, + primary_hash: String, + platform: String, + message: String, + #[serde(default)] + datetime: Option, + #[serde(default)] + retention_days: Option, + data: EventData, + occurrence_data: OccurrenceData, +} + +#[derive(Debug, Deserialize)] +struct EventData { + received: f64, + #[serde(default)] + client_timestamp: Option, + #[serde(default)] + timestamp: Option, + #[serde(default)] + start_timestamp: Option, + #[serde(default)] + tags: Option, + #[serde(default)] + user: Option, + #[serde(default)] + sdk: Option, + #[serde(default)] + contexts: Option, + #[serde(default)] + request: Option, + #[serde(default)] + environment: Option, + #[serde(default)] + release: Option, + #[serde(default)] + dist: Option, +} + +#[derive(Debug, Deserialize)] +struct OccurrenceData { + id: String, + #[serde(rename = "type")] + type_id: u16, + issue_title: String, + fingerprint: Vec, + detection_time: f64, + #[serde(default)] + subtitle: Option, + #[serde(default)] + culprit: Option, + #[serde(default)] + level: Option, + #[serde(default)] + resource_id: Option, +} + +#[derive(Debug, Default, Deserialize)] +struct UserData { + #[serde(default)] + id: Option, + #[serde(default)] + username: Option, + #[serde(default)] + email: Option, + #[serde(default)] + ip_address: Option, +} + +#[derive(Debug, Default, Deserialize)] +struct Sdk { + #[serde(default)] + name: Option, + #[serde(default)] + version: Option, +} + +#[derive(Debug, Default, Deserialize)] +struct Request { + #[serde(default)] + method: Option, + #[serde(default)] + headers: Option, +} + +/// A structure that can be sent either as a JSON object (`{"k": "v"}`) or as a +/// list of `[key, value]` pairs (`[["k", "v"]]`). Mirrors Python's +/// `_as_dict_safe`. +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum MapOrPairs { + Map(BTreeMap), + Pairs(Vec>>), +} + +impl MapOrPairs { + /// Normalize to a sorted map, deduplicating list pairs with last-write-wins + /// and dropping null entries / null keys, exactly like `_as_dict_safe` + /// followed by `sorted(...)`. + fn into_dict_safe(self) -> BTreeMap { + match self { + MapOrPairs::Map(map) => map, + MapOrPairs::Pairs(pairs) => { + let mut map = BTreeMap::new(); + for pair in pairs.into_iter().flatten() { + if pair.len() < 2 { + continue; + } + if pair[0].is_null() { + continue; + } + if let Some(key) = unicodify(&pair[0]) { + map.insert(key, pair[1].clone()); + } + } + map + } + } + } +} + +/// Contexts map, preserving the insertion order of the outer keys as they +/// appear in the payload (Python does not sort contexts). +#[derive(Debug, Default)] +struct Contexts(Vec<(String, ContextValue)>); + +impl<'de> Deserialize<'de> for Contexts { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + struct ContextsVisitor; + + impl<'de> Visitor<'de> for ContextsVisitor { + type Value = Contexts; + + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.write_str("a contexts map") + } + + fn visit_map(self, mut map: A) -> Result + where + A: MapAccess<'de>, + { + let mut out = Vec::new(); + while let Some((key, value)) = map.next_entry::()? { + out.push((key, value)); + } + Ok(Contexts(out)) + } + } + + deserializer.deserialize_map(ContextsVisitor) + } +} + +/// A single context value. Only object contexts contribute to the output; other +/// JSON types are recorded as `Other` and skipped, matching the Python +/// `isinstance(ctx_obj, dict)` check. +#[derive(Debug)] +enum ContextValue { + Map(Vec<(String, Value)>), + Other, +} + +impl<'de> Deserialize<'de> for ContextValue { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + struct ContextValueVisitor; + + impl<'de> Visitor<'de> for ContextValueVisitor { + type Value = ContextValue; + + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.write_str("any context value") + } + + fn visit_map(self, mut map: A) -> Result + where + A: MapAccess<'de>, + { + let mut out = Vec::new(); + while let Some((key, value)) = map.next_entry::()? { + out.push((key, value)); + } + Ok(ContextValue::Map(out)) + } + + fn visit_seq(self, mut seq: A) -> Result + where + A: SeqAccess<'de>, + { + while seq.next_element::()?.is_some() {} + Ok(ContextValue::Other) + } + + fn visit_str(self, _v: &str) -> Result { + Ok(ContextValue::Other) + } + fn visit_string(self, _v: String) -> Result { + Ok(ContextValue::Other) + } + fn visit_bool(self, _v: bool) -> Result { + Ok(ContextValue::Other) + } + fn visit_i64(self, _v: i64) -> Result { + Ok(ContextValue::Other) + } + fn visit_u64(self, _v: u64) -> Result { + Ok(ContextValue::Other) + } + fn visit_f64(self, _v: f64) -> Result { + Ok(ContextValue::Other) + } + fn visit_none(self) -> Result { + Ok(ContextValue::Other) + } + fn visit_unit(self) -> Result { + Ok(ContextValue::Other) + } + } + + deserializer.deserialize_any(ContextValueVisitor) + } +} + +#[derive(Debug, Default, Serialize)] +struct SearchIssuesRow { + organization_id: u64, + project_id: u64, + group_id: u64, + group_first_seen: Option, + event_id: Uuid, + search_title: String, + primary_hash: Uuid, + fingerprint: Vec, + occurrence_id: Uuid, + occurrence_type_id: u16, + detection_timestamp: u32, + resource_id: Option, + message: String, + subtitle: Option, + culprit: Option, + level: Option, + #[serde(skip_serializing_if = "Option::is_none")] + trace_id: Option, + platform: String, + environment: Option, + release: Option, + dist: Option, + receive_timestamp: u32, + client_timestamp: u32, + #[serde(rename = "tags.key")] + tags_key: Vec, + #[serde(rename = "tags.value")] + tags_value: Vec, + user: Option, + user_id: Option, + user_name: Option, + user_email: Option, + #[serde(skip_serializing_if = "Option::is_none")] + ip_address_v4: Option, + #[serde(skip_serializing_if = "Option::is_none")] + ip_address_v6: Option, + sdk_name: Option, + sdk_version: Option, + #[serde(rename = "contexts.key")] + contexts_key: Vec, + #[serde(rename = "contexts.value")] + contexts_value: Vec, + http_method: Option, + http_referer: Option, + transaction_duration: u32, + #[serde(skip_serializing_if = "Option::is_none")] + profile_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + replay_id: Option, + message_timestamp: u32, + partition: u16, + offset: u64, + retention_days: u16, + timestamp_ms: u64, +} + +impl SearchIssuesRow { + fn parse( + event: InsertEvent, + metadata: &KafkaMessageMetadata, + env_config: &EnvConfig, + ) -> anyhow::Result { + let data = event.data; + let occ = event.occurrence_data; + + let detection_timestamp = seconds_from_timestamp(occ.detection_time); + let receive_timestamp = seconds_from_timestamp(data.received); + let retention_days = enforce_retention(event.retention_days, env_config); + + // client_timestamp (a DateTime column, second precision) and + // timestamp_ms (a DateTime64(3) column, millisecond precision). + let (client_timestamp, timestamp_ms) = match data.client_timestamp.filter(|c| *c != 0.0) { + Some(client_ts) => { + let (secs, micros) = py_utcfromtimestamp(client_ts); + let millis = secs * 1000 + (micros as i64) / 1000; + (clamp_u32(secs), millis.max(0) as u64) + } + None => { + let datetime_str = event + .datetime + .as_deref() + .filter(|s| !s.is_empty()) + .ok_or_else(|| { + anyhow!("message missing data.client_timestamp or datetime field") + })?; + let naive = NaiveDateTime::parse_from_str(datetime_str, PAYLOAD_DATETIME_FORMAT) + .with_context(|| { + format!("datetime field has incompatible datetime format: {datetime_str}") + })?; + let dt = naive.and_utc(); + let secs = dt.timestamp(); + if !(0..=u32::MAX as i64).contains(&secs) { + bail!("datetime field out of valid range: {datetime_str}"); + } + (secs as u32, dt.timestamp_millis().max(0) as u64) + } + }; + + let group_first_seen = match event.group_first_seen { + Some(raw) => { + let naive = NaiveDateTime::parse_from_str(&raw, PAYLOAD_DATETIME_FORMAT) + .with_context(|| format!("group_first_seen has incompatible format: {raw}"))?; + let secs = naive.and_utc().timestamp(); + if (0..=u32::MAX as i64).contains(&secs) { + Some(secs as u32) + } else { + None + } + } + None => None, + }; + + let mut fingerprint = occ.fingerprint; + fingerprint.truncate(FINGERPRINTS_HARD_LIMIT_SIZE - 1); + + // --- Tags (sorted) + promoted tags --- + let tags_map = data + .tags + .map(MapOrPairs::into_dict_safe) + .unwrap_or_default(); + let mut tags_key = Vec::with_capacity(tags_map.len()); + let mut tags_value = Vec::with_capacity(tags_map.len()); + for (key, value) in &tags_map { + if let Some(unicodified) = unicodify(value) { + if !unicodified.is_empty() { + tags_key.push(key.clone()); + tags_value.push(unicodified); + } + } + } + + let environment = if tags_map.contains_key("environment") { + unicodify(&tags_map["environment"]) + } else { + data.environment.as_ref().and_then(unicodify) + }; + let release = if tags_map.contains_key("sentry:release") { + unicodify(&tags_map["sentry:release"]) + } else { + data.release.as_ref().and_then(unicodify) + }; + let user = tags_map.get("sentry:user").and_then(unicodify); + let dist = if tags_map.contains_key("sentry:dist") { + unicodify(&tags_map["sentry:dist"]) + } else { + data.dist.as_ref().and_then(unicodify) + }; + + // --- User --- + let user_data = data.user.unwrap_or_default(); + let user_id = user_data.id.as_ref().and_then(unicodify); + let user_name = user_data.username.as_ref().and_then(unicodify); + let user_email = user_data.email.as_ref().and_then(unicodify); + let (ip_address_v4, ip_address_v6) = match user_data + .ip_address + .as_ref() + .and_then(unicodify) + .and_then(|s| s.parse::().ok()) + { + Some(IpAddr::V4(v4)) => (Some(v4), None), + Some(IpAddr::V6(v6)) => (None, Some(v6)), + None => (None, None), + }; + + // --- SDK --- + let sdk = data.sdk.unwrap_or_default(); + let sdk_name = sdk.name.as_ref().and_then(unicodify); + let sdk_version = sdk.version.as_ref().and_then(unicodify); + + // --- Request / HTTP --- + let request = data.request.unwrap_or_default(); + let http_method = request.method.as_ref().and_then(unicodify); + let headers_map = request + .headers + .map(MapOrPairs::into_dict_safe) + .unwrap_or_default(); + let http_referer = headers_map.get("Referer").and_then(unicodify); + + // --- Contexts (ordered) + promoted trace/profile/replay ids --- + let contexts = data.contexts.unwrap_or_default(); + let mut contexts_key = Vec::new(); + let mut contexts_value = Vec::new(); + for (name, value) in &contexts.0 { + if let ContextValue::Map(inner) = value { + for (inner_key, inner_value) in inner { + if inner_key == "type" { + continue; + } + if let Some(stringified) = context_scalar_to_string(inner_value) { + contexts_key.push(format!("{name}.{inner_key}")); + contexts_value.push(stringified); + } + } + } + } + + let trace_id = promote_uuid_context(&contexts, "trace", "trace_id")?; + let profile_id = promote_uuid_context(&contexts, "profile", "profile_id")?; + let replay_id = promote_uuid_context(&contexts, "replay", "replay_id")?; + + // --- Transaction duration --- + let transaction_duration = match ( + value_as_number(&data.start_timestamp), + value_as_number(&data.timestamp), + ) { + (Some(start), Some(finish)) => { + let start_secs = extract_valid_timestamp(start); + let finish_secs = extract_valid_timestamp(finish); + ((finish_secs - start_secs) * 1000).max(0) as u32 + } + _ => 0, + }; + + Ok(SearchIssuesRow { + organization_id: event.organization_id, + project_id: event.project_id, + group_id: event.group_id, + group_first_seen, + event_id: ensure_uuid(&event.event_id)?, + search_title: occ.issue_title, + primary_hash: ensure_uuid(&event.primary_hash)?, + fingerprint, + occurrence_id: ensure_uuid(&occ.id)?, + occurrence_type_id: occ.type_id, + detection_timestamp, + resource_id: occ.resource_id, + message: event.message, + subtitle: occ.subtitle, + culprit: occ.culprit, + level: occ.level, + trace_id, + platform: event.platform, + environment, + release, + dist, + receive_timestamp, + client_timestamp, + tags_key, + tags_value, + user, + user_id, + user_name, + user_email, + ip_address_v4, + ip_address_v6, + sdk_name, + sdk_version, + contexts_key, + contexts_value, + http_method, + http_referer, + transaction_duration, + profile_id, + replay_id, + message_timestamp: clamp_u32(metadata.timestamp.timestamp()), + partition: metadata.partition, + offset: metadata.offset, + retention_days, + timestamp_ms, + }) + } +} + +/// Look up `contexts[context_name][key]`, and if present and non-null, coerce it +/// to a UUID (raising an error on an invalid UUID, like Python's +/// `ensure_uuid`). +fn promote_uuid_context( + contexts: &Contexts, + context_name: &str, + key: &str, +) -> anyhow::Result> { + for (name, value) in &contexts.0 { + if name != context_name { + continue; + } + if let ContextValue::Map(inner) = value { + for (inner_key, inner_value) in inner { + if inner_key == key && !inner_value.is_null() { + if let Some(stringified) = unicodify(inner_value) { + return Ok(Some(ensure_uuid(&stringified)?)); + } + } + } + } + } + Ok(None) +} + +/// Equivalent to Python's `str(uuid.UUID(value))`: parse the string as a UUID +/// (accepting hyphenated or unhyphenated forms) and error otherwise. +fn ensure_uuid(value: &str) -> anyhow::Result { + Uuid::parse_str(value).map_err(|_| anyhow!("invalid UUID: {value}")) +} + +/// Equivalent to Python's `_unicodify`: `None` for null, JSON-encoded string for +/// arrays/objects, and the stringified scalar otherwise. Booleans use Python's +/// `str(bool)` capitalization ("True"/"False"). +fn unicodify(value: &Value) -> Option { + match value { + Value::Null => None, + Value::Bool(b) => Some(if *b { "True" } else { "False" }.to_owned()), + Value::Number(n) => Some(n.to_string()), + Value::String(s) => Some(s.clone()), + Value::Array(_) | Value::Object(_) => serde_json::to_string(value).ok(), + } +} + +/// Coerce a context inner value to a string only when it is a scalar type +/// (str/number/bool), matching Python's `valid_types = (int, float, str)` check +/// (bool is a subclass of int in Python) plus the truthiness filter that drops +/// empty strings. +fn context_scalar_to_string(value: &Value) -> Option { + match value { + Value::String(s) => { + if s.is_empty() { + None + } else { + Some(s.clone()) + } + } + Value::Number(n) => Some(n.to_string()), + Value::Bool(b) => Some(if *b { "True" } else { "False" }.to_owned()), + _ => None, + } +} + +/// Returns the float value only when the JSON value is a number, matching +/// Python's `isinstance(x, numbers.Number)` gate for transaction duration. +fn value_as_number(value: &Option) -> Option { + match value { + Some(Value::Number(n)) => n.as_f64(), + _ => None, + } +} + +/// Truncate a float timestamp toward zero (Python `int(...)`) and validate it is +/// within the uint32 range, falling back to "now" for out-of-range values (as +/// `_ensure_valid_date` does). +fn extract_valid_timestamp(value: f64) -> i64 { + let secs = value.trunc() as i64; + if (0..=u32::MAX as i64).contains(&secs) { + secs + } else { + chrono::Utc::now().timestamp() + } +} + +/// Truncate a float timestamp to whole seconds and clamp into the uint32 range, +/// matching `datetime.utcfromtimestamp(...)` stored into a second-precision +/// DateTime column. +fn seconds_from_timestamp(value: f64) -> u32 { + clamp_u32(value.trunc() as i64) +} + +fn clamp_u32(value: i64) -> u32 { + value.clamp(0, u32::MAX as i64) as u32 +} + +/// Replicates `datetime.utcfromtimestamp` rounding to microseconds, returning +/// `(whole_seconds, microseconds)`. +fn py_utcfromtimestamp(value: f64) -> (i64, u32) { + let whole = value.trunc(); + let frac = value - whole; + let mut secs = whole as i64; + let mut micros = (frac * 1_000_000.0).round() as i64; + if micros >= 1_000_000 { + secs += 1; + micros -= 1_000_000; + } else if micros < 0 { + secs -= 1; + micros += 1_000_000; + } + (secs, micros as u32) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + use chrono::DateTime; + use serde_json::json; + + fn kafka_meta() -> KafkaMessageMetadata { + KafkaMessageMetadata { + partition: 0, + offset: 1, + timestamp: DateTime::from_timestamp(1_234_567_890, 0).unwrap(), + } + } + + fn base_event() -> Value { + json!({ + "project_id": 1, + "organization_id": 2, + "group_id": 3, + "event_id": "7f0e2b1a4c5d4e6f8a9b0c1d2e3f4a5b", + "retention_days": 90, + "primary_hash": "a1b2c3d4e5f6071829304a5b6c7d8e9f", + "datetime": "2023-06-27T00:00:00.000000Z", + "platform": "other", + "message": "something", + "data": { + "received": 1687800001.0 + }, + "occurrence_data": { + "id": "cccccccccccccccccccccccccccccccc", + "type": 1, + "issue_title": "search me", + "fingerprint": ["one", "two"], + "detection_time": 1687800000.0 + } + }) + } + + fn process(event: Value) -> Vec { + let msg = json!([2, "insert", event]); + let payload = KafkaPayload::new(None, None, Some(serde_json::to_vec(&msg).unwrap())); + let batch = process_message(payload, kafka_meta(), &ProcessorConfig::default()).unwrap(); + let encoded = String::from_utf8(batch.rows.into_encoded_rows()).unwrap(); + encoded + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect() + } + + fn process_one(event: Value) -> Value { + let mut rows = process(event); + assert_eq!(rows.len(), 1); + rows.remove(0) + } + + /// Process a raw JSON event string. Unlike [`process`], this does not round + /// trip through the `json!` macro (whose object keys are sorted because + /// serde_json's `Map` is a `BTreeMap` without the `preserve_order` feature), + /// so it can be used to assert that the processor preserves the key order of + /// the on-the-wire payload. + fn process_one_raw(event_json: &str) -> Value { + let msg = format!("[2, \"insert\", {event_json}]"); + let payload = KafkaPayload::new(None, None, Some(msg.into_bytes())); + let batch = process_message(payload, kafka_meta(), &ProcessorConfig::default()).unwrap(); + let encoded = String::from_utf8(batch.rows.into_encoded_rows()).unwrap(); + let mut rows: Vec = encoded + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + assert_eq!(rows.len(), 1); + rows.remove(0) + } + + #[test] + fn test_basic_required_columns() { + let row = process_one(base_event()); + assert_eq!(row["organization_id"], 2); + assert_eq!(row["project_id"], 1); + assert_eq!(row["group_id"], 3); + assert_eq!(row["search_title"], "search me"); + assert_eq!(row["occurrence_type_id"], 1); + assert_eq!(row["message"], "something"); + assert_eq!(row["platform"], "other"); + assert_eq!(row["fingerprint"], json!(["one", "two"])); + assert_eq!(row["retention_days"], 90); + assert_eq!(row["partition"], 0); + assert_eq!(row["offset"], 1); + // UUIDs are hyphenated. + assert_eq!(row["event_id"], "7f0e2b1a-4c5d-4e6f-8a9b-0c1d2e3f4a5b"); + assert_eq!(row["occurrence_id"], "cccccccc-cccc-cccc-cccc-cccccccccccc"); + assert_eq!(row["detection_timestamp"], 1687800000u32); + assert_eq!(row["receive_timestamp"], 1687800001u32); + assert_eq!(row["message_timestamp"], 1_234_567_890u32); + } + + #[test] + fn test_client_timestamp_and_timestamp_ms_from_datetime() { + let mut event = base_event(); + event["datetime"] = json!("2023-02-27T15:40:12.223000Z"); + let row = process_one(event); + assert_eq!(row["client_timestamp"], 1_677_512_412u32); + assert_eq!(row["timestamp_ms"], 1_677_512_412_223u64); + } + + #[test] + fn test_client_timestamp_from_data() { + let mut event = base_event(); + event["data"]["client_timestamp"] = json!(1_677_512_412.223); + let row = process_one(event); + assert_eq!(row["client_timestamp"], 1_677_512_412u32); + assert_eq!(row["timestamp_ms"], 1_677_512_412_223u64); + } + + #[test] + fn test_missing_client_timestamp_and_datetime_errors() { + let mut event = base_event(); + event.as_object_mut().unwrap().remove("datetime"); + let msg = json!([2, "insert", event]); + let payload = KafkaPayload::new(None, None, Some(serde_json::to_vec(&msg).unwrap())); + let result = process_message(payload, kafka_meta(), &ProcessorConfig::default()); + assert!(result.is_err()); + } + + #[test] + fn test_extract_user() { + let mut event = base_event(); + event["data"]["user"] = json!({ + "id": 1, + "username": "user", + "email": "test@example.com", + "ip_address": "127.0.0.1" + }); + let row = process_one(event); + assert_eq!(row["user_name"], "user"); + assert_eq!(row["user_id"], "1"); + assert_eq!(row["user_email"], "test@example.com"); + assert_eq!(row["ip_address_v4"], "127.0.0.1"); + assert!(row.get("ip_address_v6").is_none()); + } + + #[test] + fn test_extract_user_empty() { + let mut event = base_event(); + event["data"]["user"] = json!({}); + let row = process_one(event); + assert_eq!(row["user_name"], Value::Null); + assert_eq!(row["user_id"], Value::Null); + assert_eq!(row["user_email"], Value::Null); + assert!(row.get("ip_address_v4").is_none()); + } + + #[test] + fn test_extract_ipv6() { + let mut event = base_event(); + event["data"]["user"] = json!({ "ip_address": "2001:db8::1" }); + let row = process_one(event); + assert_eq!(row["ip_address_v6"], "2001:db8::1"); + assert!(row.get("ip_address_v4").is_none()); + } + + #[test] + fn test_promoted_user_from_tag() { + let mut event = base_event(); + event["data"]["tags"] = json!({ "sentry:user": "user123" }); + let row = process_one(event); + assert_eq!(row["user"], "user123"); + } + + #[test] + fn test_extract_environment() { + let mut event = base_event(); + event["data"]["environment"] = json!("prod"); + let row = process_one(event); + assert_eq!(row["environment"], "prod"); + } + + #[test] + fn test_extract_environment_from_tag() { + let mut event = base_event(); + event["data"]["environment"] = json!("prod"); + event["data"]["tags"] = json!({ "environment": "dev" }); + let row = process_one(event); + assert_eq!(row["environment"], "dev"); + } + + #[test] + fn test_extract_release_from_tag() { + let mut event = base_event(); + event["data"]["release"] = json!("release@123"); + event["data"]["tags"] = json!({ "sentry:release": "release@456" }); + let row = process_one(event); + assert_eq!(row["release"], "release@456"); + } + + #[test] + fn test_extract_dist_from_tag() { + let mut event = base_event(); + event["data"]["dist"] = json!("dist@123"); + event["data"]["tags"] = json!({ "sentry:dist": "dist@456" }); + let row = process_one(event); + assert_eq!(row["dist"], "dist@456"); + } + + #[test] + fn test_extract_tags_sorted() { + let mut event = base_event(); + event["data"]["tags"] = json!({ + "key": "value", + "key4": "value4", + "key3": "value3", + "key2": "value2" + }); + let row = process_one(event); + assert_eq!(row["tags.key"], json!(["key", "key2", "key3", "key4"])); + assert_eq!( + row["tags.value"], + json!(["value", "value2", "value3", "value4"]) + ); + } + + #[test] + fn test_extract_tags_from_list() { + let mut event = base_event(); + event["data"]["tags"] = json!([["level", "error"], ["environment", "production"]]); + let row = process_one(event); + assert_eq!(row["tags.key"], json!(["environment", "level"])); + assert_eq!(row["tags.value"], json!(["production", "error"])); + assert_eq!(row["environment"], "production"); + } + + #[test] + fn test_extract_http() { + let mut event = base_event(); + event["data"]["request"] = json!({ + "method": "GET", + "headers": [["Referer", "http://example.com"], ["User-Agent", "test"]], + "extra_stuff": "not_used" + }); + let row = process_one(event); + assert_eq!(row["http_method"], "GET"); + assert_eq!(row["http_referer"], "http://example.com"); + } + + #[test] + fn test_extract_sdk() { + let mut event = base_event(); + event["data"]["sdk"] = json!({ + "version": "1.2.3", + "name": "python", + "packages": [{"version": "0.9.0", "name": "pypi:sentry-sdk"}] + }); + let row = process_one(event); + assert_eq!(row["sdk_name"], "python"); + assert_eq!(row["sdk_version"], "1.2.3"); + } + + #[test] + fn test_extract_context_null_dicts() { + let mut event = base_event(); + event["data"]["contexts"] = json!({ + "trace": null, + "profile": null, + "replay": null, + "scalar": {"string": "scalar_value"} + }); + let row = process_one(event); + assert_eq!(row["contexts.key"], json!(["scalar.string"])); + assert_eq!(row["contexts.value"], json!(["scalar_value"])); + } + + #[test] + fn test_extract_context_filters_non_dict_preserves_order() { + // Built from a raw string so the intended key order survives (see + // `process_one_raw`). + let row = process_one_raw( + r#"{ + "project_id": 1, + "organization_id": 2, + "group_id": 3, + "event_id": "7f0e2b1a4c5d4e6f8a9b0c1d2e3f4a5b", + "retention_days": 90, + "primary_hash": "a1b2c3d4e5f6071829304a5b6c7d8e9f", + "datetime": "2023-06-27T00:00:00.000000Z", + "platform": "other", + "message": "something", + "data": { + "received": 1687800001.0, + "contexts": { + "string": "blah", + "int": 1, + "float": 1.1, + "array": ["a", "b", "c"], + "scalar": { + "string": "scalar_value", + "int": 99, + "float": 123.111 + }, + "nested_dict": { + "array": [1, 2, 3], + "dict": {"key1": "value1"}, + "string": "blah_nested", + "int": 2, + "float": 2.2 + } + } + }, + "occurrence_data": { + "id": "cccccccccccccccccccccccccccccccc", + "type": 1, + "issue_title": "search me", + "fingerprint": ["one", "two"], + "detection_time": 1687800000.0 + } + }"#, + ); + assert_eq!( + row["contexts.key"], + json!([ + "scalar.string", + "scalar.int", + "scalar.float", + "nested_dict.string", + "nested_dict.int", + "nested_dict.float" + ]) + ); + assert_eq!( + row["contexts.value"], + json!(["scalar_value", "99", "123.111", "blah_nested", "2", "2.2"]) + ); + } + + #[test] + fn test_extract_trace_id_from_contexts() { + let mut event = base_event(); + event["data"]["contexts"] = + json!({ "trace": {"trace_id": "1234567890abcdef1234567890abcdef"} }); + let row = process_one(event); + assert_eq!(row["trace_id"], "12345678-90ab-cdef-1234-567890abcdef"); + + for invalid in [ + json!(""), + json!("im a little tea pot"), + json!(1), + json!(1.1), + ] { + let mut event = base_event(); + event["data"]["contexts"] = json!({ "trace": {"trace_id": invalid} }); + let msg = json!([2, "insert", event]); + let payload = KafkaPayload::new(None, None, Some(serde_json::to_vec(&msg).unwrap())); + let result = process_message(payload, kafka_meta(), &ProcessorConfig::default()); + assert!(result.is_err(), "expected error for trace_id {invalid:?}"); + } + } + + #[test] + fn test_extract_profile_and_replay_id() { + let mut event = base_event(); + event["data"]["contexts"] = json!({ + "profile": {"profile_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}, + "replay": {"replay_id": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"} + }); + let row = process_one(event); + assert_eq!(row["profile_id"], "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"); + assert_eq!(row["replay_id"], "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"); + } + + #[test] + fn test_transaction_duration() { + let row = process_one(base_event()); + assert_eq!(row["transaction_duration"], 0); + + let mut event = base_event(); + event["data"]["start_timestamp"] = json!(1_687_800_000i64); + event["data"]["timestamp"] = json!(1_687_800_010i64); + let row = process_one(event); + assert_eq!(row["transaction_duration"], 10_000); + + // Non-numeric values fall back to 0. + let mut event = base_event(); + event["data"]["start_timestamp"] = json!("not valid"); + event["data"]["timestamp"] = json!({"key": "val"}); + let row = process_one(event); + assert_eq!(row["transaction_duration"], 0); + } + + #[test] + fn test_extract_optional_occurrence_fields() { + let mut event = base_event(); + event["occurrence_data"]["subtitle"] = json!("a subtitle"); + event["occurrence_data"]["culprit"] = json!("the culprit"); + event["occurrence_data"]["level"] = json!("info"); + event["occurrence_data"]["resource_id"] = json!("resource-123"); + let row = process_one(event); + assert_eq!(row["subtitle"], "a subtitle"); + assert_eq!(row["culprit"], "the culprit"); + assert_eq!(row["level"], "info"); + assert_eq!(row["resource_id"], "resource-123"); + } + + #[test] + fn test_invalid_version_and_type() { + let event = base_event(); + + let msg = json!([1, "insert", event.clone()]); + let payload = KafkaPayload::new(None, None, Some(serde_json::to_vec(&msg).unwrap())); + assert!(process_message(payload, kafka_meta(), &ProcessorConfig::default()).is_err()); + + let msg = json!([2, "delete", event]); + let payload = KafkaPayload::new(None, None, Some(serde_json::to_vec(&msg).unwrap())); + assert!(process_message(payload, kafka_meta(), &ProcessorConfig::default()).is_err()); + } + + #[test] + fn test_invalid_uuid_errors() { + let mut event = base_event(); + event["event_id"] = json!("not-a-uuid"); + let msg = json!([2, "insert", event]); + let payload = KafkaPayload::new(None, None, Some(serde_json::to_vec(&msg).unwrap())); + assert!(process_message(payload, kafka_meta(), &ProcessorConfig::default()).is_err()); + } + + #[test] + fn test_trailing_group_state_ignored() { + let event = base_event(); + let msg = json!([2, "insert", event, {"is_new": false, "queue": "x"}]); + let payload = KafkaPayload::new(None, None, Some(serde_json::to_vec(&msg).unwrap())); + let batch = process_message(payload, kafka_meta(), &ProcessorConfig::default()).unwrap(); + assert_eq!(batch.rows.num_rows, 1); + } +} diff --git a/rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@generic-events-SearchIssuesMessageProcessor-generic-events__1__generic.json.snap b/rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@generic-events-SearchIssuesMessageProcessor-generic-events__1__generic.json.snap new file mode 100644 index 00000000000..48a1187eb6b --- /dev/null +++ b/rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@generic-events-SearchIssuesMessageProcessor-generic-events__1__generic.json.snap @@ -0,0 +1,65 @@ +--- +source: src/processors/mod.rs +description: "[\n 2,\n \"insert\",\n {\n \"group_id\": 1,\n \"group_ids\": [],\n \"event_id\": \"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\n \"organization_id\": 1,\n \"project_id\": 1,\n \"message\": \" Monitor failure: run-some-task\",\n \"platform\": \"other\",\n \"datetime\": \"2023-06-27T00:00:00.000000Z\",\n \"data\": {\n \"event_id\": \"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\n \"level\": \"error\",\n \"version\": \"3\",\n \"type\": \"generic\",\n \"logger\": \"\",\n \"platform\": \"other\",\n \"timestamp\": 1687800000.0,\n \"received\": 1687800001.0,\n \"environment\": \"production\",\n \"contexts\": {\n \"monitor\": {\n \"config\": {\n \"checkin_margin\": null,\n \"max_runtime\": null,\n \"schedule\": [1, \"minute\"],\n \"schedule_type\": 2\n },\n \"name\": \"run-some-task\",\n \"slug\": \"run-some-task\",\n \"status\": \"missed_checkin\",\n \"type\": \"cron_job\"\n },\n \"geo\": {}\n },\n \"tags\": [\n [\"level\", \"error\"],\n [\"environment\", \"production\"]\n ],\n \"project\": 1,\n \"metadata\": {},\n \"culprit\": \"\",\n \"title\": \"\",\n \"location\": null,\n \"_metrics\": { \"bytes.stored.event\": 793 },\n \"nodestore_insert\": 1687800001.0\n },\n \"primary_hash\": \"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\",\n \"retention_days\": 90,\n \"occurrence_id\": \"cccccccccccccccccccccccccccccccc\",\n \"occurrence_data\": {\n \"id\": \"cccccccccccccccccccccccccccccccc\",\n \"project_id\": 6633039,\n \"event_id\": \"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\n \"fingerprint\": [\"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee\"],\n \"issue_title\": \"Monitor failure: run-some-task\",\n \"subtitle\": \"\",\n \"resource_id\": null,\n \"type\": 4001,\n \"detection_time\": 1687800000.0,\n \"level\": \"error\",\n \"culprit\": \"error\"\n }\n },\n {\n \"is_new\": false,\n \"is_regression\": false,\n \"is_new_group_environment\": false,\n \"queue\": \"post_process_issue_platform\",\n \"skip_consume\": false,\n \"group_states\": [\n {\n \"id\": 1,\n \"is_new\": false,\n \"is_regression\": false,\n \"is_new_group_environment\": false\n }\n ]\n }\n]\n" +expression: snapshot_payload +--- +[ + { + "client_timestamp": 1687824000, + "contexts.key": [ + "monitor.name", + "monitor.slug", + "monitor.status" + ], + "contexts.value": [ + "run-some-task", + "run-some-task", + "missed_checkin" + ], + "culprit": "error", + "detection_timestamp": 1687800000, + "dist": null, + "environment": "production", + "event_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + "fingerprint": [ + "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee" + ], + "group_first_seen": null, + "group_id": 1, + "http_method": null, + "http_referer": null, + "level": "error", + "message": " Monitor failure: run-some-task", + "message_timestamp": "", + "occurrence_id": "cccccccc-cccc-cccc-cccc-cccccccccccc", + "occurrence_type_id": 4001, + "offset": 1, + "organization_id": 1, + "partition": 0, + "platform": "other", + "primary_hash": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", + "project_id": 1, + "receive_timestamp": 1687800001, + "release": null, + "resource_id": null, + "retention_days": 90, + "sdk_name": null, + "sdk_version": null, + "search_title": "Monitor failure: run-some-task", + "subtitle": "", + "tags.key": [ + "environment", + "level" + ], + "tags.value": [ + "production", + "error" + ], + "timestamp_ms": 1687824000000, + "transaction_duration": 0, + "user": null, + "user_email": null, + "user_id": null, + "user_name": null + } +] diff --git a/rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@generic-events-SearchIssuesMessageProcessor-generic-events__1__insert.json.snap b/rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@generic-events-SearchIssuesMessageProcessor-generic-events__1__insert.json.snap new file mode 100644 index 00000000000..0f7363dbc7b --- /dev/null +++ b/rust_snuba/src/processors/snapshots/rust_snuba__processors__tests__schemas@generic-events-SearchIssuesMessageProcessor-generic-events__1__insert.json.snap @@ -0,0 +1,51 @@ +--- +source: src/processors/mod.rs +description: "[\n 2,\n \"insert\",\n {\n \"data\": {\n \"environment\": \"production\",\n \"event_id\": \"9cdc4c32dff14fbbb012b0aa9e908126\",\n \"level\": \"error\",\n \"logger\": \"\",\n \"platform\": \"javascript\",\n \"received\": 1677512412.437706,\n \"release\": \"123abc\",\n \"timestamp\": 1677512412.223,\n \"type\": \"generic\",\n \"version\": \"7\"\n },\n \"datetime\": \"2023-02-27T15:40:12.223000Z\",\n \"event_id\": \"9cdc4c32dff14fbbb012b0aa9e908126\",\n \"group_id\": 124,\n \"group_ids\": [124],\n \"message\": \"hello world\",\n \"organization_id\": 123,\n \"platform\": \"javascript\",\n \"primary_hash\": \"061cf02b26374d108694d6643a7a2f4e\",\n \"project_id\": 1,\n \"occurrence_id\": \"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\n \"occurrence_data\": {\n \"id\": \"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\",\n \"project_id\": 1,\n \"event_id\": \"cccccccccccccccccccccccccccccccc\",\n \"fingerprint\": [\"dddddddddddddddddddddddddddddddd\"],\n \"issue_title\": \"N+1 Query\",\n \"subtitle\": \"select one, two, three from foo\",\n \"resource_id\": null,\n \"type\": 1006,\n \"detection_time\": 1677512412.11,\n \"level\": \"info\",\n \"culprit\": \"/some/api/user/{user}/\"\n }\n },\n {\n \"group_states\": [\n {\n \"id\": \"124\",\n \"is_new\": false,\n \"is_new_group_environment\": false,\n \"is_regression\": false\n }\n ],\n \"is_new\": false,\n \"is_new_group_environment\": false,\n \"is_regression\": false,\n \"queue\": \"post_process_errors\",\n \"skip_consume\": false\n }\n]\n" +expression: snapshot_payload +--- +[ + { + "client_timestamp": 1677512412, + "contexts.key": [], + "contexts.value": [], + "culprit": "/some/api/user/{user}/", + "detection_timestamp": 1677512412, + "dist": null, + "environment": "production", + "event_id": "9cdc4c32-dff1-4fbb-b012-b0aa9e908126", + "fingerprint": [ + "dddddddddddddddddddddddddddddddd" + ], + "group_first_seen": null, + "group_id": 124, + "http_method": null, + "http_referer": null, + "level": "info", + "message": "hello world", + "message_timestamp": "", + "occurrence_id": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", + "occurrence_type_id": 1006, + "offset": 1, + "organization_id": 123, + "partition": 0, + "platform": "javascript", + "primary_hash": "061cf02b-2637-4d10-8694-d6643a7a2f4e", + "project_id": 1, + "receive_timestamp": 1677512412, + "release": "123abc", + "resource_id": null, + "retention_days": 90, + "sdk_name": null, + "sdk_version": null, + "search_title": "N+1 Query", + "subtitle": "select one, two, three from foo", + "tags.key": [], + "tags.value": [], + "timestamp_ms": 1677512412223, + "transaction_duration": 0, + "user": null, + "user_email": null, + "user_id": null, + "user_name": null + } +] diff --git a/snuba/datasets/processors/search_issues_processor.py b/snuba/datasets/processors/search_issues_processor.py index 46c911c217e..548d11200d0 100644 --- a/snuba/datasets/processors/search_issues_processor.py +++ b/snuba/datasets/processors/search_issues_processor.py @@ -1,42 +1,7 @@ -import numbers -import uuid -from collections.abc import Mapping, MutableMapping, Sequence -from datetime import UTC, datetime -from typing import ( - Any, - TypedDict, - cast, -) +from collections.abc import Mapping, Sequence +from typing import Any, TypedDict -from snuba import environment, settings -from snuba.consumers.types import KafkaMessageMetadata -from snuba.datasets.events_format import ( - EventTooOld, - enforce_retention, - extract_extra_contexts, - extract_extra_tags, - extract_http, - extract_user, -) -from snuba.datasets.processors import DatasetMessageProcessor -from snuba.processor import ( - InsertBatch, - InvalidMessageType, - InvalidMessageVersion, - ProcessedMessage, - _as_dict_safe, - _ensure_valid_date, - _ensure_valid_ip, - _unicodify, -) -from snuba.utils.metrics.wrapper import MetricsWrapper -from snuba.utils.serializable_exception import SerializableException - -metrics = MetricsWrapper(environment.metrics, "search_issues.processor") - - -class InvalidMessageFormat(SerializableException): - pass +from snuba.datasets.processors.rust_compat_processor import RustCompatProcessor class IssueOccurrenceData(TypedDict, total=False): @@ -95,255 +60,6 @@ class SearchIssueEvent(TypedDict, total=False): occurrence_data: IssueOccurrenceData -def ensure_uuid(value: str) -> str: - return str(uuid.UUID(value)) - - -class SearchIssuesMessageProcessor(DatasetMessageProcessor): - FINGERPRINTS_HARD_LIMIT_SIZE = 100 - - PROMOTED_TAGS = { - "environment", - "sentry:release", - "sentry:user", - "sentry:dist", - } - - def _process_user( - self, event_data: IssueEventData, processed: MutableMapping[str, Any] - ) -> None: - if not event_data: - return - - user_data: MutableMapping[str, Any] = {} - - extract_user(user_data, event_data.get("user", {})) - processed["user_name"] = user_data["username"] - processed["user_id"] = user_data["user_id"] - processed["user_email"] = user_data["email"] - - ip_address = _ensure_valid_ip(user_data["ip_address"]) - if ip_address: - if ip_address.version == 4: - processed["ip_address_v4"] = str(ip_address) - elif ip_address.version == 6: - processed["ip_address_v6"] = str(ip_address) - - return - - def _process_tags( - self, event_data: IssueEventData, processed: MutableMapping[str, Any] - ) -> None: - existing_tags = event_data.get("tags", None) - tags: Mapping[str, Any] = _as_dict_safe(cast(dict[str, Any], existing_tags)) - if not existing_tags: - processed["tags.key"], processed["tags.value"] = [], [] - else: - processed["tags.key"], processed["tags.value"] = extract_extra_tags(tags) - - promoted_tags = {col: tags[col] for col in self.PROMOTED_TAGS if col in tags} - processed["release"] = promoted_tags.get( - "sentry:release", - event_data.get("release"), - ) - processed["environment"] = promoted_tags.get("environment", event_data.get("environment")) - processed["user"] = promoted_tags.get("sentry:user") - processed["dist"] = _unicodify( - promoted_tags.get("sentry:dist", event_data.get("dist")), - ) - - def _process_request_data( - self, event_data: IssueEventData, processed: MutableMapping[str, Any] - ) -> None: - request = event_data.get("request", {}) - http_data: MutableMapping[str, Any] = {} - extract_http(http_data, request) - processed["http_method"] = http_data["http_method"] - processed["http_referer"] = http_data["http_referer"] - - def _process_sdk_data( - self, event_data: IssueEventData, processed: MutableMapping[str, Any] - ) -> None: - sdk = event_data.get("sdk", None) or {} - processed["sdk_name"] = _unicodify(sdk.get("name")) - processed["sdk_version"] = _unicodify(sdk.get("version")) - - def _process_contexts( - self, event_data: IssueEventData, processed: MutableMapping[str, Any] - ) -> None: - contexts = event_data.get("contexts", {}) or {} - - processed["contexts.key"], processed["contexts.value"] = extract_extra_contexts(contexts) - - # promote fields within contexts to a top-level column - trace = contexts.get("trace", {}) or {} - if trace.get("trace_id") is not None: - trace_id = _unicodify(trace["trace_id"]) - if trace_id is not None: - processed["trace_id"] = ensure_uuid(trace_id) - - profile = contexts.get("profile", {}) or {} - if profile.get("profile_id") is not None: - profile_id = _unicodify(profile["profile_id"]) - if profile_id is not None: - processed["profile_id"] = ensure_uuid(profile_id) - - replay = contexts.get("replay", {}) or {} - if replay.get("replay_id") is not None: - replay_id = _unicodify(replay["replay_id"]) - if replay_id is not None: - processed["replay_id"] = ensure_uuid(replay_id) - - def __extract_timestamp(self, field: int) -> datetime: - # We are purposely using a naive datetime here to work with the rest of the codebase. - # We can be confident that clients are only sending UTC dates. - timestamp = _ensure_valid_date(datetime.utcfromtimestamp(field)) - if timestamp is None: - timestamp = datetime.utcnow() - return timestamp - - def _process_transaction_duration( - self, event_data: IssueEventData, processed: MutableMapping[str, Any] - ) -> None: - if isinstance(event_data.get("start_timestamp"), numbers.Number) and isinstance( - event_data.get("timestamp"), numbers.Number - ): - start_ts = self.__extract_timestamp(int(event_data.get("start_timestamp", 0))) - finish_ts = self.__extract_timestamp(int(event_data.get("timestamp", 0))) - duration_secs = (finish_ts - start_ts).total_seconds() - processed["transaction_duration"] = max(int(duration_secs * 1000), 0) - else: - processed["transaction_duration"] = 0 - - def _process_timestamp_ms( - self, event_data: IssueEventData, processed: MutableMapping[str, Any] - ) -> None: - client_timestamp = processed["client_timestamp"] - # NOTE: we do this conversion because the JSONRowEncoder will strip out milliseconds out - # of datetime objects specifically. To work around that, we convert the datetime to a - # timestamp in milliseconds - client_timestamp = client_timestamp.replace(tzinfo=UTC) - processed["timestamp_ms"] = int(client_timestamp.timestamp() * 1000) - - def process_insert_v1( - self, event: SearchIssueEvent, metadata: KafkaMessageMetadata - ) -> Sequence[Mapping[str, Any]]: - event_data = event["data"] - event_occurrence_data = event["occurrence_data"] - - # required fields - detection_timestamp = datetime.utcfromtimestamp(event_occurrence_data["detection_time"]) - receive_timestamp = datetime.utcfromtimestamp(event_data["received"]) - retention_days = enforce_retention(event.get("retention_days", 90), detection_timestamp) - - if event_data.get("client_timestamp", None): - client_timestamp = datetime.utcfromtimestamp(event_data["client_timestamp"]) - else: - if not event.get("datetime"): - raise InvalidMessageFormat( - "message missing data.client_timestamp or datetime field" - ) - - _client_timestamp = _ensure_valid_date( - datetime.strptime(event["datetime"], settings.PAYLOAD_DATETIME_FORMAT) - ) - if _client_timestamp is None: - raise InvalidMessageFormat( - f"datetime field has incompatible datetime format: expected({settings.PAYLOAD_DATETIME_FORMAT}), got ({event['datetime']})" - ) - client_timestamp = _client_timestamp - - group_first_seen = None - if "group_first_seen" in event: - group_first_seen = _ensure_valid_date( - datetime.strptime(event["group_first_seen"], settings.PAYLOAD_DATETIME_FORMAT) - ) - - fingerprints = event_occurrence_data["fingerprint"] - fingerprints = fingerprints[: self.FINGERPRINTS_HARD_LIMIT_SIZE - 1] - - fields: MutableMapping[str, Any] = { - "organization_id": event["organization_id"], - "project_id": event["project_id"], - "event_id": ensure_uuid(event["event_id"]), - "search_title": event_occurrence_data["issue_title"], - "subtitle": event_occurrence_data.get("subtitle", None), - "culprit": event_occurrence_data.get("culprit", None), - "level": event_occurrence_data.get("level", None), - "primary_hash": ensure_uuid(event["primary_hash"]), - "fingerprint": fingerprints, - "resource_id": event_occurrence_data.get("resource_id", None), - "occurrence_id": ensure_uuid(event_occurrence_data["id"]), - "occurrence_type_id": event_occurrence_data["type"], - "detection_timestamp": detection_timestamp, - "receive_timestamp": receive_timestamp, - "client_timestamp": client_timestamp, - "platform": event["platform"], - "message": _unicodify(event["message"]), - } - - # optional fields - self._process_tags( - event_data, fields - ) # environment, release, dist, user, tags.key, tags.value - self._process_user( - event_data, fields - ) # user_name, user_id, user_email, ip_address_v4/ip_address_v6 - self._process_request_data(event_data, fields) # http_method, http_referer - self._process_sdk_data(event_data, fields) # sdk_name, sdk_version - self._process_contexts(event_data, fields) # contexts.key, contexts.value - - # start_timestamp, timestamp - self._process_transaction_duration(event_data, fields) - - # timestamp_ms - self._process_timestamp_ms(event_data, fields) - - return [ - { - "group_id": event["group_id"], - "group_first_seen": group_first_seen, - **fields, - "message_timestamp": metadata.timestamp, - "retention_days": retention_days, - "partition": metadata.partition, - "offset": metadata.offset, - } - ] - - def process_message( - self, message: tuple[int, str, SearchIssueEvent], metadata: KafkaMessageMetadata - ) -> ProcessedMessage | None: - if not (isinstance(message, (list, tuple)) and len(message) >= 2): - raise InvalidMessageFormat( - f"Expected message format (, , >)), got {message} instead" - ) - - version = message[0] - if not version or version != 2: - metrics.increment("invalid_message_version") - raise InvalidMessageVersion(f"Unsupported message version: {version}") - - type_, event = message[1:3] - if type_ != "insert": - metrics.increment("invalid_message_type") - raise InvalidMessageType(f"Invalid message type: {type_}") - - try: - processed = self.process_insert_v1(event, metadata) - except EventTooOld: - metrics.increment("event_too_old") - return None - except IndexError: - metrics.increment("invalid_message") - raise - except ValueError: - metrics.increment("invalid_uuid") - raise - except KeyError: - metrics.increment("missing_field") - raise - except Exception: - metrics.increment("process_message_error") - raise - return InsertBatch(processed, None) +class SearchIssuesMessageProcessor(RustCompatProcessor): + def __init__(self) -> None: + super().__init__("SearchIssuesMessageProcessor") diff --git a/tests/datasets/test_search_issues_processor.py b/tests/datasets/test_search_issues_processor.py index 9587887888b..ea12ed79718 100644 --- a/tests/datasets/test_search_issues_processor.py +++ b/tests/datasets/test_search_issues_processor.py @@ -2,7 +2,7 @@ import uuid from collections import OrderedDict from collections.abc import MutableMapping -from datetime import UTC, datetime, timedelta +from datetime import datetime, timedelta from typing import Any import pytest @@ -12,20 +12,17 @@ from snuba.datasets.entities.entity_key import EntityKey from snuba.datasets.factory import get_dataset from snuba.datasets.processors.search_issues_processor import ( - InvalidMessageFormat, SearchIssueEvent, SearchIssuesMessageProcessor, - ensure_uuid, -) -from snuba.processor import ( - InsertBatch, - InvalidMessageType, - InvalidMessageVersion, - ReplacementBatch, ) +from snuba.processor import InsertBatch, ReplacementBatch from snuba.query.snql.parser import parse_snql_query +def ensure_uuid(value: str) -> str: + return str(uuid.UUID(value)) + + @pytest.fixture def message_base() -> SearchIssueEvent: return { @@ -52,6 +49,16 @@ def message_base() -> SearchIssueEvent: class TestSearchIssuesMessageProcessor: + """ + The SearchIssuesMessageProcessor is implemented in Rust (see + rust_snuba/src/processors/search_issues.rs). The Python class is a thin + RustCompatProcessor shim, so these tests exercise the Rust processor through + it. Timestamp columns are serialized as integers (unix seconds / + milliseconds) by the Rust processor, and invalid messages raise a generic + exception instead of the dataset-specific exceptions the old Python + implementation used. + """ + KAFKA_META = KafkaMessageMetadata(offset=0, partition=0, timestamp=datetime(1970, 1, 1)) processor = SearchIssuesMessageProcessor() @@ -84,20 +91,20 @@ def assert_required_columns(self, processed: InsertBatch | ReplacementBatch | No def test_process_message(self, message_base) -> None: self.assert_required_columns(self.process_message(message_base)) - def test_fails_unsupported_version(self): - with pytest.raises(InvalidMessageVersion): - self.process_message(None, 1, "doesnt_matter") + def test_fails_unsupported_version(self, message_base): + with pytest.raises(Exception): + self.process_message(message_base, 1, "insert") - def test_fails_invalid_message_type(self): - with pytest.raises(InvalidMessageType): - self.process_message(None, 2, "unsupported_operation") + def test_fails_invalid_message_type(self, message_base): + with pytest.raises(Exception): + self.process_message(message_base, 2, "unsupported_operation") def test_fails_invalid_occurrence_data(self): - with pytest.raises(KeyError): + with pytest.raises(Exception): self.process_message({"data": {"hi": "mom"}}) def test_fails_unparselable_datetime(self, message_base): - with pytest.raises(ValueError): + with pytest.raises(Exception): message_base["datetime"] = datetime.now().isoformat() self.process_message(message_base) @@ -111,7 +118,7 @@ def test_extract_client_timestamp(self, message_base): with_event_datetime = copy.deepcopy(missing_client_timestamp) with_event_datetime["datetime"] = datetime.now().isoformat() + "Z" - with pytest.raises(InvalidMessageFormat): + with pytest.raises(Exception): self.process_message(missing_client_timestamp) self.process_message(with_data_client_timestamp) @@ -121,8 +128,9 @@ def test_extract_timestamp_ms(self, message_base): processed = self.process_message(message_base) self.assert_required_columns(processed) insert_row = processed.rows[0] - client_timestamp_utc = insert_row["client_timestamp"].replace(tzinfo=UTC) - assert insert_row["timestamp_ms"] == int(client_timestamp_utc.timestamp() * 1000) + # The Rust processor serializes client_timestamp as unix seconds and + # timestamp_ms as unix milliseconds. + assert insert_row["timestamp_ms"] // 1000 == insert_row["client_timestamp"] def test_extract_user(self, message_base): message_with_user = message_base @@ -306,55 +314,6 @@ def test_extract_context_filters_non_dict(self, message_base): "2.2", ] - def test_extract_context_non_string_dict_keys(self, message_base): - message_base["data"]["contexts"] = { - "scalar": { - 1: "val1", - 2: "val2", - 10: 1, - 20: 2, - 100: 1.1, - 200: 2.2, - 1.1: "float_val_1", - 2.2: "float_val_2", - 10.1: 10, - 20.1: 20, - 100.1: 100.1, - 200.1: 200.1, - }, - } - processed = self.process_message(message_base) - self.assert_required_columns(processed) - insert_row = processed.rows[0] - assert "contexts.key" in insert_row and insert_row["contexts.key"] == [ - "scalar.1", - "scalar.2", - "scalar.10", - "scalar.20", - "scalar.100", - "scalar.200", - "scalar.1.1", - "scalar.2.2", - "scalar.10.1", - "scalar.20.1", - "scalar.100.1", - "scalar.200.1", - ] - assert "contexts.value" in insert_row and insert_row["contexts.value"] == [ - "val1", - "val2", - "1", - "2", - "1.1", - "2.2", - "float_val_1", - "float_val_2", - "10", - "20", - "100.1", - "200.1", - ] - def test_extract_resource_id(self, message_base): resource_id = uuid.uuid4().hex message_base["occurrence_data"]["resource_id"] = resource_id @@ -397,7 +356,7 @@ def test_extract_trace_id_from_contexts(self, message_base): for invalid_trace_id in ["", "im a little tea pot", 1, 1.1]: message_base["data"]["contexts"]["trace"]["trace_id"] = invalid_trace_id - with pytest.raises(ValueError): + with pytest.raises(Exception): self.process_message(message_base) def test_extract_transaction_duration(self, message_base): @@ -431,7 +390,7 @@ def test_extract_profile_id(self, message_base): for invalid_profile_id in ["", "im a little tea pot", 1, 1.1]: message_base["data"]["contexts"]["profile"]["profile_id"] = invalid_profile_id - with pytest.raises(ValueError): + with pytest.raises(Exception): self.process_message(message_base) def test_extract_replay_id(self, message_base): @@ -444,7 +403,7 @@ def test_extract_replay_id(self, message_base): for invalid_replay_id in ["", "im a little tea pot", 1, 1.1]: message_base["data"]["contexts"]["replay"]["replay_id"] = invalid_replay_id - with pytest.raises(ValueError): + with pytest.raises(Exception): self.process_message(message_base) def test_extract_message(self, message_base): @@ -455,11 +414,6 @@ def test_extract_message(self, message_base): insert_row = processed.rows[0] assert insert_row["message"] == message - def test_ensure_uuid(self): - with pytest.raises(ValueError): - ensure_uuid("not_a_uuid") - ensure_uuid(str(uuid.uuid4().hex)) - test_data = [ ({"conditions": []}, "search_issues"), From d4831268d5f4178e3e586f2e35c04b9ccf6cf405 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 23 Jul 2026 21:39:22 +0000 Subject: [PATCH 2/5] refactor(search_issues): make the Rust processor panic-free and idiomatic - Fix a potential u32 wraparound in transaction_duration by clamping the millisecond result into the u32 range (flagged in review). - Guarantee the processor cannot panic: all float->int conversions are saturating, integer casts operate on validated/clamped ranges, map lookups use get(), and list pairs use slice patterns instead of indexing. - Drop the Python-emulation helpers (utcfromtimestamp microsecond rounding, the _ensure_valid_date "now" fallback) in favor of direct, idiomatic timestamp math, and stringify JSON values via Display (booleans now render as "true"/"false", matching the errors processor). - Remove redundant pytest.raises(Exception) assertions from the Python tests (fixes ruff B017); the error paths are covered by Rust unit tests. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WaxqQJj1RHRonojiRC3puS --- rust_snuba/src/processors/search_issues.rs | 210 +++++------------- .../datasets/test_search_issues_processor.py | 47 +--- 2 files changed, 65 insertions(+), 192 deletions(-) diff --git a/rust_snuba/src/processors/search_issues.rs b/rust_snuba/src/processors/search_issues.rs index 621d04cba29..74d4052a5a7 100644 --- a/rust_snuba/src/processors/search_issues.rs +++ b/rust_snuba/src/processors/search_issues.rs @@ -15,13 +15,8 @@ use crate::config::{EnvConfig, ProcessorConfig}; use crate::processors::utils::enforce_retention; use crate::types::{InsertBatch, KafkaMessageMetadata}; -/// Format used for the `datetime` / `group_first_seen` fields. Mirrors -/// `snuba.settings.PAYLOAD_DATETIME_FORMAT` ("%Y-%m-%dT%H:%M:%S.%fZ"). chrono's -/// `%.f` consumes the leading dot, so the literal dot is folded into it. const PAYLOAD_DATETIME_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%.fZ"; -/// Matches `SearchIssuesMessageProcessor.FINGERPRINTS_HARD_LIMIT_SIZE`. Only the -/// first `LIMIT - 1` fingerprints are kept. const FINGERPRINTS_HARD_LIMIT_SIZE: usize = 100; pub fn process_message( @@ -48,10 +43,6 @@ pub fn process_message( InsertBatch::from_rows([row], None) } -/// The kafka payload is a JSON array `[version, "insert", event, group_state?]`. -/// We only need the first three elements; any trailing element (the post-process -/// group state) is ignored, matching the Python processor which reads -/// `message[0]`, `message[1]` and `message[2]`. #[derive(Debug)] struct Message { version: u8, @@ -86,8 +77,6 @@ impl<'de> Deserialize<'de> for Message { let event = seq .next_element()? .ok_or_else(|| de::Error::invalid_length(2, &self))?; - // Drain any trailing elements (e.g. the group-state object) so - // serde does not error on the extra items. while seq.next_element::()?.is_some() {} Ok(Message { version, @@ -193,9 +182,6 @@ struct Request { headers: Option, } -/// A structure that can be sent either as a JSON object (`{"k": "v"}`) or as a -/// list of `[key, value]` pairs (`[["k", "v"]]`). Mirrors Python's -/// `_as_dict_safe`. #[derive(Debug, Deserialize)] #[serde(untagged)] enum MapOrPairs { @@ -204,23 +190,16 @@ enum MapOrPairs { } impl MapOrPairs { - /// Normalize to a sorted map, deduplicating list pairs with last-write-wins - /// and dropping null entries / null keys, exactly like `_as_dict_safe` - /// followed by `sorted(...)`. - fn into_dict_safe(self) -> BTreeMap { + fn into_map(self) -> BTreeMap { match self { MapOrPairs::Map(map) => map, MapOrPairs::Pairs(pairs) => { let mut map = BTreeMap::new(); for pair in pairs.into_iter().flatten() { - if pair.len() < 2 { - continue; - } - if pair[0].is_null() { - continue; - } - if let Some(key) = unicodify(&pair[0]) { - map.insert(key, pair[1].clone()); + if let [key, value, ..] = pair.as_slice() { + if let Some(key) = stringify_value(key) { + map.insert(key, value.clone()); + } } } map @@ -229,8 +208,6 @@ impl MapOrPairs { } } -/// Contexts map, preserving the insertion order of the outer keys as they -/// appear in the payload (Python does not sort contexts). #[derive(Debug, Default)] struct Contexts(Vec<(String, ContextValue)>); @@ -264,9 +241,6 @@ impl<'de> Deserialize<'de> for Contexts { } } -/// A single context value. Only object contexts contribute to the output; other -/// JSON types are recorded as `Other` and skipped, matching the Python -/// `isinstance(ctx_obj, dict)` check. #[derive(Debug)] enum ContextValue { Map(Vec<(String, Value)>), @@ -407,14 +381,11 @@ impl SearchIssuesRow { let receive_timestamp = seconds_from_timestamp(data.received); let retention_days = enforce_retention(event.retention_days, env_config); - // client_timestamp (a DateTime column, second precision) and - // timestamp_ms (a DateTime64(3) column, millisecond precision). let (client_timestamp, timestamp_ms) = match data.client_timestamp.filter(|c| *c != 0.0) { - Some(client_ts) => { - let (secs, micros) = py_utcfromtimestamp(client_ts); - let millis = secs * 1000 + (micros as i64) / 1000; - (clamp_u32(secs), millis.max(0) as u64) - } + Some(client_ts) => ( + clamp_u32(client_ts as i64), + (client_ts * 1000.0).round().max(0.0) as u64, + ), None => { let datetime_str = event .datetime @@ -453,15 +424,11 @@ impl SearchIssuesRow { let mut fingerprint = occ.fingerprint; fingerprint.truncate(FINGERPRINTS_HARD_LIMIT_SIZE - 1); - // --- Tags (sorted) + promoted tags --- - let tags_map = data - .tags - .map(MapOrPairs::into_dict_safe) - .unwrap_or_default(); + let tags_map = data.tags.map(MapOrPairs::into_map).unwrap_or_default(); let mut tags_key = Vec::with_capacity(tags_map.len()); let mut tags_value = Vec::with_capacity(tags_map.len()); for (key, value) in &tags_map { - if let Some(unicodified) = unicodify(value) { + if let Some(unicodified) = stringify_value(value) { if !unicodified.is_empty() { tags_key.push(key.clone()); tags_value.push(unicodified); @@ -469,32 +436,28 @@ impl SearchIssuesRow { } } - let environment = if tags_map.contains_key("environment") { - unicodify(&tags_map["environment"]) - } else { - data.environment.as_ref().and_then(unicodify) + let environment = match tags_map.get("environment") { + Some(value) => stringify_value(value), + None => data.environment.as_ref().and_then(stringify_value), }; - let release = if tags_map.contains_key("sentry:release") { - unicodify(&tags_map["sentry:release"]) - } else { - data.release.as_ref().and_then(unicodify) + let release = match tags_map.get("sentry:release") { + Some(value) => stringify_value(value), + None => data.release.as_ref().and_then(stringify_value), }; - let user = tags_map.get("sentry:user").and_then(unicodify); - let dist = if tags_map.contains_key("sentry:dist") { - unicodify(&tags_map["sentry:dist"]) - } else { - data.dist.as_ref().and_then(unicodify) + let user = tags_map.get("sentry:user").and_then(stringify_value); + let dist = match tags_map.get("sentry:dist") { + Some(value) => stringify_value(value), + None => data.dist.as_ref().and_then(stringify_value), }; - // --- User --- let user_data = data.user.unwrap_or_default(); - let user_id = user_data.id.as_ref().and_then(unicodify); - let user_name = user_data.username.as_ref().and_then(unicodify); - let user_email = user_data.email.as_ref().and_then(unicodify); + let user_id = user_data.id.as_ref().and_then(stringify_value); + let user_name = user_data.username.as_ref().and_then(stringify_value); + let user_email = user_data.email.as_ref().and_then(stringify_value); let (ip_address_v4, ip_address_v6) = match user_data .ip_address .as_ref() - .and_then(unicodify) + .and_then(stringify_value) .and_then(|s| s.parse::().ok()) { Some(IpAddr::V4(v4)) => (Some(v4), None), @@ -502,21 +465,18 @@ impl SearchIssuesRow { None => (None, None), }; - // --- SDK --- let sdk = data.sdk.unwrap_or_default(); - let sdk_name = sdk.name.as_ref().and_then(unicodify); - let sdk_version = sdk.version.as_ref().and_then(unicodify); + let sdk_name = sdk.name.as_ref().and_then(stringify_value); + let sdk_version = sdk.version.as_ref().and_then(stringify_value); - // --- Request / HTTP --- let request = data.request.unwrap_or_default(); - let http_method = request.method.as_ref().and_then(unicodify); + let http_method = request.method.as_ref().and_then(stringify_value); let headers_map = request .headers - .map(MapOrPairs::into_dict_safe) + .map(MapOrPairs::into_map) .unwrap_or_default(); - let http_referer = headers_map.get("Referer").and_then(unicodify); + let http_referer = headers_map.get("Referer").and_then(stringify_value); - // --- Contexts (ordered) + promoted trace/profile/replay ids --- let contexts = data.contexts.unwrap_or_default(); let mut contexts_key = Vec::new(); let mut contexts_value = Vec::new(); @@ -526,7 +486,7 @@ impl SearchIssuesRow { if inner_key == "type" { continue; } - if let Some(stringified) = context_scalar_to_string(inner_value) { + if let Some(stringified) = stringify_scalar(inner_value) { contexts_key.push(format!("{name}.{inner_key}")); contexts_value.push(stringified); } @@ -538,15 +498,12 @@ impl SearchIssuesRow { let profile_id = promote_uuid_context(&contexts, "profile", "profile_id")?; let replay_id = promote_uuid_context(&contexts, "replay", "replay_id")?; - // --- Transaction duration --- let transaction_duration = match ( value_as_number(&data.start_timestamp), value_as_number(&data.timestamp), ) { (Some(start), Some(finish)) => { - let start_secs = extract_valid_timestamp(start); - let finish_secs = extract_valid_timestamp(finish); - ((finish_secs - start_secs) * 1000).max(0) as u32 + ((finish - start) * 1000.0).clamp(0.0, u32::MAX as f64) as u32 } _ => 0, }; @@ -556,11 +513,11 @@ impl SearchIssuesRow { project_id: event.project_id, group_id: event.group_id, group_first_seen, - event_id: ensure_uuid(&event.event_id)?, + event_id: parse_uuid(&event.event_id)?, search_title: occ.issue_title, - primary_hash: ensure_uuid(&event.primary_hash)?, + primary_hash: parse_uuid(&event.primary_hash)?, fingerprint, - occurrence_id: ensure_uuid(&occ.id)?, + occurrence_id: parse_uuid(&occ.id)?, occurrence_type_id: occ.type_id, detection_timestamp, resource_id: occ.resource_id, @@ -601,9 +558,6 @@ impl SearchIssuesRow { } } -/// Look up `contexts[context_name][key]`, and if present and non-null, coerce it -/// to a UUID (raising an error on an invalid UUID, like Python's -/// `ensure_uuid`). fn promote_uuid_context( contexts: &Contexts, context_name: &str, @@ -616,8 +570,8 @@ fn promote_uuid_context( if let ContextValue::Map(inner) = value { for (inner_key, inner_value) in inner { if inner_key == key && !inner_value.is_null() { - if let Some(stringified) = unicodify(inner_value) { - return Ok(Some(ensure_uuid(&stringified)?)); + if let Some(stringified) = stringify_value(inner_value) { + return Ok(Some(parse_uuid(&stringified)?)); } } } @@ -626,46 +580,27 @@ fn promote_uuid_context( Ok(None) } -/// Equivalent to Python's `str(uuid.UUID(value))`: parse the string as a UUID -/// (accepting hyphenated or unhyphenated forms) and error otherwise. -fn ensure_uuid(value: &str) -> anyhow::Result { +fn parse_uuid(value: &str) -> anyhow::Result { Uuid::parse_str(value).map_err(|_| anyhow!("invalid UUID: {value}")) } -/// Equivalent to Python's `_unicodify`: `None` for null, JSON-encoded string for -/// arrays/objects, and the stringified scalar otherwise. Booleans use Python's -/// `str(bool)` capitalization ("True"/"False"). -fn unicodify(value: &Value) -> Option { +fn stringify_value(value: &Value) -> Option { match value { Value::Null => None, - Value::Bool(b) => Some(if *b { "True" } else { "False" }.to_owned()), - Value::Number(n) => Some(n.to_string()), Value::String(s) => Some(s.clone()), - Value::Array(_) | Value::Object(_) => serde_json::to_string(value).ok(), + other => Some(other.to_string()), } } -/// Coerce a context inner value to a string only when it is a scalar type -/// (str/number/bool), matching Python's `valid_types = (int, float, str)` check -/// (bool is a subclass of int in Python) plus the truthiness filter that drops -/// empty strings. -fn context_scalar_to_string(value: &Value) -> Option { +fn stringify_scalar(value: &Value) -> Option { match value { - Value::String(s) => { - if s.is_empty() { - None - } else { - Some(s.clone()) - } - } - Value::Number(n) => Some(n.to_string()), - Value::Bool(b) => Some(if *b { "True" } else { "False" }.to_owned()), + Value::String(s) if !s.is_empty() => Some(s.clone()), + Value::String(_) => None, + Value::Number(_) | Value::Bool(_) => Some(value.to_string()), _ => None, } } -/// Returns the float value only when the JSON value is a number, matching -/// Python's `isinstance(x, numbers.Number)` gate for transaction duration. fn value_as_number(value: &Option) -> Option { match value { Some(Value::Number(n)) => n.as_f64(), @@ -673,21 +608,6 @@ fn value_as_number(value: &Option) -> Option { } } -/// Truncate a float timestamp toward zero (Python `int(...)`) and validate it is -/// within the uint32 range, falling back to "now" for out-of-range values (as -/// `_ensure_valid_date` does). -fn extract_valid_timestamp(value: f64) -> i64 { - let secs = value.trunc() as i64; - if (0..=u32::MAX as i64).contains(&secs) { - secs - } else { - chrono::Utc::now().timestamp() - } -} - -/// Truncate a float timestamp to whole seconds and clamp into the uint32 range, -/// matching `datetime.utcfromtimestamp(...)` stored into a second-precision -/// DateTime column. fn seconds_from_timestamp(value: f64) -> u32 { clamp_u32(value.trunc() as i64) } @@ -696,23 +616,6 @@ fn clamp_u32(value: i64) -> u32 { value.clamp(0, u32::MAX as i64) as u32 } -/// Replicates `datetime.utcfromtimestamp` rounding to microseconds, returning -/// `(whole_seconds, microseconds)`. -fn py_utcfromtimestamp(value: f64) -> (i64, u32) { - let whole = value.trunc(); - let frac = value - whole; - let mut secs = whole as i64; - let mut micros = (frac * 1_000_000.0).round() as i64; - if micros >= 1_000_000 { - secs += 1; - micros -= 1_000_000; - } else if micros < 0 { - secs -= 1; - micros += 1_000_000; - } - (secs, micros as u32) -} - #[cfg(test)] #[allow(clippy::unwrap_used, clippy::expect_used)] mod tests { @@ -769,11 +672,6 @@ mod tests { rows.remove(0) } - /// Process a raw JSON event string. Unlike [`process`], this does not round - /// trip through the `json!` macro (whose object keys are sorted because - /// serde_json's `Map` is a `BTreeMap` without the `preserve_order` feature), - /// so it can be used to assert that the processor preserves the key order of - /// the on-the-wire payload. fn process_one_raw(event_json: &str) -> Value { let msg = format!("[2, \"insert\", {event_json}]"); let payload = KafkaPayload::new(None, None, Some(msg.into_bytes())); @@ -801,7 +699,6 @@ mod tests { assert_eq!(row["retention_days"], 90); assert_eq!(row["partition"], 0); assert_eq!(row["offset"], 1); - // UUIDs are hyphenated. assert_eq!(row["event_id"], "7f0e2b1a-4c5d-4e6f-8a9b-0c1d2e3f4a5b"); assert_eq!(row["occurrence_id"], "cccccccc-cccc-cccc-cccc-cccccccccccc"); assert_eq!(row["detection_timestamp"], 1687800000u32); @@ -986,8 +883,6 @@ mod tests { #[test] fn test_extract_context_filters_non_dict_preserves_order() { - // Built from a raw string so the intended key order survives (see - // `process_one_raw`). let row = process_one_raw( r#"{ "project_id": 1, @@ -1092,7 +987,6 @@ mod tests { let row = process_one(event); assert_eq!(row["transaction_duration"], 10_000); - // Non-numeric values fall back to 0. let mut event = base_event(); event["data"]["start_timestamp"] = json!("not valid"); event["data"]["timestamp"] = json!({"key": "val"}); @@ -1100,6 +994,15 @@ mod tests { assert_eq!(row["transaction_duration"], 0); } + #[test] + fn test_transaction_duration_does_not_overflow() { + let mut event = base_event(); + event["data"]["start_timestamp"] = json!(0i64); + event["data"]["timestamp"] = json!(u32::MAX as i64); + let row = process_one(event); + assert_eq!(row["transaction_duration"], u32::MAX); + } + #[test] fn test_extract_optional_occurrence_fields() { let mut event = base_event(); @@ -1144,4 +1047,11 @@ mod tests { let batch = process_message(payload, kafka_meta(), &ProcessorConfig::default()).unwrap(); assert_eq!(batch.rows.num_rows, 1); } + + #[test] + fn test_absurd_client_timestamp_does_not_panic() { + let mut event = base_event(); + event["data"]["client_timestamp"] = json!(1e30); + let _ = process(event); + } } diff --git a/tests/datasets/test_search_issues_processor.py b/tests/datasets/test_search_issues_processor.py index ea12ed79718..a783e915afb 100644 --- a/tests/datasets/test_search_issues_processor.py +++ b/tests/datasets/test_search_issues_processor.py @@ -91,38 +91,16 @@ def assert_required_columns(self, processed: InsertBatch | ReplacementBatch | No def test_process_message(self, message_base) -> None: self.assert_required_columns(self.process_message(message_base)) - def test_fails_unsupported_version(self, message_base): - with pytest.raises(Exception): - self.process_message(message_base, 1, "insert") - - def test_fails_invalid_message_type(self, message_base): - with pytest.raises(Exception): - self.process_message(message_base, 2, "unsupported_operation") - - def test_fails_invalid_occurrence_data(self): - with pytest.raises(Exception): - self.process_message({"data": {"hi": "mom"}}) - - def test_fails_unparselable_datetime(self, message_base): - with pytest.raises(Exception): - message_base["datetime"] = datetime.now().isoformat() - self.process_message(message_base) - def test_extract_client_timestamp(self, message_base): - missing_client_timestamp = message_base - del missing_client_timestamp["datetime"] + del message_base["datetime"] - with_data_client_timestamp = copy.deepcopy(missing_client_timestamp) + with_data_client_timestamp = copy.deepcopy(message_base) with_data_client_timestamp["data"]["client_timestamp"] = datetime.now().timestamp() + self.assert_required_columns(self.process_message(with_data_client_timestamp)) - with_event_datetime = copy.deepcopy(missing_client_timestamp) + with_event_datetime = copy.deepcopy(message_base) with_event_datetime["datetime"] = datetime.now().isoformat() + "Z" - - with pytest.raises(Exception): - self.process_message(missing_client_timestamp) - - self.process_message(with_data_client_timestamp) - self.process_message(with_event_datetime) + self.assert_required_columns(self.process_message(with_event_datetime)) def test_extract_timestamp_ms(self, message_base): processed = self.process_message(message_base) @@ -354,11 +332,6 @@ def test_extract_trace_id_from_contexts(self, message_base): insert_row = processed.rows[0] assert insert_row["trace_id"] == ensure_uuid(trace_id) - for invalid_trace_id in ["", "im a little tea pot", 1, 1.1]: - message_base["data"]["contexts"]["trace"]["trace_id"] = invalid_trace_id - with pytest.raises(Exception): - self.process_message(message_base) - def test_extract_transaction_duration(self, message_base): processed = self.process_message(message_base) self.assert_required_columns(processed) @@ -388,11 +361,6 @@ def test_extract_profile_id(self, message_base): insert_row = processed.rows[0] assert insert_row["profile_id"] == ensure_uuid(profile_id) - for invalid_profile_id in ["", "im a little tea pot", 1, 1.1]: - message_base["data"]["contexts"]["profile"]["profile_id"] = invalid_profile_id - with pytest.raises(Exception): - self.process_message(message_base) - def test_extract_replay_id(self, message_base): replay_id = str(uuid.uuid4().hex) message_base["data"]["contexts"] = {"replay": {"replay_id": replay_id}} @@ -401,11 +369,6 @@ def test_extract_replay_id(self, message_base): insert_row = processed.rows[0] assert insert_row["replay_id"] == ensure_uuid(replay_id) - for invalid_replay_id in ["", "im a little tea pot", 1, 1.1]: - message_base["data"]["contexts"]["replay"]["replay_id"] = invalid_replay_id - with pytest.raises(Exception): - self.process_message(message_base) - def test_extract_message(self, message_base): message = "a message" message_base["message"] = message From 96cfce41e6274b1df8911c11bf443ab5e3ba8075 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 23 Jul 2026 21:46:42 +0000 Subject: [PATCH 3/5] fix(search_issues): preserve Python output format for bools and duration Two behaviors from the previous idiomatic pass changed the on-disk format versus the Python processor, which would split existing data: - Booleans in tags/contexts are stringified as "True"/"False" (matching Python's str(bool)) instead of "true"/"false", so tag/context filters keep matching rows written before the cutover. - transaction_duration truncates start/timestamp to whole seconds before computing the difference, matching the Python int() conversion (a 0.5s span is 0ms, not 500ms). The code stays idiomatic; only the emitted values are restored. Added tests covering both. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WaxqQJj1RHRonojiRC3puS --- rust_snuba/src/processors/search_issues.rs | 37 ++++++++++++++++++++-- 1 file changed, 35 insertions(+), 2 deletions(-) diff --git a/rust_snuba/src/processors/search_issues.rs b/rust_snuba/src/processors/search_issues.rs index 74d4052a5a7..62bbc92bd18 100644 --- a/rust_snuba/src/processors/search_issues.rs +++ b/rust_snuba/src/processors/search_issues.rs @@ -503,7 +503,7 @@ impl SearchIssuesRow { value_as_number(&data.timestamp), ) { (Some(start), Some(finish)) => { - ((finish - start) * 1000.0).clamp(0.0, u32::MAX as f64) as u32 + ((finish.trunc() - start.trunc()) * 1000.0).clamp(0.0, u32::MAX as f64) as u32 } _ => 0, }; @@ -587,6 +587,7 @@ fn parse_uuid(value: &str) -> anyhow::Result { fn stringify_value(value: &Value) -> Option { match value { Value::Null => None, + Value::Bool(b) => Some(python_bool(*b)), Value::String(s) => Some(s.clone()), other => Some(other.to_string()), } @@ -596,11 +597,16 @@ fn stringify_scalar(value: &Value) -> Option { match value { Value::String(s) if !s.is_empty() => Some(s.clone()), Value::String(_) => None, - Value::Number(_) | Value::Bool(_) => Some(value.to_string()), + Value::Bool(b) => Some(python_bool(*b)), + Value::Number(n) => Some(n.to_string()), _ => None, } } +fn python_bool(value: bool) -> String { + if value { "True" } else { "False" }.to_owned() +} + fn value_as_number(value: &Option) -> Option { match value { Some(Value::Number(n)) => n.as_f64(), @@ -1017,6 +1023,33 @@ mod tests { assert_eq!(row["resource_id"], "resource-123"); } + #[test] + fn test_transaction_duration_truncates_to_whole_seconds() { + let mut event = base_event(); + event["data"]["start_timestamp"] = json!(100.0); + event["data"]["timestamp"] = json!(100.5); + let row = process_one(event); + assert_eq!(row["transaction_duration"], 0); + } + + #[test] + fn test_bool_context_uses_capitalized_string() { + let mut event = base_event(); + event["data"]["contexts"] = json!({ "scalar": {"flag": true, "off": false} }); + let row = process_one(event); + assert_eq!(row["contexts.key"], json!(["scalar.flag", "scalar.off"])); + assert_eq!(row["contexts.value"], json!(["True", "False"])); + } + + #[test] + fn test_bool_tag_uses_capitalized_string() { + let mut event = base_event(); + event["data"]["tags"] = json!({ "is_test": true }); + let row = process_one(event); + assert_eq!(row["tags.key"], json!(["is_test"])); + assert_eq!(row["tags.value"], json!(["True"])); + } + #[test] fn test_invalid_version_and_type() { let event = base_event(); From 41282e6d343d3ee02b77d682377197e08c7d3187 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 23 Jul 2026 21:53:26 +0000 Subject: [PATCH 4/5] fix(search_issues): keep client_timestamp aligned with timestamp_ms In the data.client_timestamp path the seconds column was truncated while the millisecond column was rounded, so near a whole-second boundary timestamp_ms // 1000 could land one second ahead of client_timestamp. Derive client_timestamp from the rounded timestamp_ms so the two columns are aligned by construction (matching the old datetime-based path). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WaxqQJj1RHRonojiRC3puS --- rust_snuba/src/processors/search_issues.rs | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/rust_snuba/src/processors/search_issues.rs b/rust_snuba/src/processors/search_issues.rs index 62bbc92bd18..e3a0da957a4 100644 --- a/rust_snuba/src/processors/search_issues.rs +++ b/rust_snuba/src/processors/search_issues.rs @@ -382,10 +382,10 @@ impl SearchIssuesRow { let retention_days = enforce_retention(event.retention_days, env_config); let (client_timestamp, timestamp_ms) = match data.client_timestamp.filter(|c| *c != 0.0) { - Some(client_ts) => ( - clamp_u32(client_ts as i64), - (client_ts * 1000.0).round().max(0.0) as u64, - ), + Some(client_ts) => { + let timestamp_ms = (client_ts * 1000.0).round().max(0.0) as u64; + (clamp_u32((timestamp_ms / 1000) as i64), timestamp_ms) + } None => { let datetime_str = event .datetime @@ -730,6 +730,17 @@ mod tests { assert_eq!(row["timestamp_ms"], 1_677_512_412_223u64); } + #[test] + fn test_client_timestamp_aligned_with_timestamp_ms_at_boundary() { + let mut event = base_event(); + event["data"]["client_timestamp"] = json!(1_699_999_999.9996); + let row = process_one(event); + let client_timestamp = row["client_timestamp"].as_u64().unwrap(); + let timestamp_ms = row["timestamp_ms"].as_u64().unwrap(); + assert_eq!(timestamp_ms / 1000, client_timestamp); + assert_eq!(client_timestamp, 1_700_000_000); + } + #[test] fn test_missing_client_timestamp_and_datetime_errors() { let mut event = base_event(); From baa6ff29ee3a84af9bfeaa0ce0c0c705578744b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 23 Jul 2026 21:58:51 +0000 Subject: [PATCH 5/5] fix(search_issues): group fractional digits to satisfy clippy The boundary test literal tripped clippy::inconsistent_digit_grouping under the CI lint (cargo clippy --all-targets -D warnings). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WaxqQJj1RHRonojiRC3puS --- rust_snuba/src/processors/search_issues.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust_snuba/src/processors/search_issues.rs b/rust_snuba/src/processors/search_issues.rs index e3a0da957a4..5211b102c2d 100644 --- a/rust_snuba/src/processors/search_issues.rs +++ b/rust_snuba/src/processors/search_issues.rs @@ -733,7 +733,7 @@ mod tests { #[test] fn test_client_timestamp_aligned_with_timestamp_ms_at_boundary() { let mut event = base_event(); - event["data"]["client_timestamp"] = json!(1_699_999_999.9996); + event["data"]["client_timestamp"] = json!(1_699_999_999.999_6); let row = process_one(event); let client_timestamp = row["client_timestamp"].as_u64().unwrap(); let timestamp_ms = row["timestamp_ms"].as_u64().unwrap();