From 08a66167c19f61c1b97852d3ce32c75a7ae36844 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Thu, 15 Oct 2026 11:44:17 +0000 Subject: [PATCH] chore: remove ZO_USAGE_REPORT_TO_OWN_ORG and always report to each org's own streams (#15406) --- src/common/src/meta/organization.rs | 8 +- src/config/src/config.rs | 30 ------- src/core/src/self_reporting/ingestion.rs | 86 ++++++++++----------- src/core/src/self_reporting/persistence.rs | 3 +- src/core/src/self_reporting/usage_schema.rs | 9 --- src/db/src/organization.rs | 7 +- src/usage_reporting/src/lib.rs | 3 +- 7 files changed, 45 insertions(+), 101 deletions(-) diff --git a/src/common/src/meta/organization.rs b/src/common/src/meta/organization.rs index 3703ec4e35..34a9450ded 100644 --- a/src/common/src/meta/organization.rs +++ b/src/common/src/meta/organization.rs @@ -470,10 +470,6 @@ fn default_enable_streaming_search() -> bool { /// absent setting means the per-org `usage` stream is not written and the /// Usage tab shows only the billing-API cards plus an "Enable usage reporting" /// CTA. -/// -/// The global `ZO_USAGE_REPORT_TO_OWN_ORG` remains the deploy-wide master -/// switch for whether per-org self-reporting is available at all; it no longer -/// flips the per-org default. fn default_usage_stream_enabled() -> bool { false } @@ -1625,9 +1621,7 @@ mod tests { #[cfg(feature = "cloud")] #[test] fn test_usage_stream_enabled_defaults_off_in_cloud() { - // Own-org usage reporting is OFF by default on cloud; orgs opt in - // explicitly. The global ZO_USAGE_REPORT_TO_OWN_ORG stays the master - // switch but no longer flips the per-org default. + // Own-org usage reporting is OFF by default on cloud; orgs opt in explicitly. assert!(!default_usage_stream_enabled()); assert!(!OrganizationSetting::default().usage_stream_enabled); diff --git a/src/config/src/config.rs b/src/config/src/config.rs index d74d54919a..6681cc50ca 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -2434,12 +2434,6 @@ pub struct Common { pub min_auto_refresh_interval: u32, #[env_config(name = "ZO_ADDITIONAL_REPORTING_ORGS", default = "")] pub additional_reporting_orgs: String, - #[env_config( - name = "ZO_USAGE_REPORT_TO_OWN_ORG", - default = true, - help = "Report alert/report triggers to the originating organization in addition to _meta org" - )] - pub usage_report_to_own_org: bool, #[env_config( name = "ZO_USE_STREAM_SETTINGS_FOR_PARTITIONS_ENABLED", default = false, @@ -5557,30 +5551,6 @@ mod tests { assert!(check_route_config(&cfg).is_err()); } - #[test] - fn test_usage_report_to_own_org_field_exists() { - // Test that usage_report_to_own_org field exists and is accessible - let cfg = Config::init().unwrap(); - // Verify the field is accessible as a boolean - let _value: bool = cfg.common.usage_report_to_own_org; - // Test passes if we can access the field without error - } - - #[test] - fn test_usage_report_to_own_org_env_override() { - // Test that environment variable can override the default - unsafe { - std::env::set_var("ZO_USAGE_REPORT_TO_OWN_ORG", "false"); - } - let cfg = Config::init().unwrap(); - // Note: This test may fail if the config is already loaded - // In that case, we just verify the field exists - let _ = cfg.common.usage_report_to_own_org; - unsafe { - std::env::remove_var("ZO_USAGE_REPORT_TO_OWN_ORG"); - } - } - #[test] fn test_ensure_not_empty_valid() { assert!(ensure_not_empty("valid", "TEST").is_ok()); diff --git a/src/core/src/self_reporting/ingestion.rs b/src/core/src/self_reporting/ingestion.rs index f44914bc9d..c1955a6075 100644 --- a/src/core/src/self_reporting/ingestion.rs +++ b/src/core/src/self_reporting/ingestion.rs @@ -213,52 +213,50 @@ pub(super) async fn ingest_usages(mut curr_usages: Vec) { } // Distribute usage to individual orgs if usage_stream_enabled is set - if cfg.common.usage_report_to_own_org { - let mut per_org_map: HashMap> = HashMap::new(); + let mut per_org_map: HashMap> = HashMap::new(); - for usage in &report_data { - if let Some(org_id) = usage.get("org_id").and_then(|v| v.as_str()) { - // Skip meta org (already handled above) - if org_id == META_ORG_ID { - continue; - } - per_org_map - .entry(org_id.to_string()) - .or_default() - .push(usage.clone()); + for usage in &report_data { + if let Some(org_id) = usage.get("org_id").and_then(|v| v.as_str()) { + // Skip meta org (already handled above) + if org_id == META_ORG_ID { + continue; + } + per_org_map + .entry(org_id.to_string()) + .or_default() + .push(usage.clone()); + } + } + + for (org, values) in per_org_map { + // report to super org as well + #[cfg(feature = "cloud")] + if let Err(e) = send_to_super_org(&org, &values).await { + log::error!("error in reporting usage of {org} to super org : {e}"); + } + // do not skip self-reporting even if super org reporting fails + + // Check if org has usage stream enabled + match crate::db::organization::get_org_setting_usage_stream_enabled(&org).await { + Ok(true) => {} + Ok(false) => continue, // self-report not enabled, so skip + Err(e) => { + log::error!( + "[SELF-REPORTING] Could not check usage_stream_enabled for {org}: {e}" + ); + continue; } } - for (org, values) in per_org_map { - // report to super org as well - #[cfg(feature = "cloud")] - if let Err(e) = send_to_super_org(&org, &values).await { - log::error!("error in reporting usage of {org} to super org : {e}"); - } - // do not skip self-reporting even if super org reporting fails + if let Err(e) = super::usage_schema::ensure_usage_stream_initialized(&org).await { + log::warn!( + "[SELF-REPORTING] Failed to ensure usage stream initialized for {org}: {e}" + ); + } - // Check if org has usage stream enabled - match crate::db::organization::get_org_setting_usage_stream_enabled(&org).await { - Ok(true) => {} - Ok(false) => continue, // self-report not enabled, so skip - Err(e) => { - log::error!( - "[SELF-REPORTING] Could not check usage_stream_enabled for {org}: {e}" - ); - continue; - } - } - - if let Err(e) = super::usage_schema::ensure_usage_stream_initialized(&org).await { - log::warn!( - "[SELF-REPORTING] Failed to ensure usage stream initialized for {org}: {e}" - ); - } - - let org_usage_stream = StreamParams::new(&org, USAGE_STREAM, StreamType::Logs); - if let Err(e) = ingest_reporting_data(values, org_usage_stream).await { - log::error!("[SELF-REPORTING] Error ingesting usage data for org {org}: {e}"); - } + let org_usage_stream = StreamParams::new(&org, USAGE_STREAM, StreamType::Logs); + if let Err(e) = ingest_reporting_data(values, org_usage_stream).await { + log::error!("[SELF-REPORTING] Error ingesting usage data for org {org}: {e}"); } } } @@ -962,13 +960,11 @@ mod tests { ); } - /// Verifies that self-org usage reporting (when enabled) also goes through + /// Verifies that self-org usage reporting also goes through /// IngestionRequest::Usage to prevent circular reporting in the org's own stream. #[test] fn test_self_org_usage_reporting_prevents_circular_reporting() { - // When usage_report_to_own_org is enabled, usage data is also sent to - // each org's own "usage" stream via ingest_reporting_data(). - // This test ensures that path also uses IngestionRequest::Usage. + // The per-org "usage" stream path must also use IngestionRequest::Usage. let usage = create_test_usage_data( UsageEvent::Ingestion, "my-org", diff --git a/src/core/src/self_reporting/persistence.rs b/src/core/src/self_reporting/persistence.rs index b1e7620df0..98d994a787 100644 --- a/src/core/src/self_reporting/persistence.rs +++ b/src/core/src/self_reporting/persistence.rs @@ -164,8 +164,7 @@ async fn ingest_buffered_data(thread_id: usize, buffered: Vec) { } let mut per_org_map = HashMap::new(); - // If configured, automatically add each trigger's own org - if cfg.common.usage_report_to_own_org && usage_reporting_mode != "remote" { + if usage_reporting_mode != "remote" { for trigger_json in triggers { if let Ok(mut trigger) = json::from_value::(trigger_json.clone()) { trigger.normalize_legacy_outcome(); diff --git a/src/core/src/self_reporting/usage_schema.rs b/src/core/src/self_reporting/usage_schema.rs index b394ed5727..ac5ea9dc3b 100644 --- a/src/core/src/self_reporting/usage_schema.rs +++ b/src/core/src/self_reporting/usage_schema.rs @@ -17,7 +17,6 @@ use std::sync::LazyLock as Lazy; use anyhow::Result; use config::{ - get_config, meta::{ self_reporting::usage::{self}, stream::StreamType, @@ -46,20 +45,12 @@ static INITIALIZED_ORGS: Lazy> = Lazy::new(DashSet::new); /// * `org_id` - The organization ID to initialize the usage stream for /// /// # Behavior -/// - Only applies when ZO_USAGE_REPORT_TO_OWN_ORG is enabled /// - Uses lock-free per-org tracking to ensure initialization happens only once per restart /// - Auto-generates field list from UsageData struct via reflection /// - Creates schema directly without ingesting sample data /// - Does NOT manipulate defined_schema_fields (no UDS filtering for system streams) /// - Never blocks ingestion even if initialization fails pub async fn ensure_usage_stream_initialized(org_id: &str) -> Result<()> { - let cfg = get_config(); - - // Only initialize when usage_report_to_own_org is enabled - if !cfg.common.usage_report_to_own_org { - return Ok(()); - } - // Atomically check-and-insert: insert() returns true only if the value was newly inserted. // This prevents race conditions where multiple threads could pass a contains() check // and all attempt initialization simultaneously. diff --git a/src/db/src/organization.rs b/src/db/src/organization.rs index 57f397aca6..524fcc47c4 100644 --- a/src/db/src/organization.rs +++ b/src/db/src/organization.rs @@ -139,12 +139,7 @@ pub async fn get_org_setting_toggle_ingestion_logs(org_id: &str) -> Result Result { let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}"); if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) { diff --git a/src/usage_reporting/src/lib.rs b/src/usage_reporting/src/lib.rs index 292e7ed2cb..9eb92599c5 100644 --- a/src/usage_reporting/src/lib.rs +++ b/src/usage_reporting/src/lib.rs @@ -84,8 +84,7 @@ pub async fn run() { /// rate, so there is no cost argument for gating it the way there is for /// `UsageData`. /// -/// Where the record lands is still configurable: `ZO_USAGE_REPORT_TO_OWN_ORG` -/// governs own-org delivery, and `_meta` always receives a copy. +/// `_meta` always receives a copy; the originating org's own stream does unless the mode is remote. pub fn publish_triggers_usage(trigger: TriggerData) { log::debug!( "[SELF-REPORTING] Publishing trigger usage: org={}, module={:?}, key={}, status={:?}",