diff --git a/src/api/http/src/handler/http/router/mod.rs b/src/api/http/src/handler/http/router/mod.rs index cf99e180e3..df3633e1c7 100644 --- a/src/api/http/src/handler/http/router/mod.rs +++ b/src/api/http/src/handler/http/router/mod.rs @@ -1344,6 +1344,7 @@ pub fn service_routes() -> Router { // Org info .route("/{org_id}/summary", get(organization::org::org_summary)) + .route("/{org_id}/overview", get(organization::org::org_overview)) .route("/{org_id}/ingest/recent_rejections", get(ingest_rejections::recent_rejections)) .route("/{org_id}/passcode", get(organization::org::get_user_passcode).put(organization::org::update_user_passcode)) .route("/{org_id}/rumtoken", get(organization::org::get_user_rumtoken).post(organization::org::create_user_rumtoken).put(organization::org::update_user_rumtoken)) diff --git a/src/api/http/src/handler/http/router/openapi.rs b/src/api/http/src/handler/http/router/openapi.rs index 86c21bb9f1..d0d53ecb7e 100644 --- a/src/api/http/src/handler/http/router/openapi.rs +++ b/src/api/http/src/handler/http/router/openapi.rs @@ -47,6 +47,7 @@ use crate::{ openobserve_api_management::request::organization::org::rename_org, openobserve_api_management::request::organization::assume_service_account::assume_service_account, openobserve_api_management::request::organization::org::org_summary, + openobserve_api_management::request::organization::org::org_overview, openobserve_api_management::request::organization::org::get_user_passcode, openobserve_api_management::request::organization::org::update_user_passcode, openobserve_api_management::request::organization::org::get_user_rumtoken, diff --git a/src/api/management/src/request/organization/org.rs b/src/api/management/src/request/organization/org.rs index cb9d93ccd0..125516d175 100644 --- a/src/api/management/src/request/organization/org.rs +++ b/src/api/management/src/request/organization/org.rs @@ -28,6 +28,7 @@ use openobserve_api_common::extractors::Headers; use openobserve_core::{ auth::{UserEmail, check_permissions}, organization::{self, get_passcode, get_rum_token, update_passcode, update_rum_token}, + overview, }; #[cfg(feature = "cloud")] use { @@ -384,6 +385,51 @@ pub async fn org_summary(Path(org_id): Path) -> impl IntoResponse { Json(org_summary) } +/// GetOrganizationOverview + +#[utoipa::path( + get, + path = "/{org_id}/overview", + context_path = "/api", + tag = "Organizations", + operation_id = "GetOrganizationOverview", + summary = "Get organization overview", + description = "Returns the streams, alerts, pipelines and license limits sections the Overview page needs in one call. Each section carries its own status (ok, error or locked), so one failing section never fails the response, and results only include objects the caller may see. Responses are cached briefly per caller.", + security( + ("Authorization"= []) + ), + params( + ("org_id" = String, Path, description = "Organization name"), + ("top" = Option, Query, description = "Top streams per stream type by size. Default 5, max 20."), + ), + responses( + (status = 200, description = "Success", content_type = "application/json", body = Object), + (status = 400, description = "Failure", content_type = "application/json", body = ()), + (status = 403, description = "Forbidden", content_type = "application/json", body = ()), + ), + extensions( + ("x-o2-ratelimit" = json!({"module": "Summary", "operation": "get"})), + ("x-o2-mcp" = json!({"enabled": false})) + ) +)] +pub async fn org_overview( + Path(org_id): Path, + Headers(user_email): Headers, + Query(query): Query>, +) -> Response { + let top = match query.get("top").map(|v| v.parse::()) { + Some(Err(_)) => return MetaHttpResponse::bad_request("'top' must be a positive integer"), + Some(Ok(n)) => Some(n), + None => None, + }; + let scope = match overview::resolve_scope(&org_id, &user_email.user_id).await { + Ok(scope) => scope, + Err(e) => return MetaHttpResponse::forbidden(e.to_string()), + }; + let result = overview::get_overview(&org_id, &scope, overview::normalize_top_n(top)).await; + Json(result).into_response() +} + /// GetIngestToken #[utoipa::path( diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 41dd45242b..bcee22b7c9 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -2863,6 +2863,12 @@ pub struct Limit { pub req_cols_per_record_limit: usize, #[env_config(name = "ZO_NODE_HEARTBEAT_TTL", default = 30)] // seconds pub node_heartbeat_ttl: i64, + #[env_config( + name = "ZO_OVERVIEW_CACHE_TTL", + default = 30, + help = "Seconds the org overview response is cached per caller; 0 disables the cache." + )] + pub overview_cache_ttl: u64, // How long an o2-ai session->owner claim survives. Must exceed the longest // expected conversation; mirrors o2-ai's O2_AI_SESSION_OWNER_TTL default. #[env_config(name = "ZO_AI_SESSION_OWNER_TTL", default = 86400)] // seconds diff --git a/src/config/src/router.rs b/src/config/src/router.rs index d2c9ad5e9c..8e1e2f79e8 100644 --- a/src/config/src/router.rs +++ b/src/config/src/router.rs @@ -14,10 +14,11 @@ // along with this program. If not, see . /// usize indicates the number of parts to skip based on their actual paths. -const QUERIER_ROUTES: [(&str, usize); 47] = [ +const QUERIER_ROUTES: [(&str, usize); 48] = [ ("config", 0), // /config (unauthenticated bootstrap) ("config", 2), // /api/{org_id}/config (authenticated full) ("summary", 2), // /api/{org_id}/summary + ("overview", 2), // /api/{org_id}/overview ("organizations", 1), // /api/organizations ("settings", 2), // /api/{org_id}/settings/... ("schema", 3), // /api/{org_id}/streams/{stream_name}/schema @@ -205,6 +206,9 @@ mod tests { assert!(is_querier_route("/api/org1/summary")); assert!(!is_querier_route("/summary")); // Should fail as it needs org_id + assert!(is_querier_route("/api/org1/overview")); + assert!(!is_querier_route("/overview")); + // Test streams route assert!(is_querier_route("/api/org1/streams")); assert!(is_querier_route("/api/org1/streams/mystream")); diff --git a/src/core/src/lib.rs b/src/core/src/lib.rs index 02ada8c213..063da748e5 100644 --- a/src/core/src/lib.rs +++ b/src/core/src/lib.rs @@ -57,6 +57,7 @@ pub mod org_cleanup; #[cfg(feature = "cloud")] pub mod org_usage; pub mod organization; +pub mod overview; pub mod pipeline; pub mod profiles; pub mod prompts; diff --git a/src/core/src/overview.rs b/src/core/src/overview.rs new file mode 100644 index 0000000000..c8570dc293 --- /dev/null +++ b/src/core/src/overview.rs @@ -0,0 +1,691 @@ +// 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 common::meta::stream::Stream; +use config::{ + get_config, + meta::{ + alerts::{ + alert::{Alert, ListAlertsParams}, + state::AlertState, + }, + pipeline::Pipeline, + self_reporting::ai_chat::is_protected_ai_chat_stream, + stream::StreamType, + }, + utils::time::now_micros, +}; +use dashmap::DashMap; +use infra::db::get_orm_client_ro; +use serde::Serialize; +use stream::try_get_streams; + +pub const DEFAULT_TOP_N: usize = 5; +pub const MAX_TOP_N: usize = 20; +/// Stream types the overview reports; index and metadata streams are internal. +pub const OVERVIEW_STREAM_TYPES: [StreamType; 4] = [ + StreamType::Logs, + StreamType::Metrics, + StreamType::Traces, + StreamType::EnrichmentTables, +]; + +const ALERT_ID_CHUNK: usize = 500; +const CACHE_MAX_ENTRIES: usize = 1024; +const ALERT_STATE_KEYS: [&str; 9] = [ + "firing", + "pending", + "normal", + "succeeded", + "error", + "skipped", + "notify_failed", + "suppressed", + "never_evaluated", +]; +const ALERT_LEVEL_KEYS: [&str; 5] = ["ok", "warning", "critical", "no_data", "unclassified"]; + +/// Entries are keyed per caller so one user's visible set is never served to another. +static CACHE: LazyLock> = LazyLock::new(DashMap::new); + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct CacheKey { + org_id: String, + user_id: String, + top_n: usize, +} + +/// What the caller may see; a stream type or pipeline list that is absent is unrestricted. +#[derive(Clone, Debug, Default)] +pub struct OverviewScope { + pub user_id: String, + pub stream_grants: HashMap>, + pub pipeline_grants: Option>, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum SectionStatus { + Ok, + Error, + Locked, +} + +#[derive(Clone, Debug, Serialize)] +pub struct Section { + pub status: SectionStatus, + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +impl Section { + pub fn ok(data: T) -> Self { + Self { + status: SectionStatus::Ok, + data: Some(data), + error: None, + } + } + + pub fn error(message: &str) -> Self { + Self { + status: SectionStatus::Error, + data: None, + error: Some(message.to_string()), + } + } + + pub fn locked() -> Self { + Self { + status: SectionStatus::Locked, + data: None, + error: None, + } + } + + fn from_result(result: Result, what: &str) -> Self { + match result { + Ok(data) => Self::ok(data), + Err(e) => { + log::error!("[overview] failed to load {what}: {e}"); + Self::error(&format!("failed to load {what}")) + } + } + } +} + +#[derive(Clone, Debug, Serialize)] +pub struct Overview { + pub streams: Section, + pub alerts: Section, + pub pipelines: Section, + pub limits: Section, + /// Microseconds since the epoch. + pub generated_at: i64, + pub cached: bool, +} + +impl Overview { + fn has_error(&self) -> bool { + self.streams.status == SectionStatus::Error + || self.alerts.status == SectionStatus::Error + || self.pipelines.status == SectionStatus::Error + || self.limits.status == SectionStatus::Error + } +} + +#[derive(Clone, Debug, Serialize)] +pub struct StreamsData { + pub total_streams: u64, + pub total_storage_size_mb: f64, + pub by_type: Vec, +} + +#[derive(Clone, Debug, Serialize)] +pub struct StreamTypeSummary { + pub stream_type: String, + pub count: u64, + pub doc_num: i64, + pub storage_size_mb: f64, + pub compressed_size_mb: f64, + pub top: Vec, +} + +#[derive(Clone, Debug, Serialize)] +pub struct TopStream { + pub name: String, + pub doc_num: i64, + pub storage_size_mb: f64, +} + +#[derive(Clone, Debug, Serialize)] +pub struct AlertsData { + pub total: u64, + pub enabled: u64, + pub disabled: u64, + /// Enabled rules by the outcome of their last evaluation. + pub by_state: BTreeMap, + /// Enabled rules by the severity level of their last classification. + pub by_level: BTreeMap, + pub failed_evaluation: u64, +} + +#[derive(Clone, Debug, Serialize)] +pub struct PipelinesData { + pub total: u64, + pub enabled: u64, + pub disabled: u64, + pub errored: u64, +} + +/// Deployment-wide license ingestion figures, not per org. +#[derive(Clone, Debug, Serialize)] +pub struct LimitsData { + pub scope: &'static str, + pub ingestion_used_percent: f64, + pub days_over_limit: u8, +} + +/// Clamps the requested top-N to `1..=MAX_TOP_N`, defaulting to `DEFAULT_TOP_N`. +pub fn normalize_top_n(requested: Option) -> usize { + requested.unwrap_or(DEFAULT_TOP_N).clamp(1, MAX_TOP_N) +} + +/// Resolves which streams and pipelines the caller may see; errors when the grant lookup fails. +pub async fn resolve_scope(org_id: &str, user_id: &str) -> Result { + let mut stream_grants = HashMap::new(); + for stream_type in OVERVIEW_STREAM_TYPES { + let grants = crate::authz::list_objects_for_user( + org_id, + user_id, + "GET", + &ofga_object_type(&stream_type.to_string()), + ) + .await?; + if let Some(grants) = grants { + stream_grants.insert(stream_type, grants); + } + } + let pipeline_grants = + crate::authz::list_objects_for_user(org_id, user_id, "GET", &ofga_object_type("pipelines")) + .await?; + Ok(OverviewScope { + user_id: user_id.to_string(), + stream_grants, + pipeline_grants, + }) +} + +/// Returns the overview for the caller, from the short-lived per-caller cache when fresh. +pub async fn get_overview(org_id: &str, scope: &OverviewScope, top_n: usize) -> Overview { + let ttl_micros = get_config().limit.overview_cache_ttl as i64 * 1_000_000; + let key = CacheKey { + org_id: org_id.to_string(), + user_id: scope.user_id.clone(), + top_n, + }; + if ttl_micros > 0 + && let Some(hit) = CACHE.get(&key) + && now_micros() - hit.generated_at < ttl_micros + { + let mut overview = hit.clone(); + overview.cached = true; + return overview; + } + + let overview = build_overview(org_id, scope, top_n).await; + if ttl_micros > 0 && !overview.has_error() { + cache_put(key, overview.clone(), ttl_micros); + } + overview +} + +async fn build_overview(org_id: &str, scope: &OverviewScope, top_n: usize) -> Overview { + let (streams, alerts, pipelines) = tokio::join!( + load_streams(org_id, scope, top_n), + load_alerts(org_id, &scope.user_id), + load_pipelines(org_id, scope.pipeline_grants.clone()), + ); + Overview { + streams: Section::from_result(streams, "streams"), + alerts: Section::from_result(alerts, "alerts"), + pipelines: Section::from_result(pipelines, "pipelines"), + limits: limits_section().await, + generated_at: now_micros(), + cached: false, + } +} + +fn cache_put(key: CacheKey, overview: Overview, ttl_micros: i64) { + if CACHE.len() >= CACHE_MAX_ENTRIES { + let now = now_micros(); + CACHE.retain(|_, v| now - v.generated_at < ttl_micros); + if CACHE.len() >= CACHE_MAX_ENTRIES { + CACHE.clear(); + } + } + CACHE.insert(key, overview); +} + +#[cfg(feature = "enterprise")] +fn ofga_object_type(name: &str) -> String { + use o2_openfga::meta::mapping::OFGA_MODELS; + + OFGA_MODELS + .get(name) + .map_or(name, |model| model.key) + .to_string() +} + +#[cfg(not(feature = "enterprise"))] +fn ofga_object_type(name: &str) -> String { + name.to_string() +} + +async fn load_streams( + org_id: &str, + scope: &OverviewScope, + top_n: usize, +) -> Result { + let mut streams = Vec::new(); + for stream_type in OVERVIEW_STREAM_TYPES { + let grants = scope.stream_grants.get(&stream_type).cloned(); + streams.extend(try_get_streams(org_id, Some(stream_type), false, grants).await?); + } + Ok(build_streams_data(streams, top_n)) +} + +fn build_streams_data(streams: Vec, top_n: usize) -> StreamsData { + let mut by_type = Vec::with_capacity(OVERVIEW_STREAM_TYPES.len()); + for stream_type in OVERVIEW_STREAM_TYPES { + let mut of_type: Vec<&Stream> = streams + .iter() + .filter(|s| s.stream_type == stream_type && !is_protected_ai_chat_stream(&s.name)) + .collect(); + of_type.sort_by(|a, b| { + b.stats + .storage_size + .total_cmp(&a.stats.storage_size) + .then_with(|| a.name.cmp(&b.name)) + }); + by_type.push(StreamTypeSummary { + stream_type: stream_type.to_string(), + count: of_type.len() as u64, + doc_num: of_type.iter().map(|s| s.stats.doc_num).sum(), + storage_size_mb: of_type.iter().map(|s| s.stats.storage_size).sum(), + compressed_size_mb: of_type.iter().map(|s| s.stats.compressed_size).sum(), + top: of_type + .iter() + .take(top_n) + .map(|s| TopStream { + name: s.name.clone(), + doc_num: s.stats.doc_num, + storage_size_mb: s.stats.storage_size, + }) + .collect(), + }); + } + StreamsData { + total_streams: by_type.iter().map(|t| t.count).sum(), + total_storage_size_mb: by_type.iter().map(|t| t.storage_size_mb).sum(), + by_type, + } +} + +async fn load_alerts(org_id: &str, user_id: &str) -> Result { + let conn = get_orm_client_ro().await; + let alerts: Vec = + crate::alerts::alert::list_v2(conn, Some(user_id), ListAlertsParams::new(org_id)) + .await? + .into_iter() + .map(|(_, alert)| alert) + .collect(); + let ids: Vec = alerts + .iter() + .filter(|a| a.enabled) + .filter_map(|a| a.id.map(|id| id.to_string())) + .collect(); + let mut states = Vec::with_capacity(ids.len()); + for chunk in ids.chunks(ALERT_ID_CHUNK) { + states.extend(infra::table::alert_states::get_rollups(chunk).await?); + } + Ok(build_alerts_data(&alerts, &states)) +} + +fn build_alerts_data(alerts: &[Alert], states: &[AlertState]) -> AlertsData { + let by_id: HashMap = + states.iter().map(|s| (s.alert_id.clone(), s)).collect(); + let mut by_state: BTreeMap = ALERT_STATE_KEYS + .iter() + .map(|k| (k.to_string(), 0)) + .collect(); + let mut by_level: BTreeMap = ALERT_LEVEL_KEYS + .iter() + .map(|k| (k.to_string(), 0)) + .collect(); + let mut enabled = 0; + let mut failed_evaluation = 0; + for alert in alerts.iter().filter(|a| a.enabled) { + enabled += 1; + let state = alert.id.and_then(|id| by_id.get(&id.to_string()).copied()); + let outcome = state.and_then(|s| s.last_outcome.as_ref()); + let state_key = outcome.map_or("never_evaluated".to_string(), |o| o.to_string()); + *by_state.entry(state_key).or_insert(0) += 1; + let level_key = state + .and_then(|s| s.level) + .map_or("unclassified".to_string(), |l| l.to_string()); + *by_level.entry(level_key).or_insert(0) += 1; + if outcome.is_some_and(|o| o.as_str() == "error") { + failed_evaluation += 1; + } + } + AlertsData { + total: alerts.len() as u64, + enabled, + disabled: alerts.len() as u64 - enabled, + by_state, + by_level, + failed_evaluation, + } +} + +async fn load_pipelines( + org_id: &str, + grants: Option>, +) -> Result { + let pipelines = crate::pipeline::list_user_pipelines(org_id, grants).await?; + let errored: HashSet = db::pipeline_errors::list_by_org(org_id) + .await? + .into_iter() + .map(|e| e.pipeline_id) + .collect(); + Ok(build_pipelines_data(&pipelines, &errored)) +} + +fn build_pipelines_data(pipelines: &[Pipeline], errored: &HashSet) -> PipelinesData { + let enabled = pipelines.iter().filter(|p| p.enabled).count() as u64; + PipelinesData { + total: pipelines.len() as u64, + enabled, + disabled: pipelines.len() as u64 - enabled, + errored: pipelines.iter().filter(|p| errored.contains(&p.id)).count() as u64, + } +} + +#[cfg(feature = "enterprise")] +async fn limits_section() -> Section { + use o2_enterprise::enterprise::license::{ + get_license, ingestion_limit_exceeded_count, ingestion_used, + }; + + if get_license().await.is_none() { + return Section::locked(); + } + Section::ok(LimitsData { + scope: "deployment", + ingestion_used_percent: ingestion_used() * 100.0, + days_over_limit: ingestion_limit_exceeded_count(), + }) +} + +#[cfg(not(feature = "enterprise"))] +async fn limits_section() -> Section { + Section::locked() +} + +#[cfg(test)] +mod tests { + use arrow_schema::{DataType, Field, Schema}; + use config::meta::{ + alerts::level::AlertLevel, pipeline::Pipeline, self_reporting::usage::RunOutcome, + stream::StreamStats, + }; + use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache, insert_latest}; + + use super::*; + + fn stream(name: &str, stream_type: StreamType, size: f64, docs: i64) -> Stream { + Stream { + name: name.to_string(), + storage_type: "disk".to_string(), + stream_type, + stats: StreamStats { + storage_size: size, + compressed_size: size / 2.0, + doc_num: docs, + ..Default::default() + }, + schema: vec![], + uds_schema: vec![], + settings: Default::default(), + metrics_meta: None, + total_fields: 0, + pattern_associations: vec![], + is_derived: None, + } + } + + fn alert(id: &str, enabled: bool) -> Alert { + serde_json::from_value(serde_json::json!({"id": id, "enabled": enabled})).unwrap() + } + + fn state(alert: &Alert, outcome: Option, level: Option) -> AlertState { + let mut s = AlertState::empty(&alert.id.unwrap().to_string(), ""); + s.last_outcome = outcome; + s.level = level; + s + } + + fn pipeline(id: &str, enabled: bool) -> Pipeline { + Pipeline { + id: id.to_string(), + version: 0, + enabled, + org: String::new(), + name: id.to_string(), + description: String::new(), + source: Default::default(), + kind: Default::default(), + nodes: vec![], + edges: vec![], + } + } + + #[test] + fn test_normalize_top_n() { + assert_eq!(normalize_top_n(None), DEFAULT_TOP_N); + assert_eq!(normalize_top_n(Some(0)), 1); + assert_eq!(normalize_top_n(Some(7)), 7); + assert_eq!(normalize_top_n(Some(500)), MAX_TOP_N); + } + + #[test] + fn test_streams_section_totals_and_top_n_per_type() { + let streams = vec![ + stream("small", StreamType::Logs, 1.0, 10), + stream("big", StreamType::Logs, 9.0, 90), + stream("mid", StreamType::Logs, 5.0, 50), + stream("m1", StreamType::Metrics, 2.0, 20), + stream("internal", StreamType::Index, 100.0, 1), + ]; + let data = build_streams_data(streams, 2); + assert_eq!(data.total_streams, 4); + assert_eq!(data.total_storage_size_mb, 17.0); + let logs = &data.by_type[0]; + assert_eq!(logs.stream_type, "logs"); + assert_eq!(logs.count, 3); + assert_eq!(logs.doc_num, 150); + let top: Vec<&str> = logs.top.iter().map(|t| t.name.as_str()).collect(); + assert_eq!(top, vec!["big", "mid"]); + assert_eq!(data.by_type[2].count, 0); + } + + #[test] + fn test_alerts_section_counts_by_state_level_and_failures() { + let a1 = alert("2Z8rVTbWbyUjb3hIibV02e0sqHW", true); + let a2 = alert("2Z8rVTbWbyUjb3hIibV02e0sqHX", true); + let a3 = alert("2Z8rVTbWbyUjb3hIibV02e0sqHY", true); + let a4 = alert("2Z8rVTbWbyUjb3hIibV02e0sqHZ", false); + let states = vec![ + state(&a1, Some(RunOutcome::Firing), Some(AlertLevel::Critical)), + state(&a2, Some(RunOutcome::Error), None), + state(&a4, Some(RunOutcome::Firing), Some(AlertLevel::Warning)), + ]; + let data = build_alerts_data(&[a1, a2, a3, a4], &states); + assert_eq!((data.total, data.enabled, data.disabled), (4, 3, 1)); + assert_eq!(data.by_state["firing"], 1); + assert_eq!(data.by_state["error"], 1); + assert_eq!(data.by_state["never_evaluated"], 1); + assert_eq!(data.by_level["critical"], 1); + assert_eq!(data.by_level["unclassified"], 2); + assert_eq!(data.failed_evaluation, 1); + } + + #[test] + fn test_pipelines_section_counts_errored_among_given_pipelines() { + let pipelines = vec![ + pipeline("p1", true), + pipeline("p2", false), + pipeline("p3", true), + ]; + let errored: HashSet = ["p2".to_string(), "hidden".to_string()].into(); + let data = build_pipelines_data(&pipelines, &errored); + assert_eq!( + (data.total, data.enabled, data.disabled, data.errored), + (3, 2, 1, 1) + ); + } + + #[test] + fn test_section_serialization_carries_status_and_omits_empty_fields() { + let locked = serde_json::to_value(Section::::locked()).unwrap(); + assert_eq!(locked, serde_json::json!({"status": "locked"})); + let err = serde_json::to_value(Section::::error("boom")).unwrap(); + assert_eq!(err, serde_json::json!({"status": "error", "error": "boom"})); + } + + #[tokio::test] + async fn test_limits_locked_without_enterprise() { + #[cfg(not(feature = "enterprise"))] + assert_eq!(limits_section().await.status, SectionStatus::Locked); + } + + async fn seed_stream(org: &str, stream_type: StreamType, name: &str, size: f64) { + let schema = Schema::new(vec![Field::new("f", DataType::Utf8, true)]); + { + let mut w = STREAM_SCHEMAS_LATEST.write().await; + insert_latest( + &mut w, + format!("{org}/{stream_type}/{name}"), + SchemaCache::new(schema), + ); + } + infra::cache::stats::set_stream_stats( + org, + name, + stream_type, + StreamStats { + storage_size: size * config::SIZE_IN_MB, + compressed_size: size * config::SIZE_IN_MB, + doc_num: 1, + ..Default::default() + }, + ); + } + + #[tokio::test] + async fn test_restricted_caller_never_sees_hidden_streams() { + let org = "overview_rbac_org"; + seed_stream(org, StreamType::Logs, "visible", 1.0).await; + seed_stream(org, StreamType::Logs, "secret", 1000.0).await; + seed_stream(org, StreamType::Metrics, "secret_metric", 500.0).await; + + let restricted = OverviewScope { + user_id: "limited@example.com".to_string(), + stream_grants: HashMap::from([ + (StreamType::Logs, vec!["logs:visible".to_string()]), + (StreamType::Metrics, vec![]), + ]), + pipeline_grants: None, + }; + let data = load_streams(org, &restricted, 5).await.unwrap(); + assert_eq!(data.total_streams, 1); + assert_eq!(data.total_storage_size_mb, 1.0); + let logs = &data.by_type[0]; + assert_eq!((logs.count, logs.storage_size_mb), (1, 1.0)); + let names: Vec<&str> = logs.top.iter().map(|t| t.name.as_str()).collect(); + assert_eq!(names, vec!["visible"]); + assert_eq!(data.by_type[1].count, 0); + assert!(data.by_type[1].top.is_empty()); + + let unrestricted = OverviewScope { + user_id: "admin@example.com".to_string(), + ..Default::default() + }; + let data = load_streams(org, &unrestricted, 5).await.unwrap(); + assert_eq!(data.total_streams, 3); + assert_eq!(data.total_storage_size_mb, 1501.0); + } + + #[tokio::test] + async fn test_cache_is_keyed_per_caller_and_marks_hits() { + let org = "overview_cache_org"; + seed_stream(org, StreamType::Logs, "only_for_admin", 7.0).await; + let admin = OverviewScope { + user_id: "admin@example.com".to_string(), + ..Default::default() + }; + let limited = OverviewScope { + user_id: "limited@example.com".to_string(), + stream_grants: HashMap::from([(StreamType::Logs, vec![])]), + pipeline_grants: None, + }; + let key = CacheKey { + org_id: org.to_string(), + user_id: admin.user_id.clone(), + top_n: 5, + }; + let seeded = Overview { + streams: Section::ok(build_streams_data( + vec![stream("only_for_admin", StreamType::Logs, 7.0, 1)], + 5, + )), + alerts: Section::locked(), + pipelines: Section::locked(), + limits: Section::locked(), + generated_at: now_micros(), + cached: false, + }; + CACHE.insert(key, seeded); + + let hit = get_overview(org, &admin, 5).await; + assert!(hit.cached); + assert_eq!(hit.streams.data.unwrap().total_streams, 1); + + let limited_key = CacheKey { + org_id: org.to_string(), + user_id: limited.user_id.clone(), + top_n: 5, + }; + assert!(CACHE.get(&limited_key).is_none()); + } +} diff --git a/src/stream/src/lib.rs b/src/stream/src/lib.rs index 92a7cd41a9..8fbcf3b2f5 100644 --- a/src/stream/src/lib.rs +++ b/src/stream/src/lib.rs @@ -93,33 +93,28 @@ pub async fn get_streams( fetch_schema: bool, permitted_streams: Option>, ) -> Vec { - let indices = db::schema::list(org_id, stream_type, fetch_schema) + try_get_streams(org_id, stream_type, fetch_schema, permitted_streams) .await - .unwrap_or_default(); + .unwrap_or_default() +} - let filtered_indices = if let Some(s_type) = stream_type { - let s_type = match s_type { - StreamType::EnrichmentTables => "enrichment_table", - _ => s_type.as_str(), - }; - match permitted_streams { - Some(permitted_streams) => { - if permitted_streams.contains(&format!("{s_type}:_all_{org_id}")) { - indices - } else { - indices - .into_iter() - .filter(|stream_loc| { - permitted_streams - .contains(&format!("{}:{}", s_type, stream_loc.stream_name)) - }) - .collect::>() - } - } - None => indices, - } - } else { - indices +/// Like [`get_streams`] but reports a failed schema listing instead of returning an empty list. +pub async fn try_get_streams( + org_id: &str, + stream_type: Option, + fetch_schema: bool, + permitted_streams: Option>, +) -> Result, anyhow::Error> { + let indices = db::schema::list(org_id, stream_type, fetch_schema).await?; + + let filtered_indices = match permitted_streams { + Some(permitted) => indices + .into_iter() + .filter(|loc| { + is_stream_permitted(&permitted, org_id, loc.stream_type, &loc.stream_name) + }) + .collect::>(), + None => indices, }; let mut indices_res = Vec::with_capacity(filtered_indices.len()); for stream_loc in filtered_indices { @@ -155,7 +150,22 @@ pub async fn get_streams( )); } } - indices_res + Ok(indices_res) +} + +/// Checks a stream against the caller's permitted object list, using the stream's own type. +fn is_stream_permitted( + permitted: &[String], + org_id: &str, + stream_type: StreamType, + stream_name: &str, +) -> bool { + let s_type = match stream_type { + StreamType::EnrichmentTables => "enrichment_table", + _ => stream_type.as_str(), + }; + permitted.contains(&format!("{s_type}:_all_{org_id}")) + || permitted.contains(&format!("{s_type}:{stream_name}")) } pub fn stream_res( @@ -1208,6 +1218,54 @@ mod tests { // unique org so the process-wide schema caches can host these tests in parallel const SETTINGS_TEST_ORG: &str = "settings_cache_test_org"; + #[test] + fn test_is_stream_permitted_uses_each_streams_own_type() { + let permitted = vec!["logs:app".to_string(), "metrics:cpu".to_string()]; + assert!(is_stream_permitted( + &permitted, + "o", + StreamType::Logs, + "app" + )); + assert!(is_stream_permitted( + &permitted, + "o", + StreamType::Metrics, + "cpu" + )); + assert!(!is_stream_permitted( + &permitted, + "o", + StreamType::Logs, + "cpu" + )); + assert!(!is_stream_permitted( + &permitted, + "o", + StreamType::Traces, + "app" + )); + } + + #[test] + fn test_is_stream_permitted_all_grant_and_enrichment_key() { + let all = vec!["logs:_all_o".to_string()]; + assert!(is_stream_permitted(&all, "o", StreamType::Logs, "anything")); + assert!(!is_stream_permitted( + &all, + "o", + StreamType::Metrics, + "anything" + )); + let enrich = vec!["enrichment_table:geo".to_string()]; + assert!(is_stream_permitted( + &enrich, + "o", + StreamType::EnrichmentTables, + "geo" + )); + } + #[test] fn test_associated_metadata_streams() { assert_eq!(