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

refactor: extract TLS and node services from core (#13449)

Browse Source 
## Summary

- extract TLS certificate loading and client configuration from
openobserve-core into openobserve-tls
- extract node discovery, cluster information, ingester gRPC channel
selection, and Tokio runtime metrics into openobserve-node
- update Core, API, API Management, Jobs, and the root binary to depend
on the new crates directly
- preserve one-way dependencies: openobserve-core depends on
openobserve-node, while openobserve-node does not depend on Core
- register both crates in the enterprise workspace overlay

## Validation

- cargo test -p openobserve-node --lib
- CARGO_INCREMENTAL=0 cargo clippy -p openobserve-node -p
openobserve-core -p openobserve-jobs -p openobserve-api-management -p
openobserve-api -p openobserve -- -D warnings
- RUSTFLAGS="--cfg tokio_unstable" cargo check -p openobserve-node
- enterprise overlay: CARGO_INCREMENTAL=0 cargo clippy -p openobserve
--no-default-features --features enterprise -- -D warnings
- cargo deny check licenses --hide-inclusion-graph
- cargo fmt --all -- --check
- git diff --check

Companion enterprise PR: openobserve/o2-enterprise#2267
This commit is contained in:
Huaijin
2026-07-24 16:56:37 +00:00
committed by GitHub
parent 49b7c3ce5a
commit d165299bcd
42 changed files with 454 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+46 -5
View File
@@ -7187,6 +7187,7 @@ dependencies = [
"async-recursion",
"async-stream",
"async-trait",
"audit",
"aws-sdk-sns",
"axum",
"axum-client-ip",
@@ -7245,6 +7246,7 @@ dependencies = [
"openobserve-core",
"openobserve-jobs",
"openobserve-mcp",
"openobserve-node",
"opentelemetry",
"opentelemetry-otlp",
"opentelemetry-proto",
@@ -7307,6 +7309,7 @@ dependencies = [
"transform",
"uaparser",
"url",
"usage_reporting",
"utoipa",
"utoipa-swagger-ui",
"version-compare",
@@ -7365,7 +7368,9 @@ dependencies = [
"openobserve-cipher",
"openobserve-core",
"openobserve-mcp",
"openobserve-node",
"openobserve-sourcemaps",
"openobserve-tls",
"opentelemetry",
"opentelemetry-proto",
"parking_lot 0.12.5",
@@ -7395,6 +7400,7 @@ dependencies = [
"tracing-opentelemetry",
"transform",
"url",
"usage_reporting",
"utoipa",
"utoipa-swagger-ui",
"zip 8.2.0",
@@ -7436,6 +7442,7 @@ dependencies = [
"o2_openfga",
"openobserve-api-common",
"openobserve-core",
"openobserve-node",
"openobserve-org-storage",
"regex",
"reqwest 0.13.2",
@@ -7450,6 +7457,7 @@ dependencies = [
"tracing",
"transform",
"url",
"usage_reporting",
"utoipa",
"vrl",
]
@@ -7565,6 +7573,7 @@ dependencies = [
"o2_dex",
"o2_enterprise",
"o2_openfga",
"openobserve-node",
"opentelemetry",
"opentelemetry-proto",
"parking_lot 0.12.5",
@@ -7576,11 +7585,8 @@ dependencies = [
"prost-wkt-types",
"proto",
"rand 0.10.1",
"rcgen",
"regex",
"reqwest 0.13.2",
"rustls",
"rustls-native-certs",
"schema",
"sea-orm",
"search",
@@ -7603,8 +7609,6 @@ dependencies = [
"utoipa",
"version-compare",
"vrl",
"webpki-roots 1.0.6",
"x509-parser",
]
[[package]]
@@ -7634,6 +7638,7 @@ dependencies = [
"o2_openfga",
"openobserve-builtins",
"openobserve-core",
"openobserve-node",
"openobserve-org-storage",
"prometheus",
"promql-service",
@@ -7650,6 +7655,7 @@ dependencies = [
"tokio",
"tonic",
"transform",
"usage_reporting",
"wal",
]
@@ -7672,6 +7678,22 @@ dependencies = [
"utoipa",
]
[[package]]
name = "openobserve-node"
version = "0.1.0"
dependencies = [
"anyhow",
"common",
"config",
"db",
"infra",
"log",
"prometheus",
"proto",
"tokio",
"tonic",
]
[[package]]
name = "openobserve-org-storage"
version = "0.1.0"
@@ -7713,6 +7735,22 @@ dependencies = [
"zip 8.2.0",
]
[[package]]
name = "openobserve-tls"
version = "0.1.0"
dependencies = [
"anyhow",
"config",
"itertools 0.14.0",
"log",
"rcgen",
"reqwest 0.13.2",
"rustls",
"rustls-native-certs",
"webpki-roots 1.0.6",
"x509-parser",
]
[[package]]
name = "openssl-probe"
version = "0.2.1"
@@ -10138,13 +10176,16 @@ version = "0.1.0"
dependencies = [
"anyhow",
"arrow-schema",
"async-trait",
"common",
"config",
"dashmap",
"db",
"hashbrown 0.16.1",
"infra",
"ingester",
"ingestion-common",
"itertools 0.14.0",
"log",
"serde_json",
"tokio",
Cargo.toml
+9
View File
@@ -24,12 +24,14 @@ publish = false
[features]
default = []
enterprise = [
"audit/enterprise",
"openobserve-api/enterprise",
"openobserve-core/enterprise",
"openobserve-jobs/enterprise",
"super_cluster_queue/enterprise",
"search_service/enterprise",
"enrichment-data/enterprise",
"usage_reporting/enterprise",
]
vectorscan = [
"openobserve-api/vectorscan",
@@ -215,10 +217,12 @@ pin-project-lite.workspace = true
flight.workspace = true
async-stream.workspace = true
sourcemap.workspace = true
audit.workspace = true
openobserve-api.workspace = true
web.workspace = true
openobserve-api-management.workspace = true
openobserve-core.workspace = true
openobserve-node.workspace = true
ingestion-common.workspace = true
compaction.workspace = true
enrichment-data.workspace = true
@@ -227,6 +231,7 @@ db.workspace = true
search.workspace = true
search_service.workspace = true
super_cluster_queue.workspace = true
usage_reporting.workspace = true
openobserve-jobs.workspace = true
openobserve-mcp.workspace = true
@@ -259,8 +264,10 @@ members = [
"src/builtins",
"src/org_storage",
"src/sourcemaps",
"src/tls",
"src/common",
"src/core",
"src/node",
"src/transform",
"src/super_cluster_queue",
"src/search",
@@ -307,8 +314,10 @@ enrichment-data = { path = "src/enrichment_data" }
openobserve-builtins = { path = "src/builtins" }
openobserve-org-storage = { path = "src/org_storage" }
openobserve-sourcemaps = { path = "src/sourcemaps" }
openobserve-tls = { path = "src/tls" }
web = { path = "src/web" }
openobserve-core = { path = "src/core" }
openobserve-node = { path = "src/node" }
common = { path = "src/common" }
transform = { path = "src/transform" }
super_cluster_queue = { path = "src/super_cluster_queue" }
deny.toml
+2
View File
@@ -30,10 +30,12 @@ exceptions = [
{ name = "openobserve-builtins", allow = ["AGPL-3.0"] },
{ name = "common", allow = ["AGPL-3.0"] },
{ name = "openobserve-core", allow = ["AGPL-3.0"] },
{ name = "openobserve-node", allow = ["AGPL-3.0"] },
{ name = "openobserve-jobs", allow = ["AGPL-3.0"] },
{ name = "openobserve-mcp", allow = ["AGPL-3.0"] },
{ name = "openobserve-org-storage", allow = ["AGPL-3.0"] },
{ name = "openobserve-sourcemaps", allow = ["AGPL-3.0"] },
{ name = "openobserve-tls", allow = ["AGPL-3.0"] },
{ name = "transform", allow = ["AGPL-3.0"] },
{ name = "super_cluster_queue", allow = ["AGPL-3.0"] },
{ name = "search", allow = ["AGPL-3.0"] },
src/api/Cargo.toml
+4
View File
@@ -25,6 +25,7 @@ enterprise = [
"search_service/enterprise",
"enrichment-data/enterprise",
"stream/enterprise",
"usage_reporting/enterprise",
]
cloud = [
"enterprise",
@@ -58,8 +59,10 @@ common.workspace = true
audit.workspace = true
db.workspace = true
openobserve-core.workspace = true
openobserve-node.workspace = true
openobserve-builtins.workspace = true
openobserve-sourcemaps.workspace = true
openobserve-tls.workspace = true
ingestion-common.workspace = true
stream.workspace = true
enrichment-data.workspace = true
@@ -71,6 +74,7 @@ openobserve-api-management.workspace = true
openobserve-api-query.workspace = true
openobserve-api-common.workspace = true
openobserve-mcp.workspace = true
usage_reporting.workspace = true
axum.workspace = true
axum-extra.workspace = true
tower.workspace = true
src/api/src/handler/http/request/stream/mod.rs
+1 -1
View File
@@ -949,7 +949,7 @@ async fn get_super_cluster_delete_status(
let mut errors = Vec::new();
for cluster in clusters {
match openobserve_core::cluster_info::get_super_cluster_delete_job_status(
match openobserve_node::cluster_info::get_super_cluster_delete_job_status(
&trace_id,
cluster.clone(),
id,
src/api/src/handler/http/request/synthetics/mod.rs
+13 -15
View File
@@ -1091,21 +1091,19 @@ async fn process_ack(
let resp = o2_enterprise::enterprise::synthetics::job_api::ack(req, token_org).await?;
// Emit trigger usage record for synthetics telemetry.
openobserve_core::self_reporting::publish_triggers_usage(
config::meta::self_reporting::usage::TriggerData {
_timestamp: checked_at,
org: resp.org_id.clone(),
module: config::meta::self_reporting::usage::TriggerDataType::Synthetics,
key: format!("{}/{}", resp.synthetics_name, resp.synthetics_id),
start_time: checked_at,
end_time: checked_at,
status: config::meta::self_reporting::usage::TriggerDataStatus::Completed,
success_response: Some(status.clone()),
error: error.clone(),
evaluation_took_in_secs: Some(response_time_ms / 1000.0),
..Default::default()
},
);
usage_reporting::publish_triggers_usage(config::meta::self_reporting::usage::TriggerData {
_timestamp: checked_at,
org: resp.org_id.clone(),
module: config::meta::self_reporting::usage::TriggerDataType::Synthetics,
key: format!("{}/{}", resp.synthetics_name, resp.synthetics_id),
start_time: checked_at,
end_time: checked_at,
status: config::meta::self_reporting::usage::TriggerDataStatus::Completed,
success_response: Some(status.clone()),
error: error.clone(),
evaluation_took_in_secs: Some(response_time_ms / 1000.0),
..Default::default()
});
// Notify once per run, not once per job ack.
if resp.run_complete && !resp.destinations.is_empty() {
src/api/src/router/grpc/ingest/logs.rs
+1 -1
View File
@@ -15,7 +15,7 @@
use config::meta::cluster::get_internal_grpc_token;
use infra::client::grpc::MetadataMap;
use openobserve_core::grpc::get_ingester_channel;
use openobserve_node::grpc::get_ingester_channel;
use opentelemetry_proto::tonic::collector::logs::v1::{
ExportLogsServiceRequest, ExportLogsServiceResponse, logs_service_client::LogsServiceClient,
logs_service_server::LogsService,
src/api/src/router/grpc/ingest/metrics.rs
+1 -1
View File
@@ -15,7 +15,7 @@
use config::meta::cluster::get_internal_grpc_token;
use infra::client::grpc::MetadataMap;
use openobserve_core::grpc::get_ingester_channel;
use openobserve_node::grpc::get_ingester_channel;
use opentelemetry_proto::tonic::collector::metrics::v1::{
ExportMetricsServiceRequest, ExportMetricsServiceResponse,
metrics_service_client::MetricsServiceClient, metrics_service_server::MetricsService,
src/api/src/router/grpc/ingest/traces.rs
+1 -1
View File
@@ -15,7 +15,7 @@
use config::meta::cluster::get_internal_grpc_token;
use infra::client::grpc::MetadataMap;
use openobserve_core::grpc::get_ingester_channel;
use openobserve_node::grpc::get_ingester_channel;
use opentelemetry_proto::tonic::collector::trace::v1::{
ExportTraceServiceRequest, ExportTraceServiceResponse,
trace_service_client::TraceServiceClient, trace_service_server::TraceService,
src/api/src/router/http/mod.rs
+1 -1
View File
@@ -49,7 +49,7 @@ static HTTP_CLIENT: OnceLock<reqwest::Client> = OnceLock::new();
/// Returns a reference to the global HTTP client, initializing it if necessary.
fn get_http_client() -> &'static reqwest::Client {
HTTP_CLIENT.get_or_init(|| {
openobserve_core::tls::reqwest_client_tls_config()
openobserve_tls::reqwest_client_tls_config()
.expect("Failed to create HTTP client with TLS config")
})
}
src/api_management/Cargo.toml
+3
View File
@@ -16,6 +16,7 @@ enterprise = [
"dep:o2_openfga",
"openobserve-core/enterprise",
"openobserve-org-storage/enterprise",
"usage_reporting/enterprise",
]
cloud = ["enterprise", "o2_enterprise/cloud", "openobserve-core/cloud"]
@@ -29,7 +30,9 @@ search_service.workspace = true
transform.workspace = true
vrl.workspace = true
openobserve-core.workspace = true
openobserve-node.workspace = true
openobserve-org-storage.workspace = true
usage_reporting.workspace = true
anyhow.workspace = true
axum.workspace = true
axum-extra.workspace = true
src/api_management/src/auth/jwt.rs
+2 -2
View File
@@ -826,7 +826,7 @@ async fn publish_org_not_found_error(org_id: &str, user_email: &str) {
}),
};
openobserve_core::self_reporting::publish_error(error_data).await;
usage_reporting::publish_error(error_data).await;
}
#[cfg(feature = "cloud")]
@@ -1157,7 +1157,7 @@ async fn process_custom_claim_parsing(
}),
};
openobserve_core::self_reporting::publish_error(error_data).await;
usage_reporting::publish_error(error_data).await;
}) as std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>
};
src/api_management/src/request/organization/org.rs
+2 -2
View File
@@ -1431,7 +1431,7 @@ async fn get_super_cluster_nodes(regions: &[String]) -> Result<NodeListResponse,
let cluster_name = cluster.get_cluster();
// Fetch child nodes from this cluster
match openobserve_core::node::get_node_list(&trace_id, cluster).await {
match openobserve_node::node::get_node_list(&trace_id, cluster).await {
Ok(cluster_nodes) => {
for node in cluster_nodes {
response.add_node(node.clone(), region.clone(), cluster_name.clone());
@@ -1502,7 +1502,7 @@ async fn get_super_cluster_info(regions: &[String]) -> Result<ClusterInfoRespons
let cluster_name = cluster.get_cluster();
// Fetch cluster info from this cluster node
match openobserve_core::cluster_info::get_super_cluster_info(&trace_id, cluster).await {
match openobserve_node::cluster_info::get_super_cluster_info(&trace_id, cluster).await {
Ok(cluster_info_obj) => {
response.add_cluster_info(cluster_info_obj, cluster_name.clone(), region.clone());
}
src/core/Cargo.toml
+1 -5
View File
@@ -46,6 +46,7 @@ vectorscan = [
db.workspace = true
audit.workspace = true
common.workspace = true
openobserve-node.workspace = true
transform.workspace = true
vector-enrichment.workspace = true
vrl.workspace = true
@@ -93,9 +94,6 @@ proto.workspace = true
rand.workspace = true
regex.workspace = true
reqwest.workspace = true
rustls.workspace = true
rustls-native-certs.workspace = true
webpki-roots.workspace = true
sea-orm.workspace = true
serde.workspace = true
serde_json.workspace = true
@@ -111,7 +109,6 @@ tracing-opentelemetry.workspace = true
url.workspace = true
utoipa.workspace = true
version-compare.workspace = true
x509-parser.workspace = true
config.workspace = true
infra.workspace = true
ingester.workspace = true
@@ -121,4 +118,3 @@ futures-util.workspace = true
[dev-dependencies]
multer.workspace = true
rcgen.workspace = true
src/core/src/alerts/scheduler/handlers.rs
+6 -6
View File
@@ -44,6 +44,7 @@ use infra::{
#[cfg(feature = "enterprise")]
use o2_enterprise::enterprise::recommendations::service::QueryRecommendationService;
use proto::cluster_rpc;
use usage_reporting::publish_triggers_usage;
#[cfg(feature = "enterprise")]
use crate::alerts::scheduler::query_optimization_recommendation::QueryOptimizerContext;
@@ -58,7 +59,6 @@ use crate::{
db::{self, alerts::alert::set_without_updating_trigger},
ingestion::ingestion_service,
pipeline::batch_execution::ExecutablePipeline,
self_reporting::publish_triggers_usage,
};
pub async fn handle_triggers(
@@ -161,7 +161,7 @@ async fn handle_anomaly_detection_triggers(
trigger.status = db::scheduler::TriggerStatus::Waiting;
db::scheduler::update_trigger(trigger.clone(), true, "").await?;
crate::self_reporting::publish_triggers_usage(TriggerData {
usage_reporting::publish_triggers_usage(TriggerData {
_timestamp: now_micros(),
org: trigger.org.clone(),
module: TriggerDataType::AnomalyDetection,
@@ -221,7 +221,7 @@ async fn handle_anomaly_detection_triggers(
// Publish trigger run record to the triggers stream (same as alerts).
let interval_us = parse_detection_interval_to_micros(&config.schedule_interval);
let next_run = now_micros() + interval_us;
crate::self_reporting::publish_triggers_usage(TriggerData {
usage_reporting::publish_triggers_usage(TriggerData {
_timestamp: run_start_us,
org: trigger.org.clone(),
module: TriggerDataType::AnomalyDetection,
@@ -2155,7 +2155,7 @@ async fn handle_derived_stream_triggers(
error: Some(err_msg),
node_errors: HashMap::new(),
};
crate::self_reporting::publish_error(ErrorData {
usage_reporting::publish_error(ErrorData {
_timestamp: Utc::now().timestamp_micros(),
stream_params: pipeline.get_source_stream_params(),
error_source: ErrorSource::Pipeline(pipeline_error),
@@ -2303,7 +2303,7 @@ async fn handle_derived_stream_triggers(
error: Some(err),
node_errors: HashMap::new(),
};
crate::self_reporting::publish_error(ErrorData {
usage_reporting::publish_error(ErrorData {
_timestamp: Utc::now().timestamp_micros(),
stream_params: pipeline.get_source_stream_params(),
error_source: ErrorSource::Pipeline(pipeline_error),
@@ -2400,7 +2400,7 @@ async fn handle_derived_stream_triggers(
error: Some(err_msg),
node_errors: HashMap::new(),
};
crate::self_reporting::publish_error(ErrorData {
usage_reporting::publish_error(ErrorData {
_timestamp: Utc::now().timestamp_micros(),
stream_params: pipeline.get_source_stream_params(),
error_source: ErrorSource::Pipeline(pipeline_error),
src/core/src/bootstrap.rs
+32 -1
View File
@@ -16,14 +16,45 @@
use std::sync::Arc;
use ::common::infra::wal;
use config::{cache_instance_id, ider};
use config::{cache_instance_id, get_config, ider};
use db::metas;
#[cfg(feature = "enterprise")]
use crate::self_reporting::CoreAuditPublisher;
use crate::self_reporting::persistence::CoreBatchPublisher;
struct CoreOrganizationProvisioner;
#[async_trait::async_trait]
impl schema::OrganizationProvisioner for CoreOrganizationProvisioner {
fn should_auto_create_missing_orgs(&self) -> bool {
let cfg = get_config();
#[cfg(feature = "enterprise")]
let usage_enabled = true;
#[cfg(not(feature = "enterprise"))]
let usage_enabled = cfg.common.usage_enabled;
#[cfg(feature = "enterprise")]
let audit_enabled = o2_enterprise::enterprise::common::config::get_config()
.common
.audit_enabled;
#[cfg(not(feature = "enterprise"))]
let audit_enabled = false;
cfg.common.create_org_through_ingestion || usage_enabled || audit_enabled
}
async fn ensure_org_exists(&self, org_id: &str) -> Result<(), anyhow::Error> {
crate::organization::check_and_create_org(org_id)
.await
.map(|_| ())
}
}
pub async fn init() -> Result<(), anyhow::Error> {
schema::set_organization_provisioner(Arc::new(CoreOrganizationProvisioner))
.map_err(|_| anyhow::anyhow!("organization provisioner is already initialized"))?;
usage_reporting::set_batch_publisher(Arc::new(CoreBatchPublisher))
.map_err(|_| anyhow::anyhow!("usage batch publisher is already initialized"))?;
#[cfg(feature = "enterprise")]
src/core/src/functions_cache.rs
+1 -2
View File
@@ -18,8 +18,7 @@ use config::meta::{
self_reporting::error::{ErrorData, ErrorSource, FunctionError},
stream::StreamParams,
};
use crate::self_reporting::publish_error;
use usage_reporting::publish_error;
/// Loads function records into the shared cache and reports invalid entries.
pub async fn cache() -> Result<(), anyhow::Error> {
src/core/src/ingestion/ingestion_service.rs
+1 -2
View File
@@ -15,11 +15,10 @@
use config::meta::cluster::get_internal_grpc_token;
use infra::errors::{Error, Result};
use openobserve_node::grpc::get_ingester_channel;
use proto::cluster_rpc;
use tonic::{Request, codec::CompressionEncoding, metadata::MetadataValue};
use crate::grpc::get_ingester_channel;
pub async fn ingest(req: cluster_rpc::IngestionRequest) -> Result<cluster_rpc::IngestionResponse> {
let cfg = config::get_config();
let token: MetadataValue<_> = get_internal_grpc_token()
src/core/src/ingestion/mod.rs
+1 -1
View File
@@ -44,12 +44,12 @@ use infra::{
};
use ingestion_common::IngestionRequest;
use proto::cluster_rpc::IngestionType;
use usage_reporting::publish_triggers_usage;
use vrl::compiler::runtime::Runtime;
use super::{
db::alerts::alert,
pipeline::{batch_execution::ExecutablePipeline, db as pipeline},
self_reporting::publish_triggers_usage,
};
use crate::{
alerts::alert::AlertExt,
src/core/src/lib.rs
-6
View File
@@ -24,7 +24,6 @@ pub mod auth;
pub mod authz;
pub mod bootstrap;
pub mod cache;
pub mod cluster_info;
pub mod dashboards;
use ::common;
use ::db;
@@ -32,7 +31,6 @@ pub mod error_suggest;
use ::db::folders;
pub mod functions;
pub mod functions_cache;
pub mod grpc;
pub mod http;
pub mod ingestion;
pub mod ingestion_tokens;
@@ -42,7 +40,6 @@ pub mod llm_evaluations;
pub mod logs;
pub mod metadata;
pub mod metrics;
pub mod node;
#[cfg(feature = "enterprise")]
pub mod ofga;
pub mod org_cleanup;
@@ -54,8 +51,6 @@ pub mod pipeline;
pub mod providers;
#[cfg(feature = "enterprise")]
pub mod ratelimit;
pub mod runtime_metrics;
pub mod schema_watcher;
use search_service as search;
pub mod self_reporting;
pub mod service;
@@ -65,7 +60,6 @@ pub mod stream;
pub mod stream_utils;
pub mod synthetics;
pub mod system_settings;
pub mod tls;
pub mod traces;
#[cfg(feature = "cloud")]
pub mod trial_quota;
src/core/src/llm_evaluations/evaluator_trace_exporter.rs
+1 -1
View File
@@ -44,7 +44,7 @@ impl EvaluatorTraceExporter {
}
};
let (_addr, channel) = match crate::grpc::get_ingester_channel().await {
let (_addr, channel) = match openobserve_node::grpc::get_ingester_channel().await {
Ok(v) => v,
Err(e) => {
log::error!(
src/core/src/pipeline/batch_execution.rs
+2 -4
View File
@@ -53,11 +53,9 @@ use transform::{
QUERY_FUNCTIONS, apply_vrl_fn, compile_vrl_function,
js::{JSRuntimeConfig, apply_js_fn, compile_js_function},
};
use usage_reporting::publish_error;
use crate::{
alerts::{ConditionExt, ConditionGroupExt},
self_reporting::publish_error,
};
use crate::alerts::{ConditionExt, ConditionGroupExt};
// Global batch buffer for accumulating remote stream records
#[cfg(feature = "enterprise")]
src/core/src/self_reporting/mod.rs
-183
View File
@@ -13,17 +13,6 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::time::Duration;
#[cfg(feature = "enterprise")]
use audit::flush as flush_audit;
use config::{
cluster::LOCAL_NODE,
get_config,
meta::self_reporting::{EnqueueError, ReportingData, error::ErrorData, usage::TriggerData},
metrics,
};
#[cfg(feature = "cloud")]
pub mod cloud_events;
#[cfg(feature = "enterprise")]
@@ -39,178 +28,6 @@ mod usage_schema;
#[cfg(feature = "cloud")]
pub use ingestion::ingest_data_retention_usages;
pub async fn run() {
#[cfg(not(feature = "enterprise"))]
{
let cfg = get_config();
if !cfg.common.usage_enabled {
return;
}
}
if let Err(error) = usage_reporting::start().await {
log::error!("[SELF-REPORTING] Reporting queue initialization failed: {error}");
return;
}
log::debug!("[SELF-REPORTING] successfully initialized reporting queues");
}
pub fn publish_triggers_usage(trigger: TriggerData) {
#[cfg(not(feature = "enterprise"))]
{
let cfg = get_config();
if !cfg.common.usage_enabled {
log::debug!(
"[SELF-REPORTING] Skipping trigger usage publish - usage reporting disabled"
);
return;
}
}
log::debug!(
"[SELF-REPORTING] Publishing trigger usage: org={}, module={:?}, key={}, status={:?}",
trigger.org,
trigger.module,
trigger.key,
trigger.status
);
// Use non-blocking try_enqueue to prevent scheduler blocking
match usage_reporting::try_enqueue(ReportingData::Trigger(Box::new(trigger))) {
Ok(()) => {
log::debug!(
"[SELF-REPORTING] Successfully queued trigger usage data to be ingested (non-blocking)"
)
}
Err(tokio::sync::mpsc::error::TrySendError::Full(msg)) => {
// Queue is full - system is overloaded, drop the trigger data
let dropped_info = match msg {
config::meta::self_reporting::ReportingMessage::Data(ReportingData::Trigger(t)) => {
Some((t.org.clone(), t.key.clone()))
}
_ => None,
};
if let Some((org, key)) = dropped_info {
log::warn!(
"[SELF-REPORTING] Usage queue full, dropping trigger data for org={}, key={}. \
System is overloaded. Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE.",
org,
key
);
} else {
log::warn!(
"[SELF-REPORTING] Usage queue full, dropping trigger data. \
System is overloaded. Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE."
);
}
// Increment dropped triggers metric
metrics::SELF_REPORTING_DROPPED_TRIGGERS
.with_label_values(&["usage"])
.inc();
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
log::error!(
"[SELF-REPORTING] Usage queue closed, cannot send trigger data. \
Self-reporting service may have shut down."
);
}
}
}
pub async fn publish_error(error_data: ErrorData) {
let cfg = get_config();
#[cfg(not(feature = "enterprise"))]
{
if !cfg.common.usage_enabled {
log::debug!("[SELF-REPORTING] Skipping error publish - usage reporting disabled");
return;
}
}
// Important data - use shorter timeout than usage to prevent indefinite blocking
let timeout_duration = Duration::from_secs(cfg.common.error_publish_timeout_secs);
// Extract org for logging
let org = error_data.stream_params.org_id.clone();
let error_source = match &error_data.error_source {
config::meta::self_reporting::error::ErrorSource::Alert => "Alert",
config::meta::self_reporting::error::ErrorSource::Dashboard => "Dashboard",
config::meta::self_reporting::error::ErrorSource::Function(_) => "Function",
config::meta::self_reporting::error::ErrorSource::Ingestion => "Ingestion",
config::meta::self_reporting::error::ErrorSource::Pipeline(_) => "Pipeline",
config::meta::self_reporting::error::ErrorSource::Search => "Search",
config::meta::self_reporting::error::ErrorSource::Other => "Other",
config::meta::self_reporting::error::ErrorSource::SsoClaimParser(_) => "SsoClaimParser",
config::meta::self_reporting::error::ErrorSource::OrgStorage(_) => "OrgStorage",
};
log::debug!(
"[SELF-REPORTING] Publishing error data: org={}, source={}, timeout={:?}",
org,
error_source,
timeout_duration
);
let start = std::time::Instant::now();
match usage_reporting::enqueue_error_with_timeout(
ReportingData::Error(Box::new(error_data)),
timeout_duration,
)
.await
{
Ok(()) => {
let elapsed = start.elapsed();
log::debug!(
"[SELF-REPORTING] Successfully queued error data to be ingested (took {:?}, timeout-based)",
elapsed
);
}
Err(EnqueueError::Timeout) => {
log::warn!(
"[SELF-REPORTING] Timeout ({:?}) queueing error data for org={}, source={}. \
System overloaded, error reporting degraded. \
Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE.",
timeout_duration,
org,
error_source
);
metrics::SELF_REPORTING_TIMEOUT_ERRORS
.with_label_values(&["error"])
.inc();
}
Err(EnqueueError::SendFailed(e)) => {
log::error!(
"[SELF-REPORTING] Failed to send error data for org={}, source={}: {e}",
org,
error_source
);
}
}
}
pub async fn flush() {
// flush audit data
#[cfg(feature = "enterprise")]
flush_audit().await;
let cfg = get_config();
#[cfg(feature = "enterprise")]
let usage_enabled = true;
#[cfg(not(feature = "enterprise"))]
let usage_enabled = cfg.common.usage_enabled;
// only ingester and querier nodes report usage
if !usage_enabled || (!LOCAL_NODE.is_ingester() && !LOCAL_NODE.is_querier()) {
return;
}
usage_reporting::shutdown(cfg.limit.usage_reporting_thread_num).await;
}
#[cfg(feature = "enterprise")]
pub struct CoreAuditPublisher;
src/core/src/workflows/mod.rs
+1 -1
View File
@@ -28,11 +28,11 @@ use infra::table::workflows::{self, Workflow, WorkflowError, WorkflowRunData, Wo
use o2_enterprise::enterprise::common::config::get_config as get_o2_config;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use usage_reporting::publish_triggers_usage;
use crate::{
common::{infra::config::ALERTS, meta::authz::Authz, utils::get_nats_lock},
pipeline::batch_execution::{ExecutablePipeline, WorkflowResult},
self_reporting::publish_triggers_usage,
};
pub mod runtime;
src/jobs/Cargo.toml
+3
View File
@@ -22,6 +22,7 @@ enterprise = [
"search_service/enterprise",
"stream/enterprise",
"tantivy_utils/enterprise",
"usage_reporting/enterprise",
]
cloud = [
"enterprise",
@@ -41,6 +42,7 @@ audit.workspace = true
common.workspace = true
db.workspace = true
openobserve-core.workspace = true
openobserve-node.workspace = true
openobserve-builtins.workspace = true
openobserve-org-storage.workspace = true
ingestion-common.workspace = true
@@ -52,6 +54,7 @@ promql-service.workspace = true
search.workspace = true
search_service.workspace = true
tantivy_utils.workspace = true
usage_reporting.workspace = true
anyhow.workspace = true
async-trait.workspace = true
arrow.workspace = true
src/jobs/src/job/mod.rs
+7 -4
View File
@@ -15,7 +15,9 @@
use config::{DEFAULT_ORG, cluster::LOCAL_NODE, spawn_pausable_job};
use db;
use openobserve_core::{alerts, self_reporting, users};
#[cfg(feature = "cloud")]
use openobserve_core::self_reporting;
use openobserve_core::{alerts, users};
use regex::Regex;
#[cfg(feature = "enterprise")]
use {
@@ -489,7 +491,7 @@ pub async fn init() -> Result<(), anyhow::Error> {
);
}
tokio::task::spawn(self_reporting::run());
tokio::task::spawn(usage_reporting::run());
// cache short_urls
tokio::task::spawn(db::short_url::watch());
@@ -498,7 +500,8 @@ pub async fn init() -> Result<(), anyhow::Error> {
.expect("short url cache failed");
// initialize metadata watcher
tokio::task::spawn(openobserve_core::schema_watcher::watch());
let schema_watcher = schema::create_watcher()?;
tokio::task::spawn(schema_watcher);
tokio::task::spawn(db::functions::watch());
tokio::task::spawn(db::compact::retention::watch());
tokio::task::spawn(db::metrics::watch_prom_cluster_leader());
@@ -903,7 +906,7 @@ pub async fn init() -> Result<(), anyhow::Error> {
use config::meta::self_reporting::usage::{
TriggerData, TriggerDataStatus, TriggerDataType,
};
openobserve_core::self_reporting::publish_triggers_usage(TriggerData {
usage_reporting::publish_triggers_usage(TriggerData {
_timestamp: start_us,
org: org_id,
module: TriggerDataType::AnomalyDetectionTraining,
src/jobs/src/job/org_storage.rs
+1 -1
View File
@@ -82,7 +82,7 @@ async fn test_storage_validity() {
"org {} has been removed from org storage cache, and will use default storage",
provider.org_id
);
openobserve_core::self_reporting::publish_error(ErrorData {
usage_reporting::publish_error(ErrorData {
_timestamp: chrono::Utc::now().timestamp_micros(),
stream_params: Default::default(),
error_source: ErrorSource::OrgStorage(OrgStorageError {
src/jobs/src/job/promql_self_consume.rs
+2 -1
View File
@@ -24,6 +24,7 @@ use config::{
};
use hashbrown::HashSet;
use ingestion_common::{IngestUser, SystemJobType};
use openobserve_node::grpc::get_ingester_channel;
use proto::cluster_rpc::{
IngestionData, IngestionRequest, IngestionType, ingest_client::IngestClient,
};
@@ -35,7 +36,7 @@ use tonic::{
metadata::{MetadataKey, MetadataValue},
};
use crate::service::{self, grpc::get_ingester_channel};
use crate::service;
static METRICS_WHITELIST: Lazy<HashSet<String>> = Lazy::new(|| {
config::get_config()
src/main.rs
+16 -10
View File
@@ -56,10 +56,9 @@ use openobserve_api::{
},
router,
};
use openobserve_core::{
bootstrap, cluster_info::ClusterInfoService, metadata, node::NodeService, self_reporting,
};
use openobserve_core::{bootstrap, metadata};
use openobserve_jobs::job;
use openobserve_node::{cluster_info::ClusterInfoService, node::NodeService};
use opentelemetry::{KeyValue, global, trace::TracerProvider};
use opentelemetry_otlp::{WithExportConfig, WithHttpConfig, WithTonicConfig};
use opentelemetry_proto::tonic::collector::{
@@ -108,6 +107,13 @@ use tracing_subscriber::{
#[unsafe(export_name = "malloc_conf")]
pub static malloc_conf: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:16\0";
async fn flush_reporting() {
#[cfg(feature = "enterprise")]
audit::flush().await;
usage_reporting::flush().await;
}
#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
// CLI provides the path to the config file (if any)
@@ -269,7 +275,7 @@ async fn main() -> Result<(), anyhow::Error> {
// Register job runtime for metrics collection
if let Ok(handle) = tokio::runtime::Handle::try_current() {
openobserve_core::runtime_metrics::register_runtime("job".to_string(), handle);
openobserve_node::runtime_metrics::register_runtime("job".to_string(), handle);
}
job_init_tx.send(true).ok();
@@ -310,7 +316,7 @@ async fn main() -> Result<(), anyhow::Error> {
};
// Register gRPC runtime for metrics collection
openobserve_core::runtime_metrics::register_runtime(
openobserve_node::runtime_metrics::register_runtime(
"grpc".to_string(),
rt.handle().clone(),
);
@@ -334,11 +340,11 @@ async fn main() -> Result<(), anyhow::Error> {
// Register main HTTP runtime for metrics collection
if let Ok(handle) = tokio::runtime::Handle::try_current() {
openobserve_core::runtime_metrics::register_runtime("http".to_string(), handle);
openobserve_node::runtime_metrics::register_runtime("http".to_string(), handle);
}
// Start runtime metrics collector
openobserve_core::runtime_metrics::start_metrics_collector().await;
openobserve_node::runtime_metrics::start_metrics_collector().await;
// let node online
let _ = cluster::set_online().await;
@@ -429,7 +435,7 @@ async fn main() -> Result<(), anyhow::Error> {
}
// flush usage report
self_reporting::flush().await;
flush_reporting().await;
// flush service discovery
#[cfg(feature = "enterprise")]
@@ -990,7 +996,7 @@ impl opentelemetry_sdk::trace::SpanExporter for MetaOrgTraceExporter {
}
};
let (_addr, channel) = match openobserve_core::grpc::get_ingester_channel().await {
let (_addr, channel) = match openobserve_node::grpc::get_ingester_channel().await {
Ok(v) => v,
Err(e) => {
log::error!("[SEARCH-INSPECTOR] Failed to get ingester channel: {e}");
@@ -1350,7 +1356,7 @@ async fn init_action_server() -> Result<(), anyhow::Error> {
log::info!("HTTP server stopped");
// flush usage report
self_reporting::flush().await;
flush_reporting().await;
// stop telemetry
if cfg.common.telemetry_enabled {
src/node/Cargo.toml
+21
View File
@@ -0,0 +1,21 @@
[package]
name = "openobserve-node"
version.workspace = true
edition.workspace = true
license.workspace = true
publish = false
[lib]
path = "src/lib.rs"
[dependencies]
anyhow.workspace = true
common.workspace = true
config.workspace = true
db.workspace = true
infra.workspace = true
log.workspace = true
prometheus.workspace = true
proto.workspace = true
tokio.workspace = true
tonic.workspace = true
src/node/build.rs
+18
View File
@@ -0,0 +1,18 @@
// 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 <http://www.gnu.org/licenses/>.
fn main() {
println!("cargo:rustc-check-cfg=cfg(tokio_unstable)");
}
src/core/src/cluster_info.rs → src/node/src/cluster_info.rs
+1 -1
View File
@@ -69,7 +69,7 @@ impl proto::cluster_rpc::cluster_info_service_server::ClusterInfoService for Clu
let req = request.into_inner();
// Get the key from the database
match crate::db::compact::compactor_manual_jobs::get_job(&req.ksuid).await {
match db::compact::compactor_manual_jobs::get_job(&req.ksuid).await {
Ok(job) => {
// Job still exists, return pending status
Ok(Response::new(GetDeleteJobStatusResponse {
src/core/src/grpc.rs → src/node/src/grpc.rs
View File
src/node/src/lib.rs
+19
View File
@@ -0,0 +1,19 @@
// 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 <http://www.gnu.org/licenses/>.
pub mod cluster_info;
pub mod grpc;
pub mod node;
pub mod runtime_metrics;
src/core/src/node.rs → src/node/src/node.rs
View File
src/core/src/runtime_metrics.rs → src/node/src/runtime_metrics.rs
View File
src/schema/Cargo.toml
+3 -2
View File
@@ -15,15 +15,16 @@ enterprise = ["config/enterprise", "db/enterprise"]
[dependencies]
anyhow.workspace = true
arrow-schema.workspace = true
async-trait.workspace = true
common.workspace = true
config.workspace = true
dashmap.workspace = true
db.workspace = true
hashbrown.workspace = true
infra.workspace = true
ingester.workspace = true
itertools.workspace = true
log.workspace = true
ingestion-common.workspace = true
serde_json.workspace = true
[dev-dependencies]
tokio.workspace = true
src/schema/src/lib.rs
+6
View File
@@ -53,6 +53,12 @@ use infra::schema::{
use ingestion_common::StreamSchemaChk;
use serde_json::{Map, Value};
mod watcher;
pub use watcher::{
OrganizationProvisioner, SchemaWatcher, create_watcher, set_organization_provisioner,
};
const SCHEMA_CONFORMANCE_FAILED: &str = "schema_conformance_failed";
pub fn get_upto_discard_error() -> anyhow::Error {
src/core/src/schema_watcher.rs → src/schema/src/watcher.rs
+35 -19
View File
@@ -13,14 +13,18 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! Runtime watcher for schema cache updates.
//! Runtime watcher for schema cache updates and stream-local cache cleanup.
use std::{
sync::{Arc, atomic::Ordering},
future::Future,
pin::Pin,
sync::{Arc, OnceLock, atomic::Ordering},
time::Duration,
};
use arrow_schema::Schema;
use async_trait::async_trait;
use common::infra::config::ORGANIZATIONS;
use config::{
cluster::LOCAL_NODE_ID,
get_config,
@@ -29,7 +33,6 @@ use config::{
meta::stream::StreamType,
utils::{json, time::now_micros},
};
use db;
use infra::{
cache,
schema::{
@@ -37,17 +40,35 @@ use infra::{
SchemaCache, unwrap_stream_settings,
},
};
#[cfg(feature = "enterprise")]
use o2_enterprise::enterprise::common::config::get_config as get_o2_config;
use crate::{common::infra::config::ORGANIZATIONS, organization::check_and_create_org};
#[async_trait]
pub trait OrganizationProvisioner: Send + Sync + 'static {
fn should_auto_create_missing_orgs(&self) -> bool;
pub async fn watch() -> Result<(), anyhow::Error> {
#[cfg(feature = "enterprise")]
let audit_enabled = get_o2_config().common.audit_enabled;
#[cfg(not(feature = "enterprise"))]
let audit_enabled = false;
let cfg = get_config();
async fn ensure_org_exists(&self, org_id: &str) -> Result<(), anyhow::Error>;
}
static ORGANIZATION_PROVISIONER: OnceLock<Arc<dyn OrganizationProvisioner>> = OnceLock::new();
pub type SchemaWatcher = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + 'static>>;
pub fn set_organization_provisioner(
provisioner: Arc<dyn OrganizationProvisioner>,
) -> Result<(), Arc<dyn OrganizationProvisioner>> {
ORGANIZATION_PROVISIONER.set(provisioner)
}
pub fn create_watcher() -> Result<SchemaWatcher, anyhow::Error> {
let organization_provisioner = ORGANIZATION_PROVISIONER
.get()
.cloned()
.ok_or_else(|| anyhow::anyhow!("organization provisioner is not initialized"))?;
Ok(Box::pin(watch(organization_provisioner)))
}
async fn watch(
organization_provisioner: Arc<dyn OrganizationProvisioner>,
) -> Result<(), anyhow::Error> {
let key = "/schema/";
let cluster_coordinator = infra::db::get_coordinator().await;
let mut events = cluster_coordinator.watch(key).await?;
@@ -175,16 +196,11 @@ pub async fn watch() -> Result<(), anyhow::Error> {
let keys = item_key.split('/').collect::<Vec<&str>>();
let org_id = keys[0];
#[cfg(feature = "enterprise")]
let usage_enabled = true;
#[cfg(not(feature = "enterprise"))]
let usage_enabled = cfg.common.usage_enabled;
// if create_org_through_ingestion is enabled, we need to create the org
// if it doesn't exist. Hence, we need to check if the org exists in the cache
if (cfg.common.create_org_through_ingestion || usage_enabled || audit_enabled)
if organization_provisioner.should_auto_create_missing_orgs()
&& !ORGANIZATIONS.read().await.contains_key(org_id)
&& let Err(e) = check_and_create_org(org_id).await
&& let Err(e) = organization_provisioner.ensure_org_exists(org_id).await
{
log::error!("Failed to save organization in database: {e}");
}
src/tls/Cargo.toml
+23
View File
@@ -0,0 +1,23 @@
[package]
name = "openobserve-tls"
version.workspace = true
edition.workspace = true
license.workspace = true
publish = false
[lib]
path = "src/lib.rs"
[dependencies]
anyhow.workspace = true
config.workspace = true
itertools.workspace = true
log.workspace = true
reqwest.workspace = true
rustls.workspace = true
rustls-native-certs.workspace = true
webpki-roots.workspace = true
x509-parser.workspace = true
[dev-dependencies]
rcgen.workspace = true
src/core/src/tls/mod.rs → src/tls/src/lib.rs
+2
View File
@@ -13,6 +13,8 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! TLS configuration and certificate utilities.
use std::{
io::BufReader,
net::{Ipv4Addr, Ipv6Addr},
src/usage_reporting/src/lib.rs
+165 -4
View File
@@ -5,16 +5,22 @@
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
use std::sync::{Arc, LazyLock as Lazy, OnceLock};
use std::{
sync::{Arc, LazyLock as Lazy, OnceLock},
time::Duration,
};
use async_trait::async_trait;
use chrono::{DateTime, Datelike, Timelike};
use config::{
SIZE_IN_MB, get_config,
SIZE_IN_MB,
cluster::LOCAL_NODE,
get_config,
meta::{
self_reporting::{
ReportingData, ReportingMessage, ReportingQueue, ReportingRunner,
usage::{RequestStats, UsageData, UsageEvent, UsageType},
EnqueueError, ReportingData, ReportingMessage, ReportingQueue, ReportingRunner,
error::ErrorData,
usage::{RequestStats, TriggerData, UsageData, UsageEvent, UsageType},
},
stream::StreamType,
},
@@ -43,6 +49,161 @@ pub fn set_batch_publisher(
BATCH_PUBLISHER.set(publisher)
}
pub async fn run() {
#[cfg(not(feature = "enterprise"))]
{
if !get_config().common.usage_enabled {
return;
}
}
if let Err(error) = start().await {
log::error!("[SELF-REPORTING] Reporting queue initialization failed: {error}");
return;
}
log::debug!("[SELF-REPORTING] successfully initialized reporting queues");
}
pub fn publish_triggers_usage(trigger: TriggerData) {
#[cfg(not(feature = "enterprise"))]
{
if !get_config().common.usage_enabled {
log::debug!(
"[SELF-REPORTING] Skipping trigger usage publish - usage reporting disabled"
);
return;
}
}
log::debug!(
"[SELF-REPORTING] Publishing trigger usage: org={}, module={:?}, key={}, status={:?}",
trigger.org,
trigger.module,
trigger.key,
trigger.status
);
match try_enqueue(ReportingData::Trigger(Box::new(trigger))) {
Ok(()) => {
log::debug!(
"[SELF-REPORTING] Successfully queued trigger usage data to be ingested (non-blocking)"
)
}
Err(tokio::sync::mpsc::error::TrySendError::Full(message)) => {
let dropped_info = match message {
ReportingMessage::Data(ReportingData::Trigger(trigger)) => {
Some((trigger.org.clone(), trigger.key.clone()))
}
_ => None,
};
if let Some((org, key)) = dropped_info {
log::warn!(
"[SELF-REPORTING] Usage queue full, dropping trigger data for org={}, key={}. \
System is overloaded. Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE.",
org,
key
);
} else {
log::warn!(
"[SELF-REPORTING] Usage queue full, dropping trigger data. \
System is overloaded. Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE."
);
}
metrics::SELF_REPORTING_DROPPED_TRIGGERS
.with_label_values(&["usage"])
.inc();
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
log::error!(
"[SELF-REPORTING] Usage queue closed, cannot send trigger data. \
Self-reporting service may have shut down."
);
}
}
}
pub async fn publish_error(error_data: ErrorData) {
let cfg = get_config();
#[cfg(not(feature = "enterprise"))]
{
if !cfg.common.usage_enabled {
log::debug!("[SELF-REPORTING] Skipping error publish - usage reporting disabled");
return;
}
}
let timeout_duration = Duration::from_secs(cfg.common.error_publish_timeout_secs);
let org = error_data.stream_params.org_id.clone();
let error_source = match &error_data.error_source {
config::meta::self_reporting::error::ErrorSource::Alert => "Alert",
config::meta::self_reporting::error::ErrorSource::Dashboard => "Dashboard",
config::meta::self_reporting::error::ErrorSource::Function(_) => "Function",
config::meta::self_reporting::error::ErrorSource::Ingestion => "Ingestion",
config::meta::self_reporting::error::ErrorSource::Pipeline(_) => "Pipeline",
config::meta::self_reporting::error::ErrorSource::Search => "Search",
config::meta::self_reporting::error::ErrorSource::Other => "Other",
config::meta::self_reporting::error::ErrorSource::SsoClaimParser(_) => "SsoClaimParser",
config::meta::self_reporting::error::ErrorSource::OrgStorage(_) => "OrgStorage",
};
log::debug!(
"[SELF-REPORTING] Publishing error data: org={}, source={}, timeout={:?}",
org,
error_source,
timeout_duration
);
let start = std::time::Instant::now();
match enqueue_error_with_timeout(ReportingData::Error(Box::new(error_data)), timeout_duration)
.await
{
Ok(()) => {
log::debug!(
"[SELF-REPORTING] Successfully queued error data to be ingested (took {:?}, timeout-based)",
start.elapsed()
);
}
Err(EnqueueError::Timeout) => {
log::warn!(
"[SELF-REPORTING] Timeout ({:?}) queueing error data for org={}, source={}. \
System overloaded, error reporting degraded. \
Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE.",
timeout_duration,
org,
error_source
);
metrics::SELF_REPORTING_TIMEOUT_ERRORS
.with_label_values(&["error"])
.inc();
}
Err(EnqueueError::SendFailed(error)) => {
log::error!(
"[SELF-REPORTING] Failed to send error data for org={}, source={}: {error}",
org,
error_source
);
}
}
}
pub async fn flush() {
let cfg = get_config();
#[cfg(feature = "enterprise")]
let usage_enabled = true;
#[cfg(not(feature = "enterprise"))]
let usage_enabled = cfg.common.usage_enabled;
if !usage_enabled || (!LOCAL_NODE.is_ingester() && !LOCAL_NODE.is_querier()) {
return;
}
shutdown(cfg.limit.usage_reporting_thread_num).await;
}
pub async fn report_request_usage_stats(
stats: RequestStats,
org_id: &str,