starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 08:09:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: remove ZO_USAGE_REPORT_TO_OWN_ORG and always report to each org's own streams (#15406)

Browse Source
This commit is contained in:
Hengfei Yang
2026-10-15 11:44:17 +00:00
committed by GitHub
parent 6251f8a81a
commit 08a66167c1
7 changed files with 45 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/src/meta/organization.rs
+1 -7
View File
@@ -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);
src/config/src/config.rs
-30
View File
@@ -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());
src/core/src/self_reporting/ingestion.rs
+41 -45
View File
@@ -213,52 +213,50 @@ pub(super) async fn ingest_usages(mut curr_usages: Vec<UsageData>) {
}
// 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<String, Vec<json::Value>> = HashMap::new();
let mut per_org_map: HashMap<String, Vec<json::Value>> = 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",
src/core/src/self_reporting/persistence.rs
+1 -2
View File
@@ -164,8 +164,7 @@ async fn ingest_buffered_data(thread_id: usize, buffered: Vec<ReportingData>) {
}
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::<TriggerData>(trigger_json.clone()) {
trigger.normalize_legacy_outcome();
src/core/src/self_reporting/usage_schema.rs
-9
View File
@@ -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<DashSet<String>> = 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.
src/db/src/organization.rs
+1 -6
View File
@@ -139,12 +139,7 @@ pub async fn get_org_setting_toggle_ingestion_logs(org_id: &str) -> Result<bool,
}
/// Get the usage stream enabled setting for an org.
/// When no explicit setting exists, this falls back to
/// `OrganizationSetting::default()`, whose `usage_stream_enabled` defaults to
/// the global `ZO_USAGE_REPORT_TO_OWN_ORG` flag (see
/// `default_usage_stream_enabled`). So with self-reporting on, orgs report to
/// their own `usage` stream by default; a user can toggle it off per org
/// (which persists an explicit `false`), reverting to the billing-API view.
/// Without a stored setting this falls back to `default_usage_stream_enabled`, which is off.
pub async fn get_org_setting_usage_stream_enabled(org_id: &str) -> Result<bool, Error> {
let key = format!("{ORG_SETTINGS_KEY_PREFIX}/{org_id}");
if let Some(v) = ORGANIZATION_SETTING.read().await.get(&key) {
src/usage_reporting/src/lib.rs
+1 -2
View File
@@ -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={:?}",