From c12309a2bedacafcdcfa4078659d2a1fce03f63e Mon Sep 17 00:00:00 2001 From: Prabhat Sharma Date: Mon, 5 Oct 2026 21:49:02 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20traces=20critical=20path,=20RED=20insig?= =?UTF-8?q?hts,=20span=20percentiles,=20request-scoped=20catalog,=20span?= =?UTF-8?q?=E2=86=92profile=20link=20and=20saved=20views=20(#15073)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> --- scripts/translations/.translation_state.json | 28 + .../src/request/organization/settings.rs | 27 + src/api/search/src/profiles/meta.rs | 35 +- src/api/search/src/search/saved_view.rs | 124 +- src/common/src/meta/organization.rs | 24 + src/common/src/meta/saved_view.rs | 59 + src/core/src/anomaly_detection.rs | 56 + src/core/src/ingestion/mod.rs | 45 +- src/core/src/traces/mod.rs | 1 + src/core/src/traces/red_insights.rs | 1258 +++++++++++++++++ src/db/src/organization.rs | 23 + src/db/src/saved_view.rs | 55 +- src/jobs/src/job/mod.rs | 4 + src/jobs/src/job/red_insights.rs | 328 +++++ .../aggregate_topk_exec/mod.rs | 39 +- .../codec/aggregate_topk_exec.rs | 8 +- .../physical_optimizer/aggregate_topk.rs | 179 ++- .../pages/tracesPages/tracesPage.js | 23 +- web/src/components/alerts/AlertList.spec.ts | 109 ++ web/src/components/alerts/AlertList.vue | 45 +- .../alerts/AnomalyDetectionChart.spec.ts | 13 + .../alerts/AnomalyDetectionChart.vue | 11 + .../settings/OrganizationSettings.schema.ts | 1 + .../settings/OrganizationSettings.spec.ts | 63 + .../settings/OrganizationSettings.vue | 24 + .../dashboard/usePanelDataLoader.spec.ts | 49 + .../dashboard/usePanelDataLoader.ts | 26 +- .../dashboard/usePanelSQLExecutor.spec.ts | 248 +++- .../dashboard/usePanelSQLExecutor.ts | 531 +++---- .../dashboard/usePanelSearchHandlers.ts | 1 + .../traces/useTraceProcessing.spec.ts | 206 --- .../composables/traces/useTraceProcessing.ts | 143 -- .../composables/useLogs/useSearchBar.spec.ts | 20 +- web/src/composables/useLogs/useSearchBar.ts | 2 +- .../layouts/MainLayout.orgSettings.spec.ts | 112 ++ web/src/layouts/MainLayout.vue | 3 + web/src/locales/languages/ar-SA.json | 28 + web/src/locales/languages/de-DE.json | 28 + web/src/locales/languages/en-US.json | 28 + web/src/locales/languages/es-ES.json | 28 + web/src/locales/languages/fr-FR.json | 28 + web/src/locales/languages/it-IT.json | 28 + web/src/locales/languages/ja-JP.json | 28 + web/src/locales/languages/ko-KR.json | 28 + web/src/locales/languages/nl-NL.json | 28 + web/src/locales/languages/pl-PL.json | 28 + web/src/locales/languages/pt-PT.json | 28 + web/src/locales/languages/ru-RU.json | 28 + web/src/locales/languages/tr-TR.json | 28 + web/src/locales/languages/vi-VN.json | 28 + web/src/locales/languages/zh-CN.json | 28 + web/src/locales/languages/zh-TW.json | 28 + web/src/plugins/logs/SearchBar.vue | 1 + web/src/plugins/profiles/Index.spec.ts | 293 +++- web/src/plugins/profiles/Index.vue | 103 +- web/src/plugins/traces/Index.spec.ts | 362 ++++- web/src/plugins/traces/Index.vue | 87 +- web/src/plugins/traces/SearchBar.spec.ts | 398 +++++- web/src/plugins/traces/SearchBar.vue | 288 +++- web/src/plugins/traces/SearchResult.vue | 36 - .../plugins/traces/ServicesCatalog.spec.ts | 938 +++++++++++- web/src/plugins/traces/ServicesCatalog.vue | 458 ++++-- web/src/plugins/traces/SpanBlock.spec.ts | 99 ++ web/src/plugins/traces/SpanBlock.vue | 39 +- web/src/plugins/traces/TraceDetails.spec.ts | 335 +++++ web/src/plugins/traces/TraceDetails.vue | 167 ++- .../traces/TraceDetailsSidebar.spec.ts | 446 +++++- .../plugins/traces/TraceDetailsSidebar.vue | 311 ++++ web/src/plugins/traces/TraceTree.spec.ts | 9 + web/src/plugins/traces/TraceTree.vue | 5 + .../metrics/TracesAnalysisDashboard.spec.ts | 26 + .../metrics/TracesAnalysisDashboard.vue | 14 +- .../metrics/TracesDrillDownPage.spec.ts | 160 +++ .../traces/metrics/TracesDrillDownPage.vue | 103 ++ .../metrics/TracesMetricsDashboard.spec.ts | 52 +- .../traces/metrics/TracesMetricsDashboard.vue | 48 +- web/src/services/profiles.ts | 33 +- web/src/stores/index.ts | 1 + web/src/ts/interfaces/traces/span.types.ts | 1 - web/src/ts/interfaces/traces/trace.types.ts | 114 +- web/src/utils/alerts/anomalyBandChart.spec.ts | 83 ++ web/src/utils/alerts/anomalyBandChart.ts | 61 +- web/src/utils/traces/convertTraceData.spec.ts | 11 + web/src/utils/traces/convertTraceData.ts | 10 +- web/src/utils/traces/criticalPath.spec.ts | 321 +++++ web/src/utils/traces/criticalPath.ts | 167 +++ web/src/utils/traces/patternDetection.ts | 2 +- .../utils/traces/traceServiceGraph.spec.ts | 240 ++++ web/src/utils/traces/traceServiceGraph.ts | 139 ++ 89 files changed, 9257 insertions(+), 1075 deletions(-) create mode 100644 src/core/src/traces/red_insights.rs create mode 100644 src/jobs/src/job/red_insights.rs create mode 100644 web/src/layouts/MainLayout.orgSettings.spec.ts create mode 100644 web/src/plugins/traces/metrics/TracesDrillDownPage.spec.ts create mode 100644 web/src/plugins/traces/metrics/TracesDrillDownPage.vue create mode 100644 web/src/utils/traces/criticalPath.spec.ts create mode 100644 web/src/utils/traces/criticalPath.ts create mode 100644 web/src/utils/traces/traceServiceGraph.spec.ts create mode 100644 web/src/utils/traces/traceServiceGraph.ts diff --git a/scripts/translations/.translation_state.json b/scripts/translations/.translation_state.json index a804d5ef1e..3c980d6440 100644 --- a/scripts/translations/.translation_state.json +++ b/scripts/translations/.translation_state.json @@ -16628,6 +16628,8 @@ "queryManagement": "e804e3542cae3915", "queryManagementDesc": "f1bc895b2f1da57e", "recommendedByFieldCoverage": "a109c01355662af8", + "redInsightsEnabledHelp": "766c0400c24ccc99", + "redInsightsEnabledLabel": "4a18f94f3f6afd29", "regexPatternList": { "bulkDeleteFailed": "36c7f0d7791212dd", "bulkDeleteMessage": "af717a4ac4d27610", @@ -16741,6 +16743,7 @@ "themeManagementDescription": "5024019f661fc3dd", "toggleIngestionLogsLabel": "d62acc58665cf9bb", "traceIdFieldName": "f1fbe93390eb5290", + "tracesHeading": "998741a58ed2b411", "trialExtensionMinWeek": "e174ca7c2a4417c5", "trial_expiry": "03378fba7bf9c9ae", "updateAiCreditsFailed": "5754421c67124875", @@ -19579,6 +19582,7 @@ "applyAndSearch": "c7de18087c748cfb", "applyPendingFilters": "ce1202e0d05abe3b", "backToLogs": "ed627b7ddd4e0cca", + "backToResults": "c7ef0e278b1e08c0", "backToTraces": "f03da00bbeb8113f", "clearErrorFilter": "b634835f3e3a5324", "collapseFields": "166bd3007d204cc4", @@ -19608,6 +19612,7 @@ }, "correlationFailed": "47638dd4bdeaee84", "correlationServiceNotFound": "77ea7bc4d298d205", + "criticalPath": "bfbcf6ab87bc0dc6", "dbSpanDetails": { "batchSize": "c4857cd1742f94da", "database": "fa7fe67124e94375", @@ -19624,6 +19629,7 @@ "user": "b512d97e7cbf97c2", "viewQueryTrend": "df5dc6da572cfc5a" }, + "drillDown": "682c25289fd57604", "errLabel": "af440ad05977eb32", "error": "d98ee0e5f9399db9", "errorCount": "6b8c59fe228558f0", @@ -19796,7 +19802,9 @@ "operation": "0f044feb6ca7d676", "p50Median": "76661f0c8ac6ce54", "previousMatch": "daa2f8c3fc897cc7", + "redMetrics": "a9d1ab9a2d6d4dac", "reviewAndApplyFilters": "28a6fd8929f1ffa5", + "savedViewStreamMissing": "d9d331cfdc12f282", "searchBar": { "streamNotFound": "b46fc36172de7113" }, @@ -19921,10 +19929,22 @@ "errors": "cb702378f31507ef", "maxDuration": "e5b10d63b3485042", "requests": "ada27592c9578e90", + "requestsTooltip": "e5d6ed86bca68fdd", "serviceName": "d677190e0a9990e7", "status": "920e413c7d411b61" }, "filterPlaceholder": "6c403ddc8a77c64c", + "insights": { + "scoreAbove": "6015543c308e3576", + "signals": { + "errors": "e4fe6202728fa010", + "p95": "a81f378f0a8b5949", + "rate": "9535754daaf07fae" + }, + "title": "2a932f90172e99ea", + "viewCharts": "f91cf6c6263dc004", + "viewTraces": "998741a58ed2b411" + }, "legend": { "title": "bf7d539c44f17179" }, @@ -20171,6 +20191,13 @@ "modelParameters": "166cfbd5b6effb00", "noDataAvailable": "d2d2d48c815b0539", "notScoredYet": "a83f5a22f4cccac3", + "operationBandAboveP99": "f9bf10b75fbbdee7", + "operationBandBelowP50": "39d15e603495c20f", + "operationBandP50ToP75": "24c88939d8a0ecb1", + "operationBandP75ToP90": "1a83ac65f5b53949", + "operationBandP90ToP99": "832d86d902fc75cf", + "operationPercentileBand": "8367057943fd3f77", + "operationPercentilesTooltip": "3ccec31a0be1b901", "out": "9a3c5460432ba66a", "output": "b2439bcb8dee14b6", "outputTokens": "a0a1074faa44dfa0", @@ -20231,6 +20258,7 @@ "viewFilters": "6fae22556dc2d946", "viewLogs": "e087f046dc3cda9a", "viewLogsUnavailable": "e2468fc4f70cc910", + "viewProfile": "d4788f256f730423", "waterfall": "6b53bc225e0aae1d" }, "user": { diff --git a/src/api/management/src/request/organization/settings.rs b/src/api/management/src/request/organization/settings.rs index 5ae36f1997..4f46978aca 100644 --- a/src/api/management/src/request/organization/settings.rs +++ b/src/api/management/src/request/organization/settings.rs @@ -135,6 +135,10 @@ pub async fn create( data.usage_stream_enabled = usage_stream_enabled; } + if merge_red_insights_enabled(&mut data, settings.red_insights_enabled) { + field_found = true; + } + if let Some(cross_links) = settings.cross_links { for link in &cross_links { if link.name.is_empty() { @@ -365,8 +369,31 @@ pub async fn delete_logo_text() -> Response { (StatusCode::FORBIDDEN, Json("Not Supported")).into_response() } +fn merge_red_insights_enabled(data: &mut OrganizationSetting, value: Option) -> bool { + let Some(enabled) = value else { + return false; + }; + data.red_insights_enabled = enabled; + true +} + #[cfg(test)] mod tests { + use super::*; + + #[test] + fn test_merge_red_insights_enabled_sets_and_leaves() { + let mut data = OrganizationSetting::default(); + assert!(!merge_red_insights_enabled(&mut data, None)); + assert!(!data.red_insights_enabled); + assert!(merge_red_insights_enabled(&mut data, Some(true))); + assert!(data.red_insights_enabled); + assert!(!merge_red_insights_enabled(&mut data, None)); + assert!(data.red_insights_enabled); + assert!(merge_red_insights_enabled(&mut data, Some(false))); + assert!(!data.red_insights_enabled); + } + #[test] fn test_max_series_per_query_validation_valid_values() { // Test minimum valid value diff --git a/src/api/search/src/profiles/meta.rs b/src/api/search/src/profiles/meta.rs index 78fd50ee5f..6ef234caa8 100644 --- a/src/api/search/src/profiles/meta.rs +++ b/src/api/search/src/profiles/meta.rs @@ -108,17 +108,27 @@ pub async fn get_profiles_meta( timeout, }; + let [services_trace_id, types_trace_id] = sub_query_trace_ids(&request_trace_id); + let services_params = ProfilesSqlParams { + request_trace_id: &services_trace_id, + ..sql_params + }; + let types_params = ProfilesSqlParams { + request_trace_id: &types_trace_id, + ..sql_params + }; + // Only DISTINCT columns present in schema — async-profiler data often has no service_name. let services_fut = async { if schema_fields.iter().any(|f| f == "service_name") { - distinct_values(&sql_params, &stream, "service_name", http_span.clone()).await + distinct_values(&services_params, &stream, "service_name", http_span.clone()).await } else { Ok(Vec::new()) } }; let profile_types_fut = async { if schema_fields.iter().any(|f| f == "profile_type") { - load_profile_types(&sql_params, &stream, http_span.clone()).await + load_profile_types(&types_params, &stream, http_span.clone()).await } else { Ok(Vec::new()) } @@ -144,6 +154,14 @@ pub async fn get_profiles_meta( .into_response() } +/// Concurrent searches need their own ids: the query manager keys running searches by trace_id. +fn sub_query_trace_ids(request_trace_id: &str) -> [String; 2] { + [ + format!("{request_trace_id}-services"), + format!("{request_trace_id}-types"), + ] +} + async fn distinct_values( params: &ProfilesSqlParams<'_>, stream: &str, @@ -208,3 +226,16 @@ async fn load_profile_types( } Ok(out) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sub_query_trace_ids_are_distinct_and_keep_the_request_prefix() { + let [services, types] = sub_query_trace_ids("req123"); + assert_ne!(services, types); + assert!(services.starts_with("req123"), "{services}"); + assert!(types.starts_with("req123"), "{types}"); + } +} diff --git a/src/api/search/src/search/saved_view.rs b/src/api/search/src/search/saved_view.rs index 8c55da0e56..453a6ac495 100644 --- a/src/api/search/src/search/saved_view.rs +++ b/src/api/search/src/search/saved_view.rs @@ -251,18 +251,65 @@ mod tests { Router, body::Body, http::{Request, StatusCode}, - routing::post, + routing::{post, put}, }; + use serde_json::{Value, json}; use tower::ServiceExt; use super::*; + fn app() -> Router { + Router::new() + .route("/{org_id}/savedviews", post(create_view).get(get_views)) + .route( + "/{org_id}/savedviews/{view_id}", + put(update_view).get(get_view), + ) + } + + async fn call(method: &str, uri: &str, body: Option) -> (StatusCode, Value) { + let req = Request::builder() + .method(method) + .uri(uri) + .header("content-type", "application/json") + .body(body.map_or_else(Body::empty, |b| Body::from(b.to_string()))) + .unwrap(); + let resp = app().oneshot(req).await.unwrap(); + let status = resp.status(); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + ( + status, + serde_json::from_slice(&bytes).unwrap_or(Value::Null), + ) + } + + async fn create(org: &str, name: &str, view_type: Option<&str>) -> (StatusCode, Value) { + let mut body = json!({"data": {"v": 1}, "view_name": name}); + if let Some(t) = view_type { + body["view_type"] = json!(t); + } + call("POST", &format!("/{org}/savedviews"), Some(body)).await + } + + async fn list(org: &str) -> Vec { + let (status, body) = call("GET", &format!("/{org}/savedviews"), None).await; + assert_eq!(status, StatusCode::OK); + body["views"].as_array().cloned().unwrap_or_default() + } + + fn new_org() -> String { + format!("svorg{}", config::ider::uuid()).to_lowercase() + } + #[tokio::test] async fn test_create_view_post() { infra::db::create_table().await.unwrap(); let payload = CreateViewRequest { data: "base64-encoded-data".into(), view_name: format!("query-for-blah-{}", config::ider::uuid()), + view_type: None, }; let app = Router::new().route("/{org_id}/savedviews", post(create_view)); let req = Request::builder() @@ -274,4 +321,79 @@ mod tests { let resp = app.oneshot(req).await.unwrap(); assert!(resp.status() == StatusCode::OK || resp.status().is_success()); } + + #[tokio::test] + async fn test_traces_view_type_is_stored_listed_and_kept_on_update() { + infra::db::create_table().await.unwrap(); + let org = new_org(); + let (status, created) = create(&org, "checkout errors", Some("traces")).await; + assert_eq!(status, StatusCode::OK); + let view_id = created["view_id"].as_str().unwrap().to_string(); + + let views = list(&org).await; + assert_eq!(views.len(), 1); + assert_eq!(views[0]["view_type"], "traces"); + + let (status, updated) = call( + "PUT", + &format!("/{org}/savedviews/{view_id}"), + Some(json!({"data": {"v": 2}, "view_name": "checkout errors", "view_type": "logs"})), + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!(updated["view_type"], "traces"); + assert_eq!(list(&org).await[0]["view_type"], "traces"); + } + + #[tokio::test] + async fn test_create_view_rejects_unknown_type() { + infra::db::create_table().await.unwrap(); + let org = new_org(); + let (status, _) = create(&org, "bad", Some("metrics")).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert!(list(&org).await.is_empty()); + } + + #[tokio::test] + async fn test_view_names_are_unique_per_type() { + infra::db::create_table().await.unwrap(); + let org = new_org(); + assert_eq!(create(&org, "dup", None).await.0, StatusCode::OK); + assert_eq!(create(&org, "DUP", Some("traces")).await.0, StatusCode::OK); + assert_eq!( + create(&org, "Dup", Some("traces")).await.0, + StatusCode::BAD_REQUEST + ); + assert_eq!( + create(&org, "dup", Some("logs")).await.0, + StatusCode::BAD_REQUEST + ); + } + + #[tokio::test] + async fn test_update_traces_view_to_a_logs_view_name_succeeds() { + infra::db::create_table().await.unwrap(); + let org = new_org(); + assert_eq!(create(&org, "alpha", None).await.0, StatusCode::OK); + let (status, created) = create(&org, "beta", Some("traces")).await; + assert_eq!(status, StatusCode::OK); + let view_id = created["view_id"].as_str().unwrap(); + let (status, _) = call( + "PUT", + &format!("/{org}/savedviews/{view_id}"), + Some(json!({"data": {}, "view_name": "alpha"})), + ) + .await; + assert_eq!(status, StatusCode::OK); + } + + #[tokio::test] + async fn test_an_org_sharing_a_name_prefix_keeps_its_views_separate() { + infra::db::create_table().await.unwrap(); + let org = new_org(); + let longer_org = format!("{org}b"); + assert_eq!(create(&longer_org, "shared", None).await.0, StatusCode::OK); + assert!(list(&org).await.is_empty()); + assert_eq!(create(&org, "shared", None).await.0, StatusCode::OK); + } } diff --git a/src/common/src/meta/organization.rs b/src/common/src/meta/organization.rs index de5c7ba896..3d492cccf5 100644 --- a/src/common/src/meta/organization.rs +++ b/src/common/src/meta/organization.rs @@ -506,6 +506,8 @@ pub struct OrganizationSettingPayload { pub max_series_per_query: Option, #[serde(skip_serializing_if = "Option::is_none")] pub usage_stream_enabled: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub red_insights_enabled: Option, #[cfg(feature = "enterprise")] #[serde(skip_serializing_if = "Option::is_none")] pub claim_parser_function: Option, @@ -545,6 +547,8 @@ pub struct OrganizationSetting { pub max_series_per_query: Option, #[serde(default = "default_usage_stream_enabled")] pub usage_stream_enabled: bool, + #[serde(default)] + pub red_insights_enabled: bool, #[cfg(feature = "enterprise")] #[serde(default = "default_claim_parser_function")] pub claim_parser_function: String, @@ -584,6 +588,7 @@ impl Default for OrganizationSetting { dark_mode_theme_color, max_series_per_query: None, usage_stream_enabled: default_usage_stream_enabled(), + red_insights_enabled: false, #[cfg(feature = "enterprise")] claim_parser_function: default_claim_parser_function(), cross_links: Vec::new(), @@ -1456,6 +1461,7 @@ mod tests { dark_mode_theme_color: None, max_series_per_query: None, usage_stream_enabled: None, + red_insights_enabled: None, #[cfg(feature = "enterprise")] claim_parser_function: None, cross_links: None, @@ -1474,6 +1480,7 @@ mod tests { assert!(!obj.contains_key("dark_mode_theme_color")); assert!(!obj.contains_key("max_series_per_query")); assert!(!obj.contains_key("usage_stream_enabled")); + assert!(!obj.contains_key("red_insights_enabled")); assert!(!obj.contains_key("cross_links")); } @@ -1532,6 +1539,7 @@ mod tests { dark_mode_theme_color: None, max_series_per_query: None, usage_stream_enabled: false, + red_insights_enabled: false, #[cfg(feature = "enterprise")] claim_parser_function: String::new(), cross_links: vec![], @@ -1562,6 +1570,7 @@ mod tests { dark_mode_theme_color: Some("#000".to_string()), max_series_per_query: Some(1000), usage_stream_enabled: false, + red_insights_enabled: false, #[cfg(feature = "enterprise")] claim_parser_function: String::new(), cross_links: vec![], @@ -1638,4 +1647,19 @@ mod tests { let parsed: OrganizationSetting = serde_json::from_str(json_false).unwrap(); assert!(!parsed.usage_stream_enabled); } + + #[test] + fn test_red_insights_enabled_defaults_off_and_round_trips() { + assert!(!OrganizationSetting::default().red_insights_enabled); + let legacy: OrganizationSetting = + serde_json::from_str(r#"{"scrape_interval": 15}"#).unwrap(); + assert!(!legacy.red_insights_enabled); + let on: OrganizationSetting = + serde_json::from_str(r#"{"scrape_interval": 15, "red_insights_enabled": true}"#) + .unwrap(); + assert!(on.red_insights_enabled); + let payload: OrganizationSettingPayload = + serde_json::from_str(r#"{"red_insights_enabled": true}"#).unwrap(); + assert_eq!(payload.red_insights_enabled, Some(true)); + } } diff --git a/src/common/src/meta/saved_view.rs b/src/common/src/meta/saved_view.rs index 0aabb745dc..fea80456c6 100644 --- a/src/common/src/meta/saved_view.rs +++ b/src/common/src/meta/saved_view.rs @@ -25,6 +25,10 @@ pub struct CreateViewRequest { /// User-readable name of the view, must be unique within the organization. pub view_name: String, + + /// View kind, `logs` or `traces`; absent means `logs`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub view_type: Option, } #[derive(Serialize, Deserialize, ToSchema)] @@ -44,6 +48,8 @@ pub struct View { pub data: serde_json::Value, pub view_id: String, pub view_name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub view_type: Option, } /// Save the bandwidth for a given view, without sending the actual data @@ -53,6 +59,8 @@ pub struct ViewWithoutData { pub org_id: String, pub view_id: String, pub view_name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub view_type: Option, } #[derive(Serialize, Deserialize, ToSchema)] @@ -86,6 +94,7 @@ mod tests { let request = CreateViewRequest { data: serde_json::json!({"key": "value", "number": 42}), view_name: "test-view".to_string(), + view_type: None, }; let serialized = serde_json::to_string(&request); @@ -134,6 +143,7 @@ mod tests { data: serde_json::json!({"dashboard": "main"}), view_id: "view456".to_string(), view_name: "main-dashboard".to_string(), + view_type: None, }; let serialized = serde_json::to_string(&view); @@ -159,6 +169,7 @@ mod tests { org_id: "org123".to_string(), view_id: "view456".to_string(), view_name: "main-dashboard".to_string(), + view_type: None, }; let serialized = serde_json::to_string(&view); @@ -185,11 +196,13 @@ mod tests { org_id: "org123".to_string(), view_id: "view1".to_string(), view_name: "dashboard1".to_string(), + view_type: None, }, ViewWithoutData { org_id: "org123".to_string(), view_id: "view2".to_string(), view_name: "dashboard2".to_string(), + view_type: None, }, ], }; @@ -273,4 +286,50 @@ mod tests { assert_eq!(deserialized.view_id, "view456"); assert_eq!(deserialized.view_name, "new-dashboard"); } + + #[test] + fn test_view_type_round_trips_when_set() { + let request = CreateViewRequest { + data: serde_json::json!({}), + view_name: "v".to_string(), + view_type: Some("traces".to_string()), + }; + let json = serde_json::to_value(&request).unwrap(); + assert_eq!(json["view_type"], "traces"); + let back: CreateViewRequest = serde_json::from_value(json).unwrap(); + assert_eq!(back.view_type.as_deref(), Some("traces")); + + let view = View { + org_id: "o".to_string(), + data: serde_json::json!({}), + view_id: "id".to_string(), + view_name: "v".to_string(), + view_type: Some("traces".to_string()), + }; + let back: ViewWithoutData = + serde_json::from_str(&serde_json::to_string(&view).unwrap()).unwrap(); + assert_eq!(back.view_type.as_deref(), Some("traces")); + } + + #[test] + fn test_view_type_omitted_when_unset() { + let view = ViewWithoutData { + org_id: "o".to_string(), + view_id: "id".to_string(), + view_name: "v".to_string(), + view_type: None, + }; + let json = serde_json::to_value(&view).unwrap(); + assert!(json.get("view_type").is_none()); + } + + #[test] + fn test_legacy_json_without_view_type_deserialises_as_none() { + let legacy = r#"{"org_id":"o","data":{},"view_id":"id","view_name":"v"}"#; + let view: View = serde_json::from_str(legacy).unwrap(); + assert!(view.view_type.is_none()); + let request: CreateViewRequest = + serde_json::from_str(r#"{"data":{},"view_name":"v"}"#).unwrap(); + assert!(request.view_type.is_none()); + } } diff --git a/src/core/src/anomaly_detection.rs b/src/core/src/anomaly_detection.rs index b48fdd1e1e..8e136d2516 100644 --- a/src/core/src/anomaly_detection.rs +++ b/src/core/src/anomaly_detection.rs @@ -6483,4 +6483,60 @@ mod tests { // An untagged request yields no role group at all, which is the bug this pins. assert_ne!(RoleGroup::from(SearchEventType::UI), RoleGroup::Background); } + + #[test] + fn red_insights_templates_pass_create_validation() { + use crate::traces::red_insights::{RedSignal, StreamColumns, template}; + + for has_parent in [true, false] { + let cols = StreamColumns { + has_parent, + has_status: true, + has_duration: true, + }; + for (stream, service) in [ + ("default", "checkout"), + ("a\"b", "o'brien"), + ("t/x", "api/v1"), + ] { + for signal in RedSignal::ALL { + let req = template(stream, service, signal, &cols) + .expect("every signal is eligible here") + .into_request("default"); + validate_config_request(&req) + .unwrap_or_else(|e| panic!("{signal:?} on {stream}/{service}: {e}")); + config::meta::alerts::tags::normalize_tags(&req.tags).unwrap(); + } + } + } + } + + #[test] + fn red_insights_skips_exactly_the_rate_detectors_g4_rejects() { + use crate::traces::red_insights::{RedSignal, StreamColumns, template}; + + let cols = StreamColumns { + has_parent: true, + has_status: true, + has_duration: true, + }; + for word in ERROR_VOCABULARY { + let stream = format!("app-{word}"); + assert!( + template(&stream, "svc", RedSignal::Rate, &cols).is_none(), + "{stream}" + ); + let mut req = template("default", "svc", RedSignal::Rate, &cols) + .unwrap() + .into_request("default"); + req.stream_name = stream.clone(); + assert!(validate_config_request(&req).is_err(), "{stream}"); + for signal in [RedSignal::ErrorRatio, RedSignal::P95Latency] { + let req = template(&stream, "svc", signal, &cols) + .unwrap() + .into_request("default"); + validate_config_request(&req).unwrap_or_else(|e| panic!("{signal:?}: {e}")); + } + } + } } diff --git a/src/core/src/ingestion/mod.rs b/src/core/src/ingestion/mod.rs index d539c47e99..78ddfa842c 100644 --- a/src/core/src/ingestion/mod.rs +++ b/src/core/src/ingestion/mod.rs @@ -33,10 +33,11 @@ use config::{ promql::HASH_LABEL, self_reporting::usage::{RequestStats, RunOutcome, TriggerData, TriggerDataType}, stream::{PartitionTimeLevel, StreamParams, StreamPartition, StreamType}, + triggers::ScheduledTriggerData, }, utils::{ flatten, - json::*, + json::{self, *}, schema::format_partition_key, time::{DAY_MICRO_SECS, HOUR_MICRO_SECS}, }, @@ -268,15 +269,7 @@ pub async fn evaluate_trigger(triggers: TriggerAlertData) { // After the notification is sent successfully, we need to update // the silence period of the trigger if let Err(e) = db::scheduler::update_trigger( - db::scheduler::Trigger { - org: alert.org_id.to_string(), - module: db::scheduler::TriggerModule::Alert, - module_key, - is_silenced: true, - is_realtime: true, - next_run_at, - ..Default::default() - }, + silenced_realtime_trigger(&alert.org_id, module_key, next_run_at, now), false, "", ) @@ -813,6 +806,29 @@ pub fn refactor_map( new_map } +fn silenced_realtime_trigger( + org: &str, + module_key: String, + next_run_at: i64, + fired_at: i64, +) -> db::scheduler::Trigger { + db::scheduler::Trigger { + org: org.to_string(), + module: db::scheduler::TriggerModule::Alert, + module_key, + is_silenced: true, + is_realtime: true, + next_run_at, + // Realtime rows carry no other data, and the wakeup clones it forward. + data: json::to_string(&ScheduledTriggerData { + last_satisfied_at: Some(fired_at), + ..Default::default() + }) + .unwrap_or_default(), + ..Default::default() + } +} + /// The span of one write partition, which a record's time bucket is counted in. fn partition_bucket_micros(time_level: PartitionTimeLevel) -> i64 { match time_level { @@ -830,6 +846,15 @@ mod tests { use super::*; + #[test] + fn test_silenced_realtime_trigger_records_when_it_fired() { + let trigger = silenced_realtime_trigger("org1", "key1".to_string(), 2_000, 1_000); + assert!(trigger.is_silenced && trigger.is_realtime); + assert_eq!(trigger.next_run_at, 2_000); + let data: ScheduledTriggerData = json::from_str(&trigger.data).unwrap(); + assert_eq!(data.last_satisfied_at, Some(1_000)); + } + #[test] fn test_format_partition_key() { assert_eq!(format_partition_key("default/olympics"), "defaultolympics"); diff --git a/src/core/src/traces/mod.rs b/src/core/src/traces/mod.rs index 867d926fab..d14cad1018 100644 --- a/src/core/src/traces/mod.rs +++ b/src/core/src/traces/mod.rs @@ -56,6 +56,7 @@ use serde_json::Map; pub mod agent_signals; pub mod inferred; pub mod otel; +pub mod red_insights; pub mod service_graph; pub mod session; #[cfg(test)] diff --git a/src/core/src/traces/red_insights.rs b/src/core/src/traces/red_insights.rs new file mode 100644 index 0000000000..bd4539e2b7 --- /dev/null +++ b/src/core/src/traces/red_insights.rs @@ -0,0 +1,1258 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::{ + collections::{BTreeMap, HashMap, HashSet}, + sync::LazyLock, +}; + +use config::utils::sql::{quote_identifier, quote_sql_string}; +use regex::Regex; + +pub const MANAGED_TAG: &str = "auto:red-insights"; +pub const FOLDER_NAME: &str = "RED insights"; +pub const MAX_SERVICES: usize = 20; +pub const MIN_REQUESTS_24H: u64 = 10_000; +pub const MAX_CREATES: usize = 15; + +const HISTOGRAM_INTERVAL: &str = "5m"; +const SCHEDULE_INTERVAL: &str = "1h"; +const DETECTION_WINDOW_SECONDS: i64 = 3_900; +const MIN_BUCKET_REQUESTS: u32 = 20; +const NAME_SEPARATOR: &str = " · "; +/// Alert names forbid `/`; U+2215 reads the same and the web strip maps it back. +const NAME_SLASH: char = '\u{2215}'; +/// The enterprise custom-SQL check rejects these as substrings anywhere in the query. +const DESTRUCTIVE_KEYWORDS: [&str; 4] = ["update", "delete", "drop", "insert"]; +/// Mirrors G4's `ERROR_VOCABULARY` in anomaly_detection.rs; the gated validator test pins it. +const ERROR_VOCABULARY: [&str; 6] = ["error", "errors", "fatal", "critical", "5xx", "50x"]; +const KIND_PREDICATE: &str = "CAST(span_kind AS VARCHAR) IN ('2','5')"; +/// Any root span is a request, whatever its kind: a trace starting at a client span entered there. +const ROOT_ARM: &str = "OR (reference_parent_span_id IS NULL OR reference_parent_span_id = '')"; + +/// Tolerates reformatting, not rewrites: a condition before `service_name` stays unrecognised. +static IDENTITY: LazyLock = LazyLock::new(|| { + Regex::new(r#"(?i)\bFROM\s+("(?:[^"]|"")*")\s+WHERE\s+service_name\s*=\s*'"#).unwrap() +}); + +/// A RED signal; each managed service gets one detector per signal. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub enum RedSignal { + Rate, + ErrorRatio, + P95Latency, +} + +impl RedSignal { + pub const ALL: [RedSignal; 3] = [ + RedSignal::Rate, + RedSignal::ErrorRatio, + RedSignal::P95Latency, + ]; + + pub fn tag(self) -> &'static str { + match self { + RedSignal::Rate => "red:rate", + RedSignal::ErrorRatio => "red:errors", + RedSignal::P95Latency => "red:p95", + } + } + + pub fn label(self) -> &'static str { + match self { + RedSignal::Rate => "rate", + RedSignal::ErrorRatio => "errors", + RedSignal::P95Latency => "p95", + } + } + + fn from_tag(tag: &str) -> Option { + Self::ALL.into_iter().find(|s| s.tag() == tag) + } + + fn value_expr(self) -> &'static str { + match self { + RedSignal::Rate => "COUNT(*)", + RedSignal::ErrorRatio => { + "CAST(SUM(CASE WHEN span_status = 'ERROR' THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*)" + } + RedSignal::P95Latency => "approx_percentile_cont(duration, 0.95)", + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ServiceVolume { + pub stream: String, + pub service: String, + pub requests_24h: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ManagedDetector { + pub id: String, + pub stream: String, + pub service: String, + pub signal: RedSignal, + pub enabled: bool, + /// SQL equal to today's template; a non-current detector is deleted only by setting-off. + pub current: bool, +} + +#[derive(Debug, Default, PartialEq, Eq)] +pub struct Plan { + pub create: Vec<(String, String, RedSignal)>, + pub delete: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StreamColumns { + pub has_parent: bool, + pub has_status: bool, + pub has_duration: bool, +} + +/// What this cycle learned about a stream; a stream absent from the map counts as unread. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StreamRead { + Read(StreamColumns), + /// Its schema or volume read failed, so its detectors are left alone this cycle. + Unread, + /// Removed, or readable but unrankable, so its detectors are deleted. + Retired, +} + +/// A detector's create-request fields, engine-free so the planner is testable without enterprise. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DetectorTemplate { + pub name: String, + pub stream_name: String, + pub stream_type: String, + pub query_mode: String, + pub custom_sql: String, + pub detection_function: String, + pub detection_function_field: Option, + pub histogram_interval: String, + pub schedule_interval: String, + pub detection_window_seconds: i64, + pub alert_enabled: bool, + pub alert_direction: String, + pub tags: Vec, +} + +#[cfg(feature = "enterprise")] +impl DetectorTemplate { + pub fn into_request( + self, + folder_id: &str, + ) -> crate::anomaly_detection::CreateAnomalyConfigRequest { + crate::anomaly_detection::CreateAnomalyConfigRequest { + name: self.name, + description: None, + stream_name: self.stream_name, + stream_type: self.stream_type, + query_mode: self.query_mode, + filters: None, + custom_sql: Some(self.custom_sql), + detection_function: self.detection_function, + detection_function_field: self.detection_function_field, + histogram_interval: self.histogram_interval, + schedule_interval: self.schedule_interval, + detection_window_seconds: self.detection_window_seconds, + training_window_days: None, + retrain_interval_days: None, + percentile: None, + alert_budget_per_day: None, + level_half_width_seconds: None, + rcf_num_trees: None, + rcf_tree_size: None, + rcf_shingle_size: None, + band_width: None, + alert_direction: Some(self.alert_direction), + alert_window_buckets: None, + alert_window_fire_pct: None, + alert_window_recover_pct: None, + alert_enabled: Some(self.alert_enabled), + alert_destinations: vec![], + enabled: None, + folder_id: Some(folder_id.to_string()), + owner: None, + priority: None, + tags: self.tags, + } + } +} + +/// Off deletes all; on keeps a 2x hysteresis band and fills unoccupied triples up to `max_creates`. +pub fn plan( + enabled: bool, + volumes: &[ServiceVolume], + existing: &[ManagedDetector], + max_services: usize, + min_requests_24h: u64, + max_creates: usize, +) -> Plan { + if !enabled { + let mut delete: Vec = existing.iter().map(|d| d.id.clone()).collect(); + delete.sort(); + return Plan { + create: vec![], + delete, + }; + } + + let mut ranked: Vec<&ServiceVolume> = volumes.iter().collect(); + ranked.sort_by(|a, b| { + b.requests_24h + .cmp(&a.requests_24h) + .then_with(|| a.stream.cmp(&b.stream)) + .then_with(|| a.service.cmp(&b.service)) + }); + let kept: HashMap<(&str, &str), bool> = ranked + .iter() + .take(max_services.saturating_mul(2)) + .enumerate() + .filter(|(_, v)| v.requests_24h >= min_requests_24h / 2) + .map(|(rank, v)| { + let selected = rank < max_services && v.requests_24h >= min_requests_24h; + ((v.stream.as_str(), v.service.as_str()), selected) + }) + .collect(); + + // Older-template or user-edited SQL keeps its triple: never duplicated, never overwritten. + let occupied: HashSet<(&str, &str, RedSignal)> = existing + .iter() + .map(|d| (d.stream.as_str(), d.service.as_str(), d.signal)) + .collect(); + let mut by_triple: BTreeMap<(&str, &str, RedSignal), Vec<&ManagedDetector>> = BTreeMap::new(); + for d in existing.iter().filter(|d| d.current) { + by_triple + .entry((d.stream.as_str(), d.service.as_str(), d.signal)) + .or_default() + .push(d); + } + + let mut delete = Vec::new(); + for (&(stream, service, _), group) in &by_triple { + if kept.contains_key(&(stream, service)) { + delete.extend(duplicates(group)); + } else { + delete.extend(group.iter().map(|d| d.id.clone())); + } + } + delete.sort(); + + let create = ranked + .iter() + .filter(|v| kept.get(&(v.stream.as_str(), v.service.as_str())) == Some(&true)) + .flat_map(|v| RedSignal::ALL.map(|s| (v, s))) + .filter(|(v, s)| !occupied.contains(&(v.stream.as_str(), v.service.as_str(), *s))) + .take(max_creates) + .map(|(v, s)| (v.stream.clone(), v.service.clone(), s)) + .collect(); + + Plan { create, delete } +} + +/// Only a fully read or retired stream's detectors enter the plan, which may delete them. +pub fn on_read_streams( + existing: Vec, + reads: &BTreeMap, +) -> Vec { + existing + .into_iter() + .filter(|d| { + matches!( + reads.get(&d.stream), + Some(StreamRead::Read(_) | StreamRead::Retired) + ) + }) + .collect() +} + +/// Classifies a stream whose schema read succeeded, `has` testing for a field. +pub fn read_stream(has: impl Fn(&str) -> bool) -> StreamRead { + if !has("service_name") || !has("span_kind") { + return StreamRead::Retired; + } + StreamRead::Read(StreamColumns { + has_parent: has("reference_parent_span_id"), + has_status: has("span_status"), + has_duration: has("duration"), + }) +} + +/// `None` when the stream lacks the signal's column or the SQL would fail create validation. +pub fn template( + stream: &str, + service: &str, + signal: RedSignal, + cols: &StreamColumns, +) -> Option { + let skip = |reason: &str| { + log::info!( + "[RED insights] skipping {} for {stream}/{service}: {reason}", + signal.label() + ); + None + }; + match signal { + RedSignal::ErrorRatio if !cols.has_status => return skip("no span_status column"), + RedSignal::P95Latency if !cols.has_duration => return skip("no duration column"), + RedSignal::Rate if stream_name_is_error_restricted(stream) => { + return skip("a count over an error-named stream fails the denominator rule"); + } + _ => {} + } + let custom_sql = detector_sql( + "e_identifier(stream), + "e_sql_string(service), + signal, + cols.has_parent, + ); + let lowered = custom_sql.to_lowercase(); + if DESTRUCTIVE_KEYWORDS.iter().any(|k| lowered.contains(k)) { + return skip("the query would trip the destructive-keyword check"); + } + let (detection_function, detection_function_field, alert_direction) = match signal { + RedSignal::Rate => ("count", None, "both"), + RedSignal::ErrorRatio => ("count", None, "above"), + RedSignal::P95Latency => ("p95", Some("value".to_string()), "above"), + }; + Some(DetectorTemplate { + name: managed_name(stream, service, signal), + stream_name: stream.to_string(), + stream_type: "traces".to_string(), + query_mode: "custom_sql".to_string(), + custom_sql, + detection_function: detection_function.to_string(), + detection_function_field, + histogram_interval: HISTOGRAM_INTERVAL.to_string(), + schedule_interval: SCHEDULE_INTERVAL.to_string(), + detection_window_seconds: DETECTION_WINDOW_SECONDS, + alert_enabled: false, + alert_direction: alert_direction.to_string(), + tags: vec![MANAGED_TAG.to_string(), signal.tag().to_string()], + }) +} + +/// Recovers `(service, signal, current)` from the tag and the SQL, never the editable name. +pub fn parse_managed( + tags: &[String], + custom_sql: &str, + stream_name: &str, +) -> Option<(String, RedSignal, bool)> { + let signal = tags.iter().find_map(|t| RedSignal::from_tag(t))?; + let caps = IDENTITY.captures(custom_sql)?; + let quoted_stream = quote_identifier(stream_name); + if caps[1] != quoted_stream { + return None; + } + let service = leading_sql_literal(&custom_sql[caps.get(0)?.end()..])?; + let quoted_service = quote_sql_string(&service); + let current = [true, false].into_iter().any(|has_parent| { + detector_sql("ed_stream, "ed_service, signal, has_parent) == custom_sql + }); + Some((service, signal, current)) +} + +/// Per-service request counts on one trace stream; the caller sets the 24h window. +pub fn volume_sql(stream: &str, cols: &StreamColumns) -> String { + format!( + "SELECT service_name, COUNT(*) FILTER (WHERE {}) AS requests FROM {} \ + GROUP BY service_name ORDER BY requests DESC LIMIT {}", + request_predicate(cols.has_parent), + quote_identifier(stream), + MAX_SERVICES * 2 + ) +} + +/// Shared with the catalog's requestPredicate; unlike the service graph, roots of any kind count. +fn request_predicate(has_parent: bool) -> String { + if has_parent { + format!("{KIND_PREDICATE} {ROOT_ARM}") + } else { + KIND_PREDICATE.to_string() + } +} + +/// Changing this template leaves stored detectors on their old SQL until a migration is added. +fn detector_sql( + quoted_stream: &str, + quoted_service: &str, + signal: RedSignal, + has_parent: bool, +) -> String { + let having = match signal { + RedSignal::Rate => String::new(), + RedSignal::ErrorRatio | RedSignal::P95Latency => { + format!(" HAVING COUNT(*) >= {MIN_BUCKET_REQUESTS}") + } + }; + format!( + "SELECT histogram(_timestamp, '{HISTOGRAM_INTERVAL}') AS time_bucket, {} AS value \ + FROM {quoted_stream} WHERE service_name = {quoted_service} AND ({}) \ + GROUP BY time_bucket{having}", + signal.value_expr(), + request_predicate(has_parent), + ) +} + +fn managed_name(stream: &str, service: &str, signal: RedSignal) -> String { + [ + format!("RED {}", signal.label()), + stream.to_string(), + service.to_string(), + ] + .join(NAME_SEPARATOR) + .replace('/', &NAME_SLASH.to_string()) +} + +/// Disabled detectors are the user's opt-out, so only enabled ones are ever deduplicated away. +fn duplicates(group: &[&ManagedDetector]) -> Vec { + let mut sorted = group.to_vec(); + // KSUIDs sort by creation time, so the smallest id is the oldest. + sorted.sort_by(|a, b| a.id.cmp(&b.id)); + let any_disabled = sorted.iter().any(|d| !d.enabled); + sorted + .iter() + .enumerate() + .filter(|(i, d)| d.enabled && (any_disabled || *i > 0)) + .map(|(_, d)| d.id.clone()) + .collect() +} + +/// Reads a single-quoted SQL literal at the start of `s` (opening quote already consumed). +fn leading_sql_literal(s: &str) -> Option { + let mut out = String::new(); + let mut chars = s.chars().peekable(); + while let Some(c) = chars.next() { + if c != '\'' { + out.push(c); + continue; + } + if chars.peek() == Some(&'\'') { + chars.next(); + out.push('\''); + continue; + } + return Some(out); + } + None +} + +fn stream_name_is_error_restricted(stream: &str) -> bool { + stream + .split(|c: char| !c.is_ascii_alphanumeric()) + .any(|token| ERROR_VOCABULARY.contains(&token.to_ascii_lowercase().as_str())) +} + +#[cfg(test)] +mod tests { + use super::*; + + const ALL_COLS: StreamColumns = StreamColumns { + has_parent: true, + has_status: true, + has_duration: true, + }; + const ROOT_ARM: &str = "OR (reference_parent_span_id IS NULL OR reference_parent_span_id = '')"; + + fn vol(stream: &str, service: &str, requests_24h: u64) -> ServiceVolume { + ServiceVolume { + stream: stream.to_string(), + service: service.to_string(), + requests_24h, + } + } + + fn det(id: &str, service: &str, signal: RedSignal, enabled: bool) -> ManagedDetector { + ManagedDetector { + id: id.to_string(), + stream: "default".to_string(), + service: service.to_string(), + signal, + enabled, + current: true, + } + } + + fn edited(id: &str, service: &str, signal: RedSignal, enabled: bool) -> ManagedDetector { + ManagedDetector { + current: false, + ..det(id, service, signal, enabled) + } + } + + fn old_template_sql(service: &str, signal: RedSignal) -> String { + sql_of_service(service, signal).replace( + "CAST(span_kind AS VARCHAR) IN ('2','5')", + "CAST(span_kind AS VARCHAR) = '2' OR CAST(span_kind AS VARCHAR) = '5'", + ) + } + + fn sql_of_service(service: &str, signal: RedSignal) -> String { + template("default", service, signal, &ALL_COLS) + .unwrap() + .custom_sql + } + + fn parsed(id: &str, signal: RedSignal, sql: &str) -> ManagedDetector { + let tags = vec![MANAGED_TAG.to_string(), signal.tag().to_string()]; + let (service, signal, current) = parse_managed(&tags, sql, "default").unwrap(); + ManagedDetector { + current, + ..det(id, &service, signal, true) + } + } + + fn all_signals(id_prefix: &str, service: &str, enabled: bool) -> Vec { + RedSignal::ALL + .iter() + .enumerate() + .map(|(i, s)| det(&format!("{id_prefix}{i}"), service, *s, enabled)) + .collect() + } + + fn created_services(p: &Plan) -> Vec { + let mut out: Vec = Vec::new(); + for (_, service, _) in &p.create { + if !out.contains(service) { + out.push(service.clone()); + } + } + out + } + + fn plan_default(volumes: &[ServiceVolume], existing: &[ManagedDetector]) -> Plan { + plan(true, volumes, existing, 20, 10_000, usize::MAX) + } + + fn apply(existing: &mut Vec, p: &Plan) { + existing.retain(|d| !p.delete.contains(&d.id)); + for (i, (stream, service, signal)) in p.create.iter().enumerate() { + existing.push(ManagedDetector { + id: format!("new{i:03}"), + stream: stream.clone(), + service: service.clone(), + signal: *signal, + enabled: true, + current: true, + }); + } + } + + #[test] + fn selection_requires_the_volume_threshold() { + let p = plan_default( + &[ + vol("default", "busy", 10_000), + vol("default", "quiet", 9_999), + ], + &[], + ); + assert_eq!(created_services(&p), vec!["busy"]); + assert_eq!(p.create.len(), 3); + assert!(p.delete.is_empty()); + } + + #[test] + fn selection_keeps_only_the_top_n_across_streams() { + let volumes: Vec<_> = (0..25) + .map(|i| { + vol( + if i % 2 == 0 { "a" } else { "b" }, + &format!("svc{i:02}"), + 100_000 - i, + ) + }) + .collect(); + let p = plan_default(&volumes, &[]); + let services = created_services(&p); + assert_eq!(services.len(), 20); + assert_eq!(services.first().unwrap(), "svc00"); + assert_eq!(services.last().unwrap(), "svc19"); + } + + #[test] + fn ties_break_by_stream_then_service() { + let forward = [ + vol("b", "x", 50_000), + vol("a", "z", 50_000), + vol("a", "y", 50_000), + ]; + let mut reversed = forward.clone(); + reversed.reverse(); + for volumes in [&forward[..], &reversed[..]] { + let p = plan(true, volumes, &[], 2, 10_000, usize::MAX); + let picked: Vec<_> = p + .create + .iter() + .map(|(st, sv, _)| (st.as_str(), sv.as_str())) + .collect(); + assert_eq!(picked[0], ("a", "y")); + assert_eq!(picked[3], ("a", "z")); + assert_eq!(p.create.len(), 6); + } + } + + #[test] + fn hysteresis_keeps_but_does_not_create_at_60_percent_of_threshold() { + let volumes = [ + vol("default", "kept", 6_000), + vol("default", "fresh", 6_000), + ]; + let existing = all_signals("k", "kept", true); + let p = plan_default(&volumes, &existing); + assert!(p.create.is_empty(), "{:?}", p.create); + assert!(p.delete.is_empty(), "{:?}", p.delete); + } + + #[test] + fn hysteresis_keeps_but_does_not_create_at_rank_30() { + let mut volumes: Vec<_> = (0..29) + .map(|i| vol("default", &format!("top{i:02}"), 1_000_000 - i)) + .collect(); + volumes.push(vol("default", "rank30", 500_000)); + volumes.push(vol("default", "rank31", 400_000)); + let mut existing = all_signals("e", "rank30", true); + for i in 0..20 { + existing.extend(all_signals( + &format!("t{i:02}-"), + &format!("top{i:02}"), + true, + )); + } + let p = plan_default(&volumes, &existing); + assert!(p.create.is_empty(), "{:?}", p.create); + assert!(p.delete.is_empty(), "{:?}", p.delete); + } + + #[test] + fn detectors_outside_the_hysteresis_band_are_deleted() { + let mut volumes: Vec<_> = (0..40) + .map(|i| vol("default", &format!("top{i:02}"), 1_000_000 - i)) + .collect(); + volumes.push(vol("default", "rank41", 100_000)); + volumes.push(vol("default", "low", 4_999)); + let mut existing = vec![det("a", "rank41", RedSignal::Rate, true)]; + existing.push(det("b", "low", RedSignal::Rate, true)); + existing.push(det("c", "gone", RedSignal::Rate, true)); + let p = plan(true, &volumes, &existing, 20, 10_000, 0); + assert_eq!(p.delete, vec!["a", "b", "c"]); + } + + #[test] + fn creates_are_capped_per_cycle_in_rank_order() { + let volumes = [ + vol("default", "first", 90_000), + vol("default", "second", 80_000), + ]; + let p = plan(true, &volumes, &[], 20, 10_000, 4); + assert_eq!( + p.create, + vec![ + ("default".to_string(), "first".to_string(), RedSignal::Rate), + ( + "default".to_string(), + "first".to_string(), + RedSignal::ErrorRatio + ), + ( + "default".to_string(), + "first".to_string(), + RedSignal::P95Latency + ), + ("default".to_string(), "second".to_string(), RedSignal::Rate), + ] + ); + } + + #[test] + fn duplicate_triples_keep_the_oldest_id() { + let volumes = [vol("default", "svc", 50_000)]; + let mut existing = all_signals("x", "svc", true); + existing.push(det("2CdupNewer", "svc", RedSignal::Rate, true)); + existing.push(det("2AdupOlder", "svc", RedSignal::Rate, true)); + existing.retain(|d| !(d.signal == RedSignal::Rate && d.id.starts_with('x'))); + let p = plan_default(&volumes, &existing); + assert!(p.create.is_empty()); + assert_eq!(p.delete, vec!["2CdupNewer"]); + } + + #[test] + fn duplicate_triples_never_delete_a_disabled_detector() { + let volumes = [vol("default", "svc", 50_000)]; + let existing = vec![ + det("1older", "svc", RedSignal::Rate, true), + det("2newer", "svc", RedSignal::Rate, false), + det("3", "svc", RedSignal::ErrorRatio, true), + det("4", "svc", RedSignal::P95Latency, true), + ]; + let p = plan_default(&volumes, &existing); + assert_eq!(p.delete, vec!["1older"]); + } + + #[test] + fn disabling_deletes_every_managed_detector() { + let volumes = [vol("default", "svc", 50_000)]; + let mut existing = all_signals("a", "svc", true); + existing.push(det("off", "svc", RedSignal::Rate, false)); + let p = plan(false, &volumes, &existing, 20, 10_000, 15); + assert!(p.create.is_empty()); + assert_eq!(p.delete, vec!["a0", "a1", "a2", "off"]); + } + + #[test] + fn user_disabled_detectors_are_left_alone_while_kept() { + let volumes = [vol("default", "svc", 50_000)]; + let existing = all_signals("d", "svc", false); + let p = plan_default(&volumes, &existing); + assert!(p.create.is_empty()); + assert!(p.delete.is_empty()); + } + + #[test] + fn user_disabled_detectors_go_on_rank_loss_and_setting_off() { + let existing = all_signals("d", "svc", false); + let lost = plan_default(&[vol("default", "svc", 100)], &existing); + assert_eq!(lost.delete, vec!["d0", "d1", "d2"]); + let off = plan( + false, + &[vol("default", "svc", 50_000)], + &existing, + 20, + 10_000, + 15, + ); + assert_eq!(off.delete, vec!["d0", "d1", "d2"]); + } + + #[test] + fn planning_against_its_own_output_is_empty() { + let volumes: Vec<_> = (0..30) + .map(|i| vol("default", &format!("svc{i:02}"), 200_000 - i * 5_000)) + .collect(); + let mut existing = all_signals("old", "retired", true); + existing.push(det("dupe", "svc00", RedSignal::Rate, true)); + existing.push(det("dupf", "svc00", RedSignal::Rate, true)); + let first = plan_default(&volumes, &existing); + assert!(!first.create.is_empty()); + apply(&mut existing, &first); + let second = plan_default(&volumes, &existing); + assert!(second.create.is_empty(), "{:?}", second.create); + assert!(second.delete.is_empty(), "{:?}", second.delete); + } + + fn sql_of(signal: RedSignal, cols: &StreamColumns) -> String { + template("default", "checkout", signal, cols) + .unwrap() + .custom_sql + } + + #[test] + fn template_sql_uses_the_engine_aliases() { + for signal in RedSignal::ALL { + let sql = sql_of(signal, &ALL_COLS); + assert!( + sql.starts_with("SELECT histogram(_timestamp, '5m') AS time_bucket, "), + "{sql}" + ); + assert!(sql.contains(" AS value FROM "), "{sql}"); + assert!(sql.contains(" GROUP BY time_bucket"), "{sql}"); + } + assert_eq!( + sql_of(RedSignal::Rate, &ALL_COLS), + format!( + "SELECT histogram(_timestamp, '5m') AS time_bucket, COUNT(*) AS value FROM \"default\" \ + WHERE service_name = 'checkout' AND (CAST(span_kind AS VARCHAR) IN ('2','5') {ROOT_ARM}) \ + GROUP BY time_bucket" + ) + ); + } + + #[test] + fn template_sql_ratio_and_latency_need_twenty_requests() { + let errors = sql_of(RedSignal::ErrorRatio, &ALL_COLS); + assert!(errors.contains( + "CAST(SUM(CASE WHEN span_status = 'ERROR' THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) AS value" + )); + assert!( + errors.ends_with("GROUP BY time_bucket HAVING COUNT(*) >= 20"), + "{errors}" + ); + let p95 = sql_of(RedSignal::P95Latency, &ALL_COLS); + assert!(p95.contains("approx_percentile_cont(duration, 0.95) AS value")); + assert!( + p95.ends_with("GROUP BY time_bucket HAVING COUNT(*) >= 20"), + "{p95}" + ); + assert!(!sql_of(RedSignal::Rate, &ALL_COLS).contains("HAVING")); + } + + #[test] + fn template_sql_omits_the_root_arm_without_a_parent_column() { + let cols = StreamColumns { + has_parent: false, + ..ALL_COLS + }; + let sql = sql_of(RedSignal::Rate, &cols); + assert!( + sql.contains("AND (CAST(span_kind AS VARCHAR) IN ('2','5')) GROUP BY"), + "{sql}" + ); + assert!(!sql.contains("reference_parent_span_id")); + assert!(sql_of(RedSignal::Rate, &ALL_COLS).contains(ROOT_ARM)); + } + + #[test] + fn template_sql_quotes_the_service_and_the_stream() { + let t = template("my\"stream", "o'brien", RedSignal::Rate, &ALL_COLS).unwrap(); + assert!( + t.custom_sql + .contains("FROM \"my\"\"stream\" WHERE service_name = 'o''brien' AND") + ); + assert_eq!(t.stream_name, "my\"stream"); + } + + #[test] + fn template_fields_match_the_engine_contract() { + for signal in RedSignal::ALL { + let t = template("default", "checkout", signal, &ALL_COLS).unwrap(); + assert_eq!(t.stream_type, "traces"); + assert_eq!(t.query_mode, "custom_sql"); + assert_eq!(t.histogram_interval, "5m"); + assert_eq!(t.schedule_interval, "1h"); + assert!(t.detection_window_seconds >= 3_600 + 300); + assert!(!t.alert_enabled); + assert_eq!( + t.tags, + vec![MANAGED_TAG.to_string(), signal.tag().to_string()] + ); + assert!( + t.name + .starts_with(&format!("RED {} · default · checkout", signal.label())) + ); + } + let rate = template("default", "checkout", RedSignal::Rate, &ALL_COLS).unwrap(); + assert_eq!( + ( + rate.detection_function.as_str(), + rate.alert_direction.as_str() + ), + ("count", "both") + ); + assert_eq!(rate.detection_function_field, None); + let errors = template("default", "checkout", RedSignal::ErrorRatio, &ALL_COLS).unwrap(); + assert_eq!( + ( + errors.detection_function.as_str(), + errors.alert_direction.as_str() + ), + ("count", "above") + ); + let p95 = template("default", "checkout", RedSignal::P95Latency, &ALL_COLS).unwrap(); + assert_eq!( + ( + p95.detection_function.as_str(), + p95.alert_direction.as_str() + ), + ("p95", "above") + ); + // In custom_sql mode the field names the result column the engine reads the value from. + assert_eq!(p95.detection_function_field.as_deref(), Some("value")); + } + + #[test] + fn template_names_never_contain_a_slash() { + let t = template("team/a", "api/v1", RedSignal::Rate, &ALL_COLS).unwrap(); + assert!(!t.name.contains('/'), "{}", t.name); + assert_eq!(t.name, "RED rate · team\u{2215}a · api\u{2215}v1"); + } + + #[test] + fn template_skips_signals_the_engine_would_reject() { + assert!( + template( + "default", + "inventory-update-svc", + RedSignal::Rate, + &ALL_COLS + ) + .is_none() + ); + assert!(template("dropzone", "svc", RedSignal::P95Latency, &ALL_COLS).is_none()); + assert!(template("app-errors", "svc", RedSignal::Rate, &ALL_COLS).is_none()); + assert!(template("app-errors", "svc", RedSignal::ErrorRatio, &ALL_COLS).is_some()); + assert!(template("terror_logs", "svc", RedSignal::Rate, &ALL_COLS).is_some()); + let no_status = StreamColumns { + has_status: false, + ..ALL_COLS + }; + assert!(template("default", "svc", RedSignal::ErrorRatio, &no_status).is_none()); + assert!(template("default", "svc", RedSignal::Rate, &no_status).is_some()); + let no_duration = StreamColumns { + has_duration: false, + ..ALL_COLS + }; + assert!(template("default", "svc", RedSignal::P95Latency, &no_duration).is_none()); + assert!(template("default", "svc", RedSignal::ErrorRatio, &no_duration).is_some()); + } + + #[test] + fn detectors_on_unread_streams_are_left_out_of_the_plan() { + let reads = BTreeMap::from([ + ("default".to_string(), StreamRead::Read(ALL_COLS)), + ("flaky".to_string(), StreamRead::Unread), + ]); + let mut cold = det("b", "svc", RedSignal::Rate, false); + cold.stream = "cold".to_string(); + let mut flaky = det("f", "svc", RedSignal::Rate, true); + flaky.stream = "flaky".to_string(); + let read = det("a", "svc", RedSignal::Rate, true); + let existing = on_read_streams(vec![read.clone(), cold, flaky], &reads); + assert_eq!(existing, vec![read]); + assert_eq!(plan_default(&[], &existing).delete, vec!["a".to_string()]); + } + + #[test] + fn a_failed_volume_search_spares_only_its_own_stream() { + let reads = BTreeMap::from([ + ("default".to_string(), StreamRead::Read(ALL_COLS)), + ("flaky".to_string(), StreamRead::Unread), + ]); + let mut flaky = det("f", "svc", RedSignal::Rate, true); + flaky.stream = "flaky".to_string(); + let stale = det("s", "old", RedSignal::Rate, true); + let existing = on_read_streams(vec![flaky, stale.clone()], &reads); + assert_eq!(existing, vec![stale]); + let p = plan_default(&[vol("default", "checkout", 50_000)], &existing); + assert_eq!(p.delete, vec!["s".to_string()]); + assert_eq!(created_services(&p), vec!["checkout".to_string()]); + } + + #[test] + fn detectors_on_retired_streams_are_deleted() { + let reads = BTreeMap::from([ + ("default".to_string(), StreamRead::Read(ALL_COLS)), + ("gone".to_string(), StreamRead::Retired), + ]); + let mut gone = det("g", "svc", RedSignal::Rate, true); + gone.stream = "gone".to_string(); + let mut cold = det("c", "svc", RedSignal::Rate, true); + cold.stream = "cold".to_string(); + let existing = on_read_streams(vec![gone.clone(), cold], &reads); + assert_eq!(existing, vec![gone]); + assert_eq!(plan_default(&[], &existing).delete, vec!["g".to_string()]); + } + + #[test] + fn a_stream_without_service_name_or_span_kind_is_retired() { + let full = [ + "service_name", + "span_kind", + "reference_parent_span_id", + "span_status", + "duration", + ]; + assert_eq!( + read_stream(|name| full.contains(&name)), + StreamRead::Read(ALL_COLS) + ); + assert_eq!( + read_stream(|name| ["service_name", "span_kind"].contains(&name)), + StreamRead::Read(StreamColumns { + has_parent: false, + has_status: false, + has_duration: false, + }) + ); + assert_eq!( + read_stream(|name| name != "service_name" && full.contains(&name)), + StreamRead::Retired + ); + assert_eq!( + read_stream(|name| name != "span_kind" && full.contains(&name)), + StreamRead::Retired + ); + } + + #[test] + fn parse_managed_round_trips_every_template() { + for has_parent in [true, false] { + let cols = StreamColumns { + has_parent, + ..ALL_COLS + }; + for service in ["checkout", "o'brien's ''svc''", "a · b", "api/v1"] { + for signal in RedSignal::ALL { + let t = template("my\"stream", service, signal, &cols).unwrap(); + assert_eq!( + parse_managed(&t.tags, &t.custom_sql, "my\"stream"), + Some((service.to_string(), signal, true)), + "{}", + t.custom_sql + ); + } + } + } + } + + #[test] + fn parse_managed_marks_edited_sql_non_current_and_needs_a_tag_and_service() { + let t = template("default", "checkout", RedSignal::P95Latency, &ALL_COLS).unwrap(); + let edited_sql = t.custom_sql.replace("0.95", "0.99"); + assert_eq!( + parse_managed(&t.tags, &edited_sql, "default"), + Some(("checkout".to_string(), RedSignal::P95Latency, false)) + ); + let other_service = t.custom_sql.replace("= 'checkout'", "= 'cart' OR 1=1"); + assert_eq!( + parse_managed(&t.tags, &other_service, "default"), + Some(("cart".to_string(), RedSignal::P95Latency, false)) + ); + let rate_tags = vec![MANAGED_TAG.to_string(), RedSignal::Rate.tag().to_string()]; + assert_eq!( + parse_managed(&rate_tags, &t.custom_sql, "default"), + Some(("checkout".to_string(), RedSignal::Rate, false)) + ); + assert_eq!( + parse_managed(&[MANAGED_TAG.to_string()], &t.custom_sql, "default"), + None + ); + assert_eq!(parse_managed(&t.tags, "", "default"), None); + let no_literal = t.custom_sql.replace("= 'checkout'", "IN ('checkout')"); + assert_eq!(parse_managed(&t.tags, &no_literal, "default"), None); + } + + #[test] + fn parse_managed_recognises_the_old_template_as_non_current() { + for signal in RedSignal::ALL { + assert_eq!( + parse_managed( + &[MANAGED_TAG.to_string(), signal.tag().to_string()], + &old_template_sql("o'brien", signal), + "default" + ), + Some(("o'brien".to_string(), signal, false)) + ); + } + } + + #[test] + fn parse_managed_needs_the_from_stream_to_be_the_stream_name() { + let t = template("default", "checkout", RedSignal::Rate, &ALL_COLS).unwrap(); + assert_eq!(parse_managed(&t.tags, &t.custom_sql, "other"), None); + let old = old_template_sql("checkout", RedSignal::Rate); + assert_eq!(parse_managed(&t.tags, &old, "other"), None); + } + + #[test] + fn an_old_template_detector_blocks_creation_for_its_triple() { + let volumes = [vol("default", "svc", 50_000)]; + let existing = vec![parsed( + "o", + RedSignal::Rate, + &old_template_sql("svc", RedSignal::Rate), + )]; + let p = plan_default(&volumes, &existing); + assert!(p.delete.is_empty(), "{:?}", p.delete); + assert_eq!( + p.create, + vec![ + ( + "default".to_string(), + "svc".to_string(), + RedSignal::ErrorRatio + ), + ( + "default".to_string(), + "svc".to_string(), + RedSignal::P95Latency + ), + ] + ); + } + + #[test] + fn an_old_and_a_current_detector_for_one_triple_are_both_kept() { + let volumes = [vol("default", "svc", 50_000)]; + let mut existing = Vec::new(); + for (i, signal) in RedSignal::ALL.into_iter().enumerate() { + existing.push(parsed( + &format!("1old{i}"), + signal, + &old_template_sql("svc", signal), + )); + existing.push(parsed( + &format!("2new{i}"), + signal, + &sql_of_service("svc", signal), + )); + } + assert_eq!(plan_default(&volumes, &existing), Plan::default()); + } + + #[test] + fn a_user_edited_detector_is_never_duplicated_or_deleted_by_rank() { + let user_edited = [edited("e", "svc", RedSignal::Rate, true)]; + let kept = plan_default(&[vol("default", "svc", 50_000)], &user_edited); + assert!( + !kept.create.iter().any(|(_, _, s)| *s == RedSignal::Rate), + "{:?}", + kept.create + ); + assert!(kept.delete.is_empty()); + let dropped = plan_default(&[vol("default", "svc", 100)], &user_edited); + assert_eq!(dropped, Plan::default()); + assert_eq!(plan_default(&[], &user_edited), Plan::default()); + } + + #[test] + fn an_edited_clone_and_its_original_are_both_kept() { + let volumes = [vol("default", "svc", 50_000)]; + let mut existing: Vec<_> = RedSignal::ALL + .into_iter() + .enumerate() + .map(|(i, s)| parsed(&format!("c{i}"), s, &sql_of_service("svc", s))) + .collect(); + let clone_sql = sql_of_service("svc", RedSignal::P95Latency).replace("0.95", "0.99"); + existing.push(parsed("clone", RedSignal::P95Latency, &clone_sql)); + assert_eq!(plan_default(&volumes, &existing), Plan::default()); + } + + #[test] + fn a_parsed_edited_detector_is_stable_across_cycles() { + let volumes = [vol("default", "svc", 50_000)]; + let sql = sql_of_service("svc", RedSignal::P95Latency).replace("0.95", "0.99"); + let mut existing = vec![parsed("e", RedSignal::P95Latency, &sql)]; + assert!(!existing[0].current); + let first = plan_default(&volumes, &existing); + assert!(first.delete.is_empty(), "{:?}", first.delete); + apply(&mut existing, &first); + let second = plan_default(&volumes, &existing); + assert_eq!(second, Plan::default()); + assert_eq!( + existing + .iter() + .filter(|d| d.service == "svc" && d.signal == RedSignal::P95Latency) + .map(|d| d.id.as_str()) + .collect::>(), + vec!["e"] + ); + } + + #[test] + fn parse_managed_tolerates_whitespace_keyword_case_and_spacing_edits() { + for signal in RedSignal::ALL { + let sql = sql_of_service("svc", signal); + let tags = vec![MANAGED_TAG.to_string(), signal.tag().to_string()]; + assert_eq!( + parse_managed(&tags, &sql, "default"), + Some(("svc".to_string(), signal, true)) + ); + let reformatted = sql + .replace(" FROM ", "\n FROM\n ") + .replace(" WHERE ", "\n WHERE\t") + .replace(" AND (", "\n AND ("); + let lower_case = sql + .replace(" FROM ", " from ") + .replace(" WHERE ", " where "); + let tight = sql.replace("service_name = 'svc'", "service_name='svc'"); + let spaced = sql + .replace(" FROM ", " FROM ") + .replace(" = 'svc'", " = 'svc'"); + for variant in [reformatted, lower_case, tight, spaced] { + assert_eq!( + parse_managed(&tags, &variant, "default"), + Some(("svc".to_string(), signal, false)), + "{variant}" + ); + let p = plan_default( + &[vol("default", "svc", 50_000)], + &[parsed("v", signal, &variant)], + ); + assert!(!p.create.iter().any(|(_, _, s)| *s == signal), "{p:?}"); + assert!(p.delete.is_empty(), "{p:?}"); + } + } + } + + #[test] + fn volume_sql_counts_requests_per_service() { + assert_eq!( + volume_sql("my\"stream", &ALL_COLS), + format!( + "SELECT service_name, COUNT(*) FILTER (WHERE CAST(span_kind AS VARCHAR) IN ('2','5') {ROOT_ARM}) \ + AS requests FROM \"my\"\"stream\" GROUP BY service_name ORDER BY requests DESC LIMIT 40" + ) + ); + let cols = StreamColumns { + has_parent: false, + ..ALL_COLS + }; + assert_eq!( + volume_sql("default", &cols), + "SELECT service_name, COUNT(*) FILTER (WHERE CAST(span_kind AS VARCHAR) IN ('2','5')) \ + AS requests FROM \"default\" GROUP BY service_name ORDER BY requests DESC LIMIT 40" + ); + } + + /// The kind IN list puts `[` in the aggregate's name; the top-k rule must still match it. + #[tokio::test] + async fn volume_sql_plans_through_the_topk_rule() { + use std::sync::Arc; + + use arrow_schema::{DataType, Field, Schema}; + use datafusion::{ + config::ConfigOptions, + physical_optimizer::PhysicalOptimizerRule, + physical_plan::displayable, + prelude::{SessionConfig, SessionContext}, + }; + use search::datafusion::{ + optimizer::physical_optimizer::aggregate_topk::AggregateTopkRule, + table_provider::empty_table::NewEmptyTable, + }; + + let schema = Arc::new(Schema::new(vec![ + Field::new("service_name", DataType::Utf8, true), + Field::new("span_kind", DataType::Utf8, true), + Field::new("reference_parent_span_id", DataType::Utf8, true), + ])); + let ctx = SessionContext::new_with_config(SessionConfig::new().with_target_partitions(2)); + ctx.register_table("default", Arc::new(NewEmptyTable::new("default", schema))) + .unwrap(); + for has_parent in [true, false] { + let cols = StreamColumns { + has_parent, + ..ALL_COLS + }; + let sql = volume_sql("default", &cols); + let plan = ctx + .sql(&sql) + .await + .unwrap() + .create_physical_plan() + .await + .unwrap(); + let plan = AggregateTopkRule::new(MAX_SERVICES as i64 * 2) + .optimize(plan, &ConfigOptions::default()) + .unwrap(); + let shown = displayable(plan.as_ref()).indent(true).to_string(); + assert!(shown.contains("AggregateTopkExec"), "{sql}\n{shown}"); + } + } +} diff --git a/src/db/src/organization.rs b/src/db/src/organization.rs index 9ccdaf7ed9..57f397aca6 100644 --- a/src/db/src/organization.rs +++ b/src/db/src/organization.rs @@ -169,6 +169,29 @@ pub async fn get_org_setting_usage_stream_enabled(org_id: &str) -> Result Result { + let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}"); + if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) { + return Ok(v.red_insights_enabled); + } + + let settings: OrganizationSetting = match db::get(&key).await { + Ok(settings) => json::from_slice(&settings)?, + Err(Error::DbError(infra::errors::DbError::KeyNotExists(_))) => { + OrganizationSetting::default() + } + Err(e) => return Err(e), + }; + let red_insights_enabled = settings.red_insights_enabled; + + ORGANIZATION_SETTING + .write() + .await + .insert(key.to_string(), settings); + Ok(red_insights_enabled) +} + /// Cache the existing org settings in the beginning pub async fn org_settings_cache() -> Result<(), anyhow::Error> { let prefix = ORG_SETTINGS_KEY_PREFIX; diff --git a/src/db/src/saved_view.rs b/src/db/src/saved_view.rs index d406e9d777..4cd85a17aa 100644 --- a/src/db/src/saved_view.rs +++ b/src/db/src/saved_view.rs @@ -22,9 +22,18 @@ use infra::errors::Error; use crate as db; pub const SAVED_VIEWS_KEY_PREFIX: &str = "/organization/savedviews"; +const DEFAULT_VIEW_TYPE: &str = "logs"; +const VIEW_TYPES: [&str; 2] = ["logs", "traces"]; pub async fn set_view(org_id: &str, view: &CreateViewRequest) -> Result { - if view_exists_with_name(org_id, &view.view_name) + let view_type = view.view_type.as_deref().unwrap_or(DEFAULT_VIEW_TYPE); + if !VIEW_TYPES.contains(&view_type) { + return Err(Error::Message(format!( + "Invalid view_type '{view_type}', expected one of: {}", + VIEW_TYPES.join(", ") + ))); + } + if view_exists_with_name(org_id, &view.view_name, view_type) .await .is_some() { @@ -39,6 +48,7 @@ pub async fn set_view(org_id: &str, view: &CreateViewRequest) -> Result Result { - if let Some(existing_id) = view_exists_with_name(org_id, &view.view_name).await + let original_view = get_view(org_id, view_id).await?; + let view_type = original_view + .view_type + .as_deref() + .unwrap_or(DEFAULT_VIEW_TYPE); + if let Some(existing_id) = view_exists_with_name(org_id, &view.view_name, view_type).await && existing_id != view_id { return Err(Error::Message(format!( @@ -65,13 +80,10 @@ pub async fn update_view( ))); } let key = format!("{SAVED_VIEWS_KEY_PREFIX}/{org_id}/{view_id}"); - let updated_view = match get_view(org_id, view_id).await { - Ok(original_view) => View { - data: view.data.clone(), - view_name: view.view_name.clone(), - ..original_view - }, - Err(e) => return Err(e), + let updated_view = View { + data: view.data.clone(), + view_name: view.view_name.clone(), + ..original_view }; let val = json::to_vec(&updated_view) .map_err(|e| Error::Message(format!("Failed to serialize saved view: {e}")))?; @@ -94,8 +106,7 @@ pub async fn get_view(org_id: &str, view_id: &str) -> Result { /// Return all the saved views but query limited data only, associated with a /// provided org_id This will not contain the payload. pub async fn get_views_list_only(org_id: &str) -> Result { - let key = format!("{SAVED_VIEWS_KEY_PREFIX}/{org_id}"); - let ret = db::list_values(&key).await?; + let ret = db::list_values(&list_prefix(org_id)).await?; let mut views: Vec = ret .iter() .filter_map(|view| json::from_slice(view).ok()) @@ -114,17 +125,23 @@ pub async fn delete_view(org_id: &str, view_id: &str) -> Result<(), Error> { Ok(()) } -/// Check if a saved view with the given name already exists in the org. -/// Returns `Some(view_id)` if found, `None` otherwise. -async fn view_exists_with_name(org_id: &str, view_name: &str) -> Option { +async fn view_exists_with_name(org_id: &str, view_name: &str, view_type: &str) -> Option { let views = get_views_list_only(org_id).await.ok()?; views .views .iter() - .find(|v| v.view_name.eq_ignore_ascii_case(view_name)) + .find(|v| { + v.view_type.as_deref().unwrap_or(DEFAULT_VIEW_TYPE) == view_type + && v.view_name.eq_ignore_ascii_case(view_name) + }) .map(|v| v.view_id.clone()) } +// Defensive trailing `/`: a pure prefix-scanning store would otherwise match org `ab` for `a`. +fn list_prefix(org_id: &str) -> String { + format!("{SAVED_VIEWS_KEY_PREFIX}/{org_id}/") +} + #[cfg(test)] mod tests { use super::*; @@ -138,4 +155,12 @@ mod tests { fn test_saved_views_key_prefix_starts_with_slash() { assert!(SAVED_VIEWS_KEY_PREFIX.starts_with('/')); } + + #[test] + fn test_list_prefix_does_not_match_an_org_sharing_its_prefix() { + let own_key = format!("{SAVED_VIEWS_KEY_PREFIX}/a/view1"); + let other_key = format!("{SAVED_VIEWS_KEY_PREFIX}/ab/view1"); + assert!(own_key.starts_with(&list_prefix("a"))); + assert!(!other_key.starts_with(&list_prefix("a"))); + } } diff --git a/src/jobs/src/job/mod.rs b/src/jobs/src/job/mod.rs index fae5864443..a428a59865 100644 --- a/src/jobs/src/job/mod.rs +++ b/src/jobs/src/job/mod.rs @@ -65,6 +65,8 @@ mod pipeline_error_cleanup; mod prompt_webhook_delivery; mod promql; mod promql_self_consume; +#[cfg(feature = "enterprise")] +mod red_insights; mod scheduler; #[cfg(feature = "enterprise")] mod service_graph; @@ -1099,6 +1101,8 @@ pub async fn init() -> Result<(), anyhow::Error> { .disabled { tokio::task::spawn(anomaly_claim_supervisor()); + // Here because create_config does not check the anomaly kill switch itself. + tokio::task::spawn(red_insights::run()); } // Every node that serves writes publishes them, not only the scheduler. openobserve_synthetics::service::start_publish_queue(); diff --git a/src/jobs/src/job/red_insights.rs b/src/jobs/src/job/red_insights.rs new file mode 100644 index 0000000000..4c5dd08b9f --- /dev/null +++ b/src/jobs/src/job/red_insights.rs @@ -0,0 +1,328 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::{collections::BTreeMap, time::Duration}; + +use config::meta::{ + folder::{Folder, FolderType}, + search::{Query, Request, RequestEncoding, SearchEventType}, + stream::StreamType, +}; +use db::folders::FolderError; +use openobserve_core::{ + anomaly_detection, + traces::red_insights::{ + self, FOLDER_NAME, MANAGED_TAG, MAX_CREATES, MAX_SERVICES, MIN_REQUESTS_24H, + ManagedDetector, ServiceVolume, StreamRead, + }, +}; + +const CYCLE: Duration = Duration::from_secs(6 * 60 * 60); +/// Lets the job-cluster election resolve before the first claim check. +const FIRST_RUN_DELAY: Duration = Duration::from_secs(5 * 60); +const VOLUME_WINDOW_US: i64 = 24 * 60 * 60 * 1_000_000; + +pub async fn run() { + let start = tokio::time::Instant::now() + FIRST_RUN_DELAY; + let mut ticker = tokio::time::interval_at(start, CYCLE); + loop { + ticker.tick().await; + // The per-org dist lock is regional, so only the region holding the claim reconciles. + if !super::anomaly_holds_job_cluster_claim().await { + continue; + } + // The per-org lock waits, not skips, so each scheduler would re-reconcile in turn. + if !super::leader::is_alert_manager_leader().await { + continue; + } + let orgs = match db::organization::list(None).await { + Ok(orgs) => orgs, + Err(e) => { + log::error!("[RED insights] could not list orgs: {e}"); + continue; + } + }; + for org in orgs { + if let Err(e) = reconcile_if_needed(&org.identifier).await { + log::error!("[RED insights] org {}: {e}", org.identifier); + } + } + } +} + +async fn reconcile_if_needed(org_id: &str) -> anyhow::Result<()> { + let prefetched = if db::organization::get_org_setting_red_insights_enabled(org_id).await? { + None + } else { + let configs = managed_configs(org_id).await?; + if configs.is_empty() { + return Ok(()); + } + Some(configs) + }; + let locker = infra::dist_lock::lock(&format!("red_insights/{org_id}"), 0).await?; + let result = reconcile(org_id, prefetched).await; + if let Err(e) = infra::dist_lock::unlock(&locker).await { + log::warn!("[RED insights] org {org_id}: failed to release the lock: {e}"); + } + result +} + +async fn reconcile(org_id: &str, prefetched: Option>) -> anyhow::Result<()> { + let configs = match prefetched { + Some(configs) => configs, + None => managed_configs(org_id).await?, + }; + if !db::organization::get_org_setting_red_insights_enabled(org_id).await? { + for config in &configs { + delete(org_id, config_str(config, "anomaly_id")).await; + } + return Ok(()); + } + + let mut reads = read_streams(org_id).await; + let volumes = service_volumes(org_id, &mut reads).await; + let detectors: Vec<_> = configs.iter().filter_map(managed_detector).collect(); + retire_removed_streams(org_id, &detectors, &mut reads).await; + let existing = red_insights::on_read_streams(detectors, &reads); + // Uncapped here: a signal the template skips must not use up the per-cycle create budget. + let plan = red_insights::plan( + true, + &volumes, + &existing, + MAX_SERVICES, + MIN_REQUESTS_24H, + usize::MAX, + ); + for id in &plan.delete { + delete(org_id, id).await; + } + + let templates: Vec<_> = plan + .create + .iter() + .filter_map(|(stream, service, signal)| match reads.get(stream)? { + StreamRead::Read(cols) => red_insights::template(stream, service, *signal, cols), + _ => None, + }) + .take(MAX_CREATES) + .collect(); + if templates.is_empty() { + return Ok(()); + } + let folder_id = red_folder(org_id, &configs).await?; + for template in templates { + let name = template.name.clone(); + if let Err(e) = + anomaly_detection::create_config(org_id, template.into_request(&folder_id)).await + { + log::error!("[RED insights] org {org_id}: failed to create '{name}': {e}"); + } + } + Ok(()) +} + +async fn managed_configs(org_id: &str) -> anyhow::Result> { + let configs = anomaly_detection::list_configs(org_id, None, None).await?; + Ok(configs + .into_iter() + .filter(|c| config_tags(c).iter().any(|t| t == MANAGED_TAG)) + .collect()) +} + +fn config_tags(config: &serde_json::Value) -> Vec { + config + .get("tags") + .and_then(|v| v.as_array()) + .map(|tags| { + tags.iter() + .filter_map(|t| t.as_str().map(str::to_string)) + .collect() + }) + .unwrap_or_default() +} + +fn config_str<'a>(config: &'a serde_json::Value, key: &str) -> &'a str { + config.get(key).and_then(|v| v.as_str()).unwrap_or_default() +} + +/// An unparseable managed detector maps to `None`, so the plan leaves it alone. +fn managed_detector(config: &serde_json::Value) -> Option { + let stream = config_str(config, "stream_name"); + let (service, signal, current) = red_insights::parse_managed( + &config_tags(config), + config_str(config, "custom_sql"), + stream, + )?; + Some(ManagedDetector { + id: config_str(config, "anomaly_id").to_string(), + stream: stream.to_string(), + service, + signal, + enabled: config + .get("enabled") + .and_then(|v| v.as_bool()) + .unwrap_or(true), + current, + }) +} + +async fn delete(org_id: &str, anomaly_id: &str) { + if let Err(e) = anomaly_detection::delete_config(org_id, anomaly_id).await { + log::error!("[RED insights] org {org_id}: failed to delete {anomaly_id}: {e}"); + } +} + +/// The public folder id, which is what `create_config`'s `folder_id` takes. +async fn red_folder(org_id: &str, configs: &[serde_json::Value]) -> anyhow::Result { + if let Some(folder_id) = configs + .iter() + .map(|c| config_str(c, "folder_id")) + .find(|id| !id.is_empty()) + { + return Ok(folder_id.to_string()); + } + match db::folders::get_folder_by_name(org_id, FOLDER_NAME, FolderType::Alerts).await { + Ok(folder) => return Ok(folder.folder_id), + Err(FolderError::NotFound) => {} + Err(e) => return Err(e.into()), + } + let folder = Folder { + folder_id: String::new(), + name: FOLDER_NAME.to_string(), + description: "Anomaly detectors managed by RED insights".to_string(), + icon: None, + }; + match db::folders::save_folder(org_id, folder, FolderType::Alerts, false).await { + Ok(folder) => Ok(folder.folder_id), + Err(FolderError::FolderNameAlreadyExists) => { + Ok( + db::folders::get_folder_by_name(org_id, FOLDER_NAME, FolderType::Alerts) + .await? + .folder_id, + ) + } + Err(e) => Err(e.into()), + } +} + +async fn read_streams(org_id: &str) -> BTreeMap { + let mut reads = BTreeMap::new(); + for stream in db::schema::list_streams_from_cache(org_id, StreamType::Traces).await { + let read = match infra::schema::get(org_id, &stream, StreamType::Traces).await { + Ok(schema) => red_insights::read_stream(|name| schema.field_with_name(name).is_ok()), + Err(e) => { + log::warn!("[RED insights] org {org_id}: schema read for {stream} failed: {e}"); + StreamRead::Unread + } + }; + if read == StreamRead::Retired { + log::info!("[RED insights] org {org_id}: {stream} has no service_name/span_kind"); + } + reads.insert(stream, read); + } + reads +} + +/// A lookup error leaves the stream unread, so a transient failure never deletes its detectors. +async fn retire_removed_streams( + org_id: &str, + detectors: &[ManagedDetector], + reads: &mut BTreeMap, +) { + let unlisted: Vec = detectors + .iter() + .filter(|d| !reads.contains_key(&d.stream)) + .map(|d| d.stream.clone()) + .collect(); + for stream in unlisted { + match infra::schema::get(org_id, &stream, StreamType::Traces).await { + // A missing stream reads back as an empty schema, never as an error. + Ok(schema) if schema.fields().is_empty() => { + reads.insert(stream, StreamRead::Retired); + } + Ok(_) => {} + Err(e) => { + log::warn!("[RED insights] org {org_id}: schema read for {stream} failed: {e}") + } + } + } +} + +/// A failed or partial search marks its stream unread: zero traffic would cost it its detectors. +async fn service_volumes( + org_id: &str, + reads: &mut BTreeMap, +) -> Vec { + let end_time = config::utils::time::now_micros(); + let mut volumes = Vec::new(); + for (stream, read) in reads.iter_mut() { + let StreamRead::Read(cols) = *read else { + continue; + }; + let req = Request { + query: Query { + sql: red_insights::volume_sql(stream, &cols), + from: 0, + size: (MAX_SERVICES * 2) as i64, + start_time: end_time - VOLUME_WINDOW_US, + end_time, + track_total_hits: false, + uses_zo_fn: false, + query_fn: None, + ..Default::default() + }, + encoding: RequestEncoding::Empty, + regions: vec![], + clusters: vec![], + timeout: 0, + // Background nodes, as for anomaly training, so this never lands on interactive ones. + search_type: Some(SearchEventType::DerivedStream), + search_event_context: None, + use_cache: false, + clear_cache: false, + local_mode: None, + agent_options: None, + }; + let trace_id = config::ider::generate_trace_id(); + let resp = match search_service::search(&trace_id, org_id, StreamType::Traces, None, &req) + .await + { + Ok(resp) if !resp.is_partial => resp, + Ok(_) => { + log::warn!("[RED insights] org {org_id}: volume search on {stream} was partial"); + *read = StreamRead::Unread; + continue; + } + Err(e) => { + log::warn!("[RED insights] org {org_id}: volume search on {stream} failed: {e}"); + *read = StreamRead::Unread; + continue; + } + }; + volumes.extend(resp.hits.iter().filter_map(|hit| { + let service = hit + .get("service_name")? + .as_str() + .filter(|s| !s.is_empty())?; + Some(ServiceVolume { + stream: stream.clone(), + service: service.to_string(), + requests_24h: hit.get("requests")?.as_u64()?, + }) + })); + } + volumes +} diff --git a/src/search/src/datafusion/distributed_plan/aggregate_topk_exec/mod.rs b/src/search/src/datafusion/distributed_plan/aggregate_topk_exec/mod.rs index ee8d9abba3..0fd71b317c 100644 --- a/src/search/src/datafusion/distributed_plan/aggregate_topk_exec/mod.rs +++ b/src/search/src/datafusion/distributed_plan/aggregate_topk_exec/mod.rs @@ -17,7 +17,7 @@ use std::{fmt::Debug, sync::Arc}; use arrow::datatypes::SchemaRef; use datafusion::{ - common::{Result, tree_node::TreeNodeRecursion}, + common::{Result, internal_err, tree_node::TreeNodeRecursion}, execution::{SendableRecordBatchStream, TaskContext}, physical_expr::{EquivalenceProperties, PhysicalExpr}, physical_plan::{ @@ -42,37 +42,36 @@ pub struct AggregateTopkExec { } impl AggregateTopkExec { - /// Create a new AggregateMergeExec with explicit cache strategy - pub fn new( + /// Create a new AggregateTopkExec; errors when the input has no field for `sort_field`. + pub fn try_new( input: Arc, sort_field: &str, descending: bool, limit: u64, - ) -> Self { + ) -> Result { // Partial or no cache: cached partitions + input partitions let target_partitions = input.output_partitioning().partition_count(); let cache = Self::compute_properties(Arc::clone(&input.schema()), target_partitions); - let sort_field = input - .schema() - .fields() - .iter() - .find(|f| { - // field name like count(*)[count] - f.name() == sort_field - || f.name().split('[').next().is_some_and(|v| v == sort_field) - }) - .unwrap() - .name() - .to_string(); + let schema = input.schema(); + let Some(field) = schema.fields().iter().find(|f| { + // State fields are `[count]`; the name itself may hold `[`, e.g. an IN list. + f.name() == sort_field + || f.name() + .rsplit_once('[') + .is_some_and(|(name, _)| name == sort_field) + }) else { + return internal_err!("AggregateTopkExec: no input field for {sort_field}"); + }; + let sort_field = field.name().to_string(); - Self { + Ok(Self { input, cache, target_partitions, sort_field, descending, limit, - } + }) } fn output_partitioning_helper(n_partitions: usize) -> Partitioning { @@ -154,12 +153,12 @@ impl ExecutionPlan for AggregateTopkExec { if children.is_empty() { return Ok(self); } - Ok(Arc::new(Self::new( + Ok(Arc::new(Self::try_new( children[0].clone(), &self.sort_field, self.descending, self.limit, - ))) + )?)) } fn execute( diff --git a/src/search/src/datafusion/distributed_plan/codec/aggregate_topk_exec.rs b/src/search/src/datafusion/distributed_plan/codec/aggregate_topk_exec.rs index d0a3bb299c..c5e070bba3 100644 --- a/src/search/src/datafusion/distributed_plan/codec/aggregate_topk_exec.rs +++ b/src/search/src/datafusion/distributed_plan/codec/aggregate_topk_exec.rs @@ -31,12 +31,12 @@ pub fn try_decode( inputs: &[Arc], _registry: &dyn FunctionRegistry, ) -> Result> { - Ok(Arc::new(AggregateTopkExec::new( + Ok(Arc::new(AggregateTopkExec::try_new( inputs[0].clone(), &node.sort_field, node.descending, node.limit, - ))) + )?)) } pub fn try_encode(node: Arc, buf: &mut Vec) -> Result<()> { @@ -105,12 +105,12 @@ mod tests { Arc::new(EmptyExec::new(Arc::clone(&schema))), Arc::clone(&schema), )?; - let plan: Arc = Arc::new(AggregateTopkExec::new( + let plan: Arc = Arc::new(AggregateTopkExec::try_new( Arc::new(agg_plan) as Arc, "COUNT(1)", false, 10, - )); + )?); // encode let proto = super::super::get_physical_extension_codec(); diff --git a/src/search/src/datafusion/optimizer/physical_optimizer/aggregate_topk.rs b/src/search/src/datafusion/optimizer/physical_optimizer/aggregate_topk.rs index 31692587ca..71f61f4e33 100644 --- a/src/search/src/datafusion/optimizer/physical_optimizer/aggregate_topk.rs +++ b/src/search/src/datafusion/optimizer/physical_optimizer/aggregate_topk.rs @@ -73,7 +73,13 @@ impl PhysicalOptimizerRule for AggregateTopkRule { } if let Some(expr) = final_agg_plan.aggr_expr().first() { - if !["count", "avg", "min", "max", "sum"].contains(&expr.fun().name()) { + let eligible = match expr.fun().name() { + "min" | "max" => true, + // A DISTINCT count/sum partial state is a value list, not the final value. + "count" | "sum" => !expr.is_distinct(), + _ => false, + }; + if !eligible { return Ok(plan); } let expr_name = expr.name(); @@ -133,8 +139,11 @@ impl TreeNodeRewriter for AggregateTopkRewriter { } let input_plan = Arc::clone(agg_node); - let agg_plan = - AggregateTopkExec::new(input_plan, &self.field, self.descending, self.limit); + let Ok(agg_plan) = + AggregateTopkExec::try_new(input_plan, &self.field, self.descending, self.limit) + else { + return Ok(Transformed::no(node)); + }; let node = node.replace_children( vec![Arc::new(agg_plan) as Arc], @@ -230,8 +239,16 @@ impl<'n> TreeNodeVisitor<'n> for SortLimitVisitor { #[cfg(test)] mod tests { + use arrow::{ + array::{Array, Int64Array, RecordBatch, StringArray}, + compute::cast, + }; use arrow_schema::{DataType, Field, Schema}; - use datafusion::{physical_plan::displayable, prelude::SessionConfig}; + use datafusion::{ + datasource::MemTable, + physical_plan::{collect, displayable}, + prelude::SessionConfig, + }; use super::*; use crate::datafusion::table_provider::empty_table::NewEmptyTable; @@ -248,6 +265,10 @@ mod tests { /// Whether the rule inserts `AggregateTopkExec` for `agg(v)` grouped by `name`, sorted /// on the aggregate and limited. async fn inserts_topk(agg: &str) -> bool { + inserts_topk_for(&format!("{agg}(v)")).await + } + + async fn inserts_topk_for(select_expr: &str) -> bool { enable_topk(); let schema = Arc::new(Schema::new(vec![ Field::new("name", DataType::Utf8, false), @@ -258,8 +279,9 @@ mod tests { ); ctx.register_table("t", Arc::new(NewEmptyTable::new("t", schema))) .unwrap(); - let sql = - format!("SELECT name, {agg}(v) AS x FROM t GROUP BY name ORDER BY x DESC LIMIT 10"); + let sql = format!( + "SELECT name, {select_expr} AS x FROM t GROUP BY name ORDER BY x DESC LIMIT 10" + ); let plan = ctx .sql(&sql) .await @@ -278,11 +300,17 @@ mod tests { #[tokio::test] async fn test_topk_keeps_the_supported_aggregates() { - for agg in ["count", "avg", "min", "max", "sum"] { + for agg in ["count", "min", "max", "sum"] { assert!(inserts_topk(agg).await, "{agg} must stay eligible"); } } + #[tokio::test] + async fn test_topk_matches_an_aggregate_name_with_brackets() { + let filtered = "COUNT(*) FILTER (WHERE CAST(v AS VARCHAR) IN ('2','5'))"; + assert!(inserts_topk_for(filtered).await, "{filtered}"); + } + #[tokio::test] async fn test_topk_skips_approx_distinct() { assert!( @@ -290,4 +318,141 @@ mod tests { "approx_distinct must stay on the regular aggregation path" ); } + + #[tokio::test] + async fn test_topk_skips_avg() { + assert!( + !inserts_topk("avg").await, + "avg must stay on the regular aggregation path" + ); + } + + #[tokio::test] + async fn test_topk_skips_distinct_with_filter() { + for expr in [ + "COUNT(DISTINCT v) FILTER (WHERE v > 0)", + "SUM(DISTINCT v) FILTER (WHERE v > 0)", + ] { + assert!( + !inserts_topk_for(expr).await, + "{expr} must stay on the regular aggregation path" + ); + } + } + + #[tokio::test] + async fn test_topk_keeps_min_max_distinct_with_filter() { + for expr in [ + "MIN(DISTINCT v) FILTER (WHERE v > 0)", + "MAX(DISTINCT v) FILTER (WHERE v > 0)", + ] { + assert!(inserts_topk_for(expr).await, "{expr} must stay eligible"); + } + } + + #[tokio::test] + async fn test_avg_top_group_with_few_rows_is_not_dropped() { + enable_topk(); + // More groups than the heap keeps (1000), each outranking `top` by row count. + let mut names = vec!["top".to_string()]; + let mut values = vec![100_i64]; + for i in 0..1000 { + names.extend([format!("g{i}"), format!("g{i}")]); + values.extend([1, 1]); + } + let schema = Arc::new(Schema::new(vec![ + Field::new("name", DataType::Utf8, false), + Field::new("v", DataType::Int64, false), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(names)), + Arc::new(Int64Array::from(values)), + ], + ) + .unwrap(); + let ctx = datafusion::prelude::SessionContext::new_with_config( + SessionConfig::new().with_target_partitions(2), + ); + ctx.register_table( + "t", + Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()), + ) + .unwrap(); + let plan = ctx + .sql("SELECT name, avg(v) AS x FROM t GROUP BY name ORDER BY x DESC LIMIT 1") + .await + .unwrap() + .create_physical_plan() + .await + .unwrap(); + let plan = AggregateTopkRule::new(1) + .optimize(plan, &ConfigOptions::default()) + .unwrap(); + let batches = collect(plan, ctx.task_ctx()).await.unwrap(); + let names: Vec = batches + .iter() + .flat_map(|b| { + let col = cast(b.column(0), &DataType::Utf8).unwrap(); + let col = col.as_any().downcast_ref::().unwrap(); + col.iter().flatten().map(String::from).collect::>() + }) + .collect(); + assert_eq!(names, vec!["top".to_string()]); + } + + #[tokio::test] + async fn test_count_distinct_filter_top_group_is_not_dropped() { + enable_topk(); + // Partial state is a value list: `[5]` outranks `[1, 2, 3]` despite a lower count. + let mut names = vec!["top".to_string(); 3]; + let mut values = vec![1_i64, 2, 3]; + for i in 0..1000 { + names.extend([format!("g{i}"), format!("g{i}")]); + values.extend([5, 5]); + } + let schema = Arc::new(Schema::new(vec![ + Field::new("name", DataType::Utf8, false), + Field::new("v", DataType::Int64, false), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(names)), + Arc::new(Int64Array::from(values)), + ], + ) + .unwrap(); + let ctx = datafusion::prelude::SessionContext::new_with_config( + SessionConfig::new().with_target_partitions(2), + ); + ctx.register_table( + "t", + Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()), + ) + .unwrap(); + let sql = "SELECT name, COUNT(DISTINCT v) FILTER (WHERE v > 0) AS x FROM t \ + GROUP BY name ORDER BY x DESC LIMIT 1"; + let plan = ctx + .sql(sql) + .await + .unwrap() + .create_physical_plan() + .await + .unwrap(); + let plan = AggregateTopkRule::new(1) + .optimize(plan, &ConfigOptions::default()) + .unwrap(); + let batches = collect(plan, ctx.task_ctx()).await.unwrap(); + let names: Vec = batches + .iter() + .flat_map(|b| { + let col = cast(b.column(0), &DataType::Utf8).unwrap(); + let col = col.as_any().downcast_ref::().unwrap(); + col.iter().flatten().map(String::from).collect::>() + }) + .collect(); + assert_eq!(names, vec!["top".to_string()]); + } } diff --git a/tests/ui-testing/pages/tracesPages/tracesPage.js b/tests/ui-testing/pages/tracesPages/tracesPage.js index 0ed872d97d..165295ee48 100644 --- a/tests/ui-testing/pages/tracesPages/tracesPage.js +++ b/tests/ui-testing/pages/tracesPages/tracesPage.js @@ -110,20 +110,12 @@ export class TracesPage { this.serviceGraphPage = '[data-test="service-graph-page"]'; // ===== ANALYZE DIMENSIONS SELECTORS (VERIFIED against Vue source) ===== - // TracesMetricsDashboard.vue: data-test="insights-button" this.insightsButton = '[data-test="insights-button"]'; // SearchResult.vue: error-count badge doubles as the error-only toggle this.errorOnlyToggle = '[data-test="traces-error-count-badge"]'; - // Traces SearchBar.vue: data-test="traces-search-bar-show-metrics-toggle-btn" this.metricsToggle = '[data-test="traces-search-bar-show-metrics-toggle-btn"]'; - // TracesAnalysisDashboard.vue was migrated to ODrawer - // (data-test="traces-analysis-dashboard-drawer"). The legacy - // `analysis-dashboard-close` data-test and `.analysis-dashboard-card` - // template class were removed — `.analysis-dashboard-card` only survives - // in CSS rules now, no element actually carries the class. Scope all - // selectors via the ODrawer slug instead. this.analysisDashboardDrawer = '[data-test="traces-analysis-dashboard-drawer"]'; - this.analysisDashboardClose = '[data-test="traces-analysis-dashboard-drawer"] [data-test="o-drawer-close-btn"]'; + this.analysisDashboardClose = '[data-test="traces-analysis-dashboard-drawer"] [data-test="traces-drill-down-back-btn"]'; // TracesAnalysisDashboard.vue: dimension sidebar (visible by default, not a dialog) this.dimensionSelectorSidebar = '[data-test="dimension-selector-sidebar"]'; this.dimensionSelectorCollapseBtn = '[data-test="dimension-selector-collapse-btn"]'; @@ -471,7 +463,9 @@ export class TracesPage { } async toggleMetricsDashboard() { + await this.getMoreMenuButton().click(); await this.page.locator(this.showMetricsToggle).click(); + await this.page.keyboard.press('Escape'); } async switchToServiceMaps() { @@ -2016,12 +2010,11 @@ export class TracesPage { // ===== ANALYZE DIMENSIONS POM METHODS ===== // Selectors verified against actual Vue source code - // --- Insights Button (TracesMetricsDashboard.vue) --- + // --- Drill down button (Traces SearchBar.vue) --- /** - * Check if Insights button is visible. - * The Insights button is ALWAYS visible when metrics dashboard shows - * (does NOT require brush selection). + * Check if the Drill down button is visible. + * It shows in Spans/Traces mode once a search is applied without error. * @returns {Promise} */ async isInsightsButtonVisible() { @@ -2029,7 +2022,7 @@ export class TracesPage { } /** - * Click Insights button to open the Analysis Dashboard + * Click the Drill down button to open the Analysis Dashboard */ async clickInsightsButton() { await this.page.locator(this.insightsButton).click(); @@ -2047,7 +2040,7 @@ export class TracesPage { } /** - * Close Analysis Dashboard via close button + * Close the Drill down view via its Back button */ async closeAnalysisDashboard() { const closeBtn = this.page.locator(this.analysisDashboardClose); diff --git a/web/src/components/alerts/AlertList.spec.ts b/web/src/components/alerts/AlertList.spec.ts index 74e246586d..9cf22a0085 100644 --- a/web/src/components/alerts/AlertList.spec.ts +++ b/web/src/components/alerts/AlertList.spec.ts @@ -1850,3 +1850,112 @@ describe("AlertList - trigger", () => { ]); }); }); + +describe("AlertList - triggered in the last 15 minutes", () => { + const nowUs = () => Date.now() * 1000; + + const loadRows = async () => { + const ranUs = nowUs() - 60_000_000; + const oldUs = nowUs() - 3 * 60 * 60 * 1_000_000; + const scheduled = { is_real_time: false }; + alertsDB = [ + makeAlert(1, { ...scheduled, name: "ran-not-fired", last_triggered_at: ranUs }), + makeAlert(2, { + ...scheduled, + name: "ran-and-fired", + last_triggered_at: ranUs, + last_satisfied_at: ranUs, + }), + makeAlert(3, { + ...scheduled, + name: "ran-fired-long-ago", + last_triggered_at: ranUs, + last_satisfied_at: oldUs, + }), + { + ...makeAlert(4, { ...scheduled, name: "anomaly-ran-not-fired", last_triggered_at: ranUs }), + alert_type: "anomaly_detection", + last_satisfied_at: null, + } as any, + { + ...makeAlert(5, { + ...scheduled, + name: "anomaly-fired", + last_triggered_at: ranUs, + last_satisfied_at: ranUs, + }), + alert_type: "anomaly_detection", + } as any, + { + ...makeAlert(6, { + ...scheduled, + name: "composite-ran-not-fired", + last_triggered_at: ranUs, + }), + alert_type: "composite", + } as any, + { + ...makeAlert(7, { + ...scheduled, + name: "composite-fired", + last_triggered_at: ranUs, + last_satisfied_at: ranUs, + }), + alert_type: "composite", + } as any, + makeAlert(8, { + is_real_time: true, + name: "realtime-fired", + last_triggered_at: ranUs, + last_satisfied_at: ranUs, + }), + makeAlert(9, { + is_real_time: true, + name: "realtime-fired-long-ago", + last_triggered_at: ranUs, + last_satisfied_at: oldUs, + }), + makeAlert(10, { + is_real_time: true, + name: "realtime-woke-not-fired", + last_triggered_at: ranUs, + }), + ]; + const wrapper: any = await mountAlertList(); + await flushPromises(); + await new Promise((resolve) => setTimeout(resolve, 80)); + await flushPromises(); + return wrapper; + }; + + it("counts only alerts that fired recently, not ones that merely ran", async () => { + const wrapper = await loadRows(); + expect(wrapper.vm.filteredResults).toHaveLength(10); + expect(wrapper.vm.stateCounts.recent).toBe(4); + }); + + it("filters the recent state down to alerts that fired recently", async () => { + const wrapper = await loadRows(); + wrapper.vm.onStatSelect("recent"); + await flushPromises(); + expect(wrapper.vm.displayedAlerts.map((r: any) => r.name).sort()).toEqual([ + "anomaly-fired", + "composite-fired", + "ran-and-fired", + "realtime-fired", + ]); + }); + + it("marks the Last Satisfied time of a recent firing, not a Never cell", async () => { + const wrapper = await loadRows(); + // A stale ?action=update from an earlier spec's navigation can swap the table for the editor. + await wrapper.vm.handleActionQuery(undefined); + await flushPromises(); + const cells = wrapper.findAll('[data-test="o2-table-cell-last_satisfied_at"]'); + const never = cells.filter((c: any) => c.text().includes("Never")); + const fired = cells.filter((c: any) => !c.text().includes("Never")); + expect(never.length).toBeGreaterThan(0); + expect(never.every((c: any) => !c.find(".rounded-full").exists())).toBe(true); + expect(fired.filter((c: any) => c.find(".bg-warning-500").exists())).toHaveLength(4); + }); +}); diff --git a/web/src/components/alerts/AlertList.vue b/web/src/components/alerts/AlertList.vue index 4b927cbc07..7778ddb059 100644 --- a/web/src/components/alerts/AlertList.vue +++ b/web/src/components/alerts/AlertList.vue @@ -339,36 +339,36 @@ along with this program. If not, see . + + - - @@ -373,7 +473,7 @@ import { toRef, } from "vue"; import { useQueryPlaceholder } from "@/components/logs/useQueryPlaceholder"; -import { raw, useI18nTyped } from "@/types/i18n"; +import { useI18nTyped } from "@/types/i18n"; import { useRouter } from "vue-router"; import { useStore } from "vuex"; @@ -385,6 +485,10 @@ import OButton from "@/lib/core/Button/OButton.vue"; import OIcon from "@/lib/core/Icon/OIcon.vue"; import ODropdown from "@/lib/overlay/Dropdown/ODropdown.vue"; import ODropdownItem from "@/lib/overlay/Dropdown/ODropdownItem.vue"; +import ODropdownGroup from "@/lib/overlay/Dropdown/ODropdownGroup.vue"; +import ODropdownSeparator from "@/lib/overlay/Dropdown/ODropdownSeparator.vue"; +import ODialog from "@/lib/overlay/Dialog/ODialog.vue"; +import OInput from "@/lib/forms/Input/OInput.vue"; import OSwitch from "@/lib/forms/Switch/OSwitch.vue"; import OSelect from "@/lib/forms/Select/OSelect.vue"; import OTooltip from "@/lib/overlay/Tooltip/OTooltip.vue"; @@ -408,6 +512,16 @@ import { isDatetimeChanged } from "./tracesSearchBar.utils"; import { toast } from "@/lib/feedback/Toast/useToast"; import { downloadFile } from "@/utils/dom"; import { toCsv } from "@/utils/csv"; +import { useMutation, useQuery } from "@tanstack/vue-query"; +import { useOrgId } from "@/composables/query/useOrgId"; +import { useConfirmDialog } from "@/composables/useConfirmDialog"; +import savedViewsService from "@/services/saved_views"; +import { + savedViewsQuery, + createSavedViewMutation, + updateSavedViewMutation, + deleteSavedViewMutation, +} from "@/services/saved_views.queries"; export default defineComponent({ name: "ComponentSearchSearchBar", @@ -421,6 +535,10 @@ export default defineComponent({ OIcon, ODropdown, ODropdownItem, + ODropdownGroup, + ODropdownSeparator, + ODialog, + OInput, OSwitch, OSelect, OTooltip, @@ -436,6 +554,8 @@ export default defineComponent({ "onChangeTimezone", "service-graph-refresh", "services-catalog-refresh", + "apply-saved-view", + "drill-down", ], props: { fieldValues: { @@ -858,6 +978,116 @@ export default defineComponent({ { excludeMatchAll: true }, ); + const orgId = useOrgId(); + const savedViewsList = useQuery(() => ({ + ...savedViewsQuery(orgId.value), + enabled: !!orgId.value, + })); + const tracesSavedViews = computed(() => + (savedViewsList.data.value ?? []).filter((view) => view.view_type === "traces"), + ); + const createSavedView = useMutation(() => createSavedViewMutation(orgId.value)); + const updateSavedView = useMutation(() => updateSavedViewMutation(orgId.value)); + const deleteSavedView = useMutation(() => deleteSavedViewMutation(orgId.value)); + const isSavingView = createSavedView.isPending; + const saveViewDialogOpen = ref(false); + const newViewName = ref(""); + const { confirm } = useConfirmDialog(); + + const buildSavedViewData = () => { + const { datetime, stream } = searchObj.data; + return { + version: 1, + stream: { label: stream.selectedStream.label, value: stream.selectedStream.value }, + editorValue: searchObj.data.editorValue, + datetime: { + type: datetime.type, + relativeTimePeriod: datetime.relativeTimePeriod, + startTime: datetime.startTime, + endTime: datetime.endTime, + }, + searchMode: searchObj.meta.searchMode, + sortBy: searchObj.meta.resultGrid.sortBy, + sortOrder: searchObj.meta.resultGrid.sortOrder, + selectedFields: [...stream.selectedFields], + }; + }; + + const errorDetail = (err) => err?.response?.data?.message ?? err?.message ?? ""; + + const openSaveViewDialog = () => { + newViewName.value = ""; + saveViewDialogOpen.value = true; + }; + + const saveAsNewView = async () => { + const viewName = newViewName.value.trim(); + if (!viewName) { + toast({ message: t("logs.searchBar.provideValidViewName"), variant: "warning" }); + return; + } + try { + await createSavedView.mutateAsync({ + view_name: viewName, + view_type: "traces", + data: buildSavedViewData(), + }); + saveViewDialogOpen.value = false; + toast({ message: t("search.viewCreatedSuccessfully"), variant: "success" }); + } catch (err) { + toast({ + message: t("search.errorCreatingSavedViewDetail", { detail: errorDetail(err) }), + variant: "error", + }); + } + }; + + const updateSavedViewFromCurrent = async (view) => { + try { + await updateSavedView.mutateAsync({ + viewId: view.view_id, + view: { view_name: view.view_name, data: buildSavedViewData() }, + }); + toast({ message: t("search.viewUpdatedSuccessfully"), variant: "success" }); + } catch (err) { + toast({ + message: t("search.errorUpdatingSavedViewDetail", { detail: errorDetail(err) }), + variant: "error", + }); + } + }; + + const deleteTracesSavedView = async (view) => { + const confirmed = await confirm({ + title: t("search.deleteSavedView"), + message: t("search.deleteSavedViewConfirm"), + }); + if (!confirmed) return; + try { + await deleteSavedView.mutateAsync(view.view_id); + toast({ message: t("search.viewDeletedSuccessfully"), variant: "success" }); + } catch (err) { + toast({ + message: t("search.errorDeletingSavedViewDetail", { detail: errorDetail(err) }), + variant: "error", + }); + } + }; + + let applySavedViewSeq = 0; + const applySavedView = async (view) => { + const seq = ++applySavedViewSeq; + try { + // Uncached on purpose, as in logs: applying must use the server's current copy. + const res = await savedViewsService.getViewDetail(orgId.value, view.view_id); + if (seq !== applySavedViewSeq) return; + emit("apply-saved-view", res.data.data); + } catch (err) { + if (seq !== applySavedViewSeq) return; + toast({ message: t("search.errorWhileApplyingSavedView"), variant: "error" }); + } + }; + // Responsive toolbar — shared composable tracks available left-section width const { toolbarLeftRef, toolbarRightRef, availableLeftWidth } = useToolbarResponsive(); @@ -867,9 +1097,21 @@ export default defineComponent({ const shouldHideToggleText = computed(() => availableLeftWidth.value < 750); const shouldHideResetText = computed(() => availableLeftWidth.value < 540); + // Only while Index renders SearchResult with its header, since the click is routed to it. + const showDrillDown = computed( + () => + searchObj.meta.searchMode !== "service-graph" && + searchObj.meta.searchMode !== "services-catalog" && + !searchObj.loadingStream && + !searchObj.loading && + searchObj.data.stream.streamLists.length > 0 && + !!searchObj.data.stream.selectedStream?.value && + !searchObj.data.errorMsg?.trim()?.length && + !!searchObj.searchApplied, + ); + return { t, - raw, router, store, searchObj, @@ -906,6 +1148,16 @@ export default defineComponent({ toolbarRightRef, shouldHideToggleText, shouldHideResetText, + showDrillDown, + tracesSavedViews, + isSavingView, + saveViewDialogOpen, + newViewName, + openSaveViewDialog, + saveAsNewView, + updateSavedViewFromCurrent, + deleteTracesSavedView, + applySavedView, }; }, computed: { diff --git a/web/src/plugins/traces/SearchResult.vue b/web/src/plugins/traces/SearchResult.vue index b0b15e4439..81c8fb11ac 100644 --- a/web/src/plugins/traces/SearchResult.vue +++ b/web/src/plugins/traces/SearchResult.vue @@ -19,14 +19,12 @@ along with this program. If not, see .