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

feat: sync from enterprise branch (#5924)

Browse Source
This commit is contained in:
Hengfei Yang
2025-02-11 23:14:18 +08:00
committed by GitHub
parent c14c4d6d85
commit fbeae17f1f
32 changed files with 882 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+131 -7
View File
@@ -323,9 +323,9 @@ dependencies = [
[[package]]
name = "actix-web-lab"
version = "0.20.2"
version = "0.21.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7675c1a84eec1b179c844cdea8488e3e409d8e4984026e92fa96c87dd86f33c6"
checksum = "993cb5477d926300f11d7daede86368dbf797ccae3564aadfb41cb5b3aee2d23"
dependencies = [
"actix-http",
"actix-router",
@@ -335,7 +335,6 @@ dependencies = [
"actix-web-lab-derive",
"ahash 0.8.11",
"arc-swap",
"async-trait",
"bytes",
"bytestring",
"csv",
@@ -344,7 +343,7 @@ dependencies = [
"futures-util",
"http 0.2.12",
"impl-more",
"itertools 0.12.1",
"itertools 0.13.0",
"local-channel",
"mediatype",
"mime",
@@ -361,9 +360,9 @@ dependencies = [
[[package]]
name = "actix-web-lab-derive"
version = "0.20.0"
version = "0.21.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9aa0b287c8de4a76b691f29dbb5451e8dd5b79d777eaf87350c9b0cbfdb5e968"
checksum = "5aa1bc8506ff10e35419d82d2502e182b94bafa1a68f5651e8e1e6c6717fe1d3"
dependencies = [
"proc-macro2",
"quote",
@@ -539,6 +538,15 @@ version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "250f629c0161ad8107cf89319e990051fae62832fd343083bea452d93e2205fd"
[[package]]
name = "aligned-vec"
version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7e0966165eaf052580bd70eb1b32cb3d6245774c0104d1b2793e9650bf83b52a"
dependencies = [
"equator",
]
[[package]]
name = "alloc-no-stdlib"
version = "2.0.4"
@@ -1926,6 +1934,12 @@ dependencies = [
"syn 1.0.109",
]
[[package]]
name = "bytemuck"
version = "1.21.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ef657dfab802224e671f5818e9a4935f9b1957ed18e58292690cc39e7a4092a3"
[[package]]
name = "byteorder"
version = "1.5.0"
@@ -3692,6 +3706,26 @@ dependencies = [
"termcolor",
]
[[package]]
name = "equator"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c35da53b5a021d2484a7cc49b2ac7f2d840f8236a286f84202369bd338d761ea"
dependencies = [
"equator-macro",
]
[[package]]
name = "equator-macro"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3bf679796c0322556351f287a51b49e48f7c4986e727b5dd78c972d30e2e16cc"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.90",
]
[[package]]
name = "equivalent"
version = "1.0.1"
@@ -4807,6 +4841,24 @@ version = "2.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b248f5224d1d606005e02c97f5aa4e88eeb230488bcc03bc9ca4d7991399f2b5"
[[package]]
name = "inferno"
version = "0.11.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "232929e1d75fe899576a3d5c7416ad0d88dbfbb3c3d6aa00873a7408a50ddb88"
dependencies = [
"ahash 0.8.11",
"indexmap 2.7.0",
"is-terminal",
"itoa",
"log",
"num-format",
"once_cell",
"quick-xml 0.26.0",
"rgb",
"str_stack",
]
[[package]]
name = "influxdb-line-protocol"
version = "2.0.0"
@@ -5745,6 +5797,17 @@ dependencies = [
"unicase",
]
[[package]]
name = "nix"
version = "0.26.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "598beaf3cc6fdd9a5dfb1630c2800c7acd31df7aaf0f565796fba2b53ca1af1b"
dependencies = [
"bitflags 1.3.2",
"cfg-if",
"libc",
]
[[package]]
name = "nix"
version = "0.27.1"
@@ -5877,6 +5940,16 @@ version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9"
[[package]]
name = "num-format"
version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a652d9771a63711fd3c3deb670acfbe5c30a4072e664d7a3bf5a9e1056ac72c3"
dependencies = [
"arrayvec",
"itoa",
]
[[package]]
name = "num-integer"
version = "0.1.46"
@@ -6018,7 +6091,7 @@ dependencies = [
"md-5",
"parking_lot",
"percent-encoding",
"quick-xml",
"quick-xml 0.36.2",
"rand",
"reqwest",
"ring",
@@ -6178,6 +6251,7 @@ dependencies = [
"opentelemetry_sdk",
"parking_lot",
"parquet",
"pprof",
"prometheus",
"promql-parser",
"prost 0.13.4",
@@ -6788,6 +6862,32 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
[[package]]
name = "pprof"
version = "0.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ebbe2f8898beba44815fdc9e5a4ae9c929e21c5dc29b0c774a15555f7f58d6d0"
dependencies = [
"aligned-vec",
"backtrace",
"cfg-if",
"findshlibs",
"inferno",
"libc",
"log",
"nix 0.26.4",
"once_cell",
"parking_lot",
"prost 0.12.6",
"prost-build",
"prost-derive 0.12.6",
"sha2",
"smallvec",
"symbolic-demangle",
"tempfile",
"thiserror 1.0.69",
]
[[package]]
name = "pprof2"
version = "0.13.1"
@@ -7273,6 +7373,15 @@ dependencies = [
"winapi 0.3.9",
]
[[package]]
name = "quick-xml"
version = "0.26.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f50b1c63b38611e7d4d7f68b82d3ad0cc71a2ad2e7f61fc10f1328d917c93cd"
dependencies = [
"memchr",
]
[[package]]
name = "quick-xml"
version = "0.36.2"
@@ -7608,6 +7717,15 @@ dependencies = [
"windows-registry",
]
[[package]]
name = "rgb"
version = "0.8.50"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "57397d16646700483b67d2dd6511d79318f9d057fdbd21a4066aeac8b41d310a"
dependencies = [
"bytemuck",
]
[[package]]
name = "ring"
version = "0.17.8"
@@ -8920,6 +9038,12 @@ version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"
[[package]]
name = "str_stack"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9091b6114800a5f2141aee1d1b9d6ca3592ac062dc5decb3764ec5895a47b4eb"
[[package]]
name = "string_cache"
version = "0.8.7"
Cargo.toml
+8 -3
View File
@@ -26,7 +26,8 @@ default = []
enterprise = []
mimalloc = ["dep:mimalloc"]
jemalloc = ["dep:tikv-jemallocator"]
profiling = ["dep:pyroscope", "dep:pyroscope_pprofrs"]
profiling = ["dep:pprof"]
pyroscope = ["dep:pyroscope", "dep:pyroscope_pprofrs"]
tokio-console = ["dep:console-subscriber"]
[profile.release]
@@ -50,7 +51,7 @@ actix-http = "3.9"
actix-multipart = { version = "0.7", features = ["derive"] }
actix-web.workspace = true
actix-web-httpauth = "0.8"
actix-web-lab = "0.20"
actix-web-lab = "0.21"
actix-web-opentelemetry = { version = "0.19", features = ["metrics"] }
actix-web-prometheus.workspace = true
actix-web-rust-embed-responder = { version = "2.2", default-features = false, features = [
@@ -168,6 +169,7 @@ zstd.workspace = true
config.workspace = true
infra.workspace = true
ingester.workspace = true
wal.workspace = true
report_server.workspace = true
chromiumoxide.workspace = true
lettre.workspace = true
@@ -175,8 +177,11 @@ tantivy.workspace = true
zip.workspace = true
futures-util = "0.3.31"
tokio-tungstenite = "0.24.0"
wal.workspace = true
tokio-util = "0.7.13"
pprof = { version = "0.14", features = [
"flamegraph",
"prost-codec",
], optional = true }
[build-dependencies]
chrono = { version = "0.4", default-features = false, features = ["clock"] }
src/common/infra/cluster/mod.rs
+18 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -233,6 +233,23 @@ pub async fn update_local_node(node: &Node) -> Result<()> {
}
}
pub async fn set_unschedulable() -> Result<()> {
let node_id = LOCAL_NODE.uuid.clone();
if let Some(mut node) = get_node_by_uuid(&node_id).await {
node.scheduled = false;
update_local_node(&node).await?;
};
Ok(())
}
pub async fn set_schedulable() -> Result<()> {
let node_id = LOCAL_NODE.uuid.clone();
if let Some(mut node) = get_node_by_uuid(&node_id).await {
node.scheduled = true;
update_local_node(&node).await?;
};
Ok(())
}
pub async fn leave() -> Result<()> {
LOCAL_NODE_STATUS.store(NodeStatus::Offline as _, Ordering::Release);
src/common/meta/ingestion.rs
+2 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -98,7 +98,7 @@ pub const INGESTION_EP: [&str; 14] = [
"_json_arrow",
];
#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
pub struct BulkResponse {
pub took: u128,
pub errors: bool,
src/config/src/config.rs
+77 -10
View File
@@ -370,7 +370,7 @@ pub struct Config {
pub sns: Sns,
pub tcp: TCP,
pub prom: Prometheus,
pub profiling: Pyroscope,
pub profiling: Profiling,
pub smtp: Smtp,
pub rum: RUM,
pub chrome: Chrome,
@@ -451,16 +451,43 @@ pub struct Smtp {
}
#[derive(EnvConfig)]
pub struct Pyroscope {
#[env_config(name = "ZO_PROF_PYROSCOPE_ENABLED", default = false)]
pub enabled: bool,
pub struct Profiling {
#[env_config(
name = "ZO_PROF_PPROF_ENABLED",
default = false,
help = "Enable pprof profiling with pprof-rs"
)]
pub pprof_enabled: bool,
#[env_config(
name = "ZO_PROF_PPROF_PROTOBUF_ENABLED",
default = false,
help = "Enable pprof profiling with pprof-rs encode to protobuf format"
)]
pub pprof_protobuf_enabled: bool,
#[env_config(
name = "ZO_PROF_PPROF_FLAMEGRAPH_PATH",
default = "",
help = "Path to save flamegraph"
)]
pub pprof_flamegraph_path: String,
#[env_config(
name = "ZO_PROF_PYROSCOPE_ENABLED",
default = false,
help = "Enable pyroscope profiling with pyroscope-rs"
)]
pub pyroscope_enabled: bool,
#[env_config(
name = "ZO_PROF_PYROSCOPE_SERVER_URL",
default = "http://localhost:4040"
default = "http://localhost:4040",
help = "Pyroscope server URL"
)]
pub server_url: String,
#[env_config(name = "ZO_PROF_PYROSCOPE_PROJECT_NAME", default = "openobserve")]
pub project_name: String,
pub pyroscope_server_url: String,
#[env_config(
name = "ZO_PROF_PYROSCOPE_PROJECT_NAME",
default = "openobserve",
help = "Pyroscope project name"
)]
pub pyroscope_project_name: String,
}
#[derive(EnvConfig)]
@@ -638,6 +665,12 @@ pub struct Common {
help = "Default to 50_000 when ZO_FEATURE_JOIN_MATCH_ONE_ENABLED is true"
)]
pub feature_join_right_side_max_rows: usize,
#[env_config(
name = "ZO_FEATURE_QUERY_SKIP_WAL",
default = false,
help = "Skip WAL for query"
)]
pub feature_query_skip_wal: bool,
#[env_config(name = "ZO_UI_ENABLED", default = true)]
pub ui_enabled: bool,
#[env_config(name = "ZO_UI_SQL_BASE64_ENABLED", default = false)]
@@ -658,6 +691,18 @@ pub struct Common {
pub bloom_filter_ndv_ratio: u64,
#[env_config(name = "ZO_WAL_FSYNC_DISABLED", default = false)]
pub wal_fsync_disabled: bool,
#[env_config(
name = "ZO_WAL_WRITE_QUEUE_ENABLED",
default = false,
help = "Enable write queue for WAL"
)]
pub wal_write_queue_enabled: bool,
#[env_config(
name = "ZO_WAL_WRITE_QUEUE_FULL_REJECT",
default = false,
help = "Reject write when write queue is full"
)]
pub wal_write_queue_full_reject: bool,
#[env_config(name = "ZO_TRACING_ENABLED", default = false)]
pub tracing_enabled: bool,
#[env_config(name = "ZO_TRACING_SEARCH_ENABLED", default = false)]
@@ -989,6 +1034,8 @@ pub struct Limit {
pub mem_persist_interval: u64,
#[env_config(name = "ZO_WAL_WRITE_BUFFER_SIZE", default = 16384)] // 16 KB
pub wal_write_buffer_size: usize,
#[env_config(name = "ZO_WAL_WRITE_QUEUE_SIZE", default = 10000)] // 10k messages
pub wal_write_queue_size: usize,
#[env_config(name = "ZO_FILE_PUSH_INTERVAL", default = 10)] // seconds
pub file_push_interval: u64,
#[env_config(name = "ZO_FILE_PUSH_LIMIT", default = 0)] // files
@@ -1066,10 +1113,22 @@ pub struct Limit {
pub keep_alive: u64,
#[env_config(name = "ZO_ACTIX_KEEP_ALIVE_DISABLED", default = false)]
pub keep_alive_disabled: bool,
#[env_config(name = "ZO_ACTIX_SLOW_LOG_THRESHOLD", default = 5)] // seconds
pub http_slow_log_threshold: u64,
#[env_config(name = "ZO_ACTIX_SHUTDOWN_TIMEOUT", default = 5)] // seconds
pub http_shutdown_timeout: u64,
#[env_config(name = "ZO_ACTIX_SLOW_LOG_THRESHOLD", default = 5)] // seconds
pub http_slow_log_threshold: u64,
#[env_config(name = "ZO_CIRCUIT_BREAKER_ENABLED", default = false)]
pub circuit_breaker_enabled: bool,
#[env_config(name = "ZO_CIRCUIT_BREAKER_WATCHING_WINDOW", default = 60)] // seconds
pub circuit_breaker_watching_window: i64,
#[env_config(name = "ZO_CIRCUIT_BREAKER_RESET_WINDOW_NUM", default = 3)] // 3 * watching window
pub circuit_breaker_reset_window_num: i64,
#[env_config(
name = "ZO_CIRCUIT_BREAKER_SLOW_REQUEST_THRESHOLD",
default = 100,
help = "Trigger circuit break if over this threshold in watching window, and will be reset after 2 * watching window"
)] // slow requests
pub circuit_breaker_slow_request_threshold: u64,
#[env_config(name = "ZO_ALERT_SCHEDULE_INTERVAL", default = 10)] // seconds
pub alert_schedule_interval: i64,
#[env_config(name = "ZO_ALERT_SCHEDULE_CONCURRENCY", default = 5)]
@@ -1983,6 +2042,11 @@ fn check_path_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
cfg.common.mmdb_data_dir = format!("{}/", cfg.common.mmdb_data_dir);
}
// check for pprof flamegraph
if cfg.profiling.pprof_flamegraph_path.is_empty() {
cfg.profiling.pprof_flamegraph_path = format!("{}flamegraph.svg", cfg.common.data_dir);
}
Ok(())
}
@@ -2085,6 +2149,9 @@ fn check_memory_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
if cfg.limit.wal_write_buffer_size < 4096 {
cfg.limit.wal_write_buffer_size = 4096;
}
if cfg.limit.wal_write_queue_size == 0 {
cfg.limit.wal_write_queue_size = 10000;
}
// check query settings
if cfg.limit.query_group_base_speed == 0 {
src/handler/http/request/logs/ingest.rs
+3 -5
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -62,10 +62,8 @@ pub async fn bulk(
Ok(v) => MetaHttpResponse::json(v),
Err(e) => {
log::error!("Error processing request {org_id}/_bulk: {:?}", e);
HttpResponse::BadRequest().json(MetaHttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
e.to_string(),
))
// TODO: remove this once we have a proper error response
MetaHttpResponse::json(crate::common::meta::ingestion::BulkResponse::default())
}
},
)
src/handler/http/router/middlewares/slow_log.rs
+16 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -26,11 +26,15 @@ use futures_util::future::LocalBoxFuture;
pub struct SlowLog {
threshold_secs: u64,
circuit_breaker_enabled: bool,
}
impl SlowLog {
pub fn new(threshold_secs: u64) -> Self {
SlowLog { threshold_secs }
pub fn new(threshold_secs: u64, circuit_breaker_enabled: bool) -> Self {
SlowLog {
threshold_secs,
circuit_breaker_enabled,
}
}
}
@@ -50,6 +54,7 @@ where
ready(Ok(SlowLogMiddleware {
service,
threshold_secs: self.threshold_secs,
circuit_breaker_enabled: self.circuit_breaker_enabled,
}))
}
}
@@ -57,6 +62,7 @@ where
pub struct SlowLogMiddleware<S> {
service: S,
threshold_secs: u64,
circuit_breaker_enabled: bool,
}
impl<S, B> Service<ServiceRequest> for SlowLogMiddleware<S>
@@ -91,6 +97,7 @@ where
.to_string();
let method = req.method().to_string();
let threshold = Duration::from_secs(self.threshold_secs);
let circuit_breaker_enabled = self.circuit_breaker_enabled;
let fut = self.service.call(req);
@@ -98,6 +105,12 @@ where
let res = fut.await?;
let duration = start.elapsed();
// watch the request duration
if circuit_breaker_enabled {
crate::service::circuit_breaker::watch_request(duration.as_millis() as u64);
}
// log the slow request
if duration > threshold {
log::warn!(
"slow request detected - remote_addr: {}, method: {}, path: {}, took: {:.6}",
src/handler/http/router/mod.rs
+1 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -320,8 +320,6 @@ pub fn get_service_routes(svc: &mut web::ServiceConfig) {
super::auth::validator::oo_validator,
))
.wrap(cors.clone())
.wrap(middlewares::SlowLog::new(cfg.limit.http_slow_log_threshold))
.wrap(from_fn(middlewares::check_keep_alive))
.wrap(middleware::DefaultHeaders::new().add(("X-Api-Node", server)))
.service(users::list)
.service(users::save)
src/infra/src/schema/mod.rs
+41 -26
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -59,16 +59,8 @@ pub fn mk_key(org_id: &str, stream_type: StreamType, stream_name: &str) -> Strin
}
pub async fn get(org_id: &str, stream_name: &str, stream_type: StreamType) -> Result<Schema> {
let key = mk_key(org_id, stream_type, stream_name);
let cache_key = key.strip_prefix("/schema/").unwrap();
let r = STREAM_SCHEMAS_LATEST.read().await;
if let Some(schema) = r.get(cache_key) {
return Ok(schema.schema().as_ref().clone());
}
drop(r);
// if not found in cache, get from db
get_from_db(org_id, stream_name, stream_type).await
let schema = get_cache(org_id, stream_name, stream_type).await?;
Ok(schema.schema().as_ref().clone())
}
pub async fn get_cache(
@@ -78,15 +70,23 @@ pub async fn get_cache(
) -> Result<SchemaCache> {
let key = mk_key(org_id, stream_type, stream_name);
let cache_key = key.strip_prefix("/schema/").unwrap();
let r = STREAM_SCHEMAS_LATEST.read().await;
if let Some(schema) = r.get(cache_key) {
return Ok(schema.clone());
if let Some(schema) = STREAM_SCHEMAS_LATEST.read().await.get(cache_key).cloned() {
return Ok(schema);
}
drop(r);
// if not found in cache, get from db
// Get from DB without holding any locks
let schema = get_from_db(org_id, stream_name, stream_type).await?;
Ok(SchemaCache::new(schema))
let schema = SchemaCache::new(schema);
// Only acquire write lock after DB read is complete
let mut write_guard = STREAM_SCHEMAS_LATEST.write().await;
// Check again before inserting in case another thread updated while we were reading DB
if let Some(schema) = write_guard.get(cache_key) {
Ok(schema.clone())
} else {
write_guard.insert(cache_key.to_string(), schema.clone());
Ok(schema)
}
}
pub async fn get_from_db(
@@ -230,15 +230,30 @@ pub async fn get_settings(
stream_type: StreamType,
) -> Option<StreamSettings> {
let key = format!("{}/{}/{}", org_id, stream_type, stream_name);
let r = STREAM_SETTINGS.read().await;
if let Some(v) = r.get(&key) {
return Some(v.clone());
// Try to get from read lock first
if let Some(settings) = STREAM_SETTINGS.read().await.get(&key).cloned() {
return Some(settings);
}
// if not found in cache, get from db
get(org_id, stream_name, stream_type)
.await
.ok()
.and_then(|schema| unwrap_stream_settings(&schema))
// Get from DB without holding any locks
let settings = match get(org_id, stream_name, stream_type).await {
Ok(schema) => unwrap_stream_settings(&schema),
Err(_) => None,
};
// Only acquire write lock if we have settings to update
if let Some(ref s) = settings {
// Check cache again before updating as another thread might have updated while we were
// reading DB
let mut write_guard = STREAM_SETTINGS.write().await;
if !write_guard.contains_key(&key) {
write_guard.insert(key, s.clone());
}
drop(write_guard);
}
settings
}
pub fn unwrap_stream_settings(schema: &Schema) -> Option<StreamSettings> {
src/ingester/src/errors.rs
+8 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -106,5 +106,12 @@ pub enum Error {
TokioMpscSendError {
source: tokio::sync::mpsc::error::SendError<PathBuf>,
},
TokioMpscSendEntriesError {
source: tokio::sync::mpsc::error::SendError<(
crate::WriterSignal,
Vec<crate::entry::Entry>,
bool,
)>,
},
MemoryTableOverflowError {},
}
src/ingester/src/lib.rs
+8 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -40,6 +40,13 @@ pub static WAL_PARQUET_METADATA: Lazy<RwAHashMap<String, config::meta::stream::F
pub static WAL_DIR_DEFAULT_PREFIX: &str = "logs";
// writer signal
pub enum WriterSignal {
Produce,
Rotate,
Close,
}
pub async fn init() -> errors::Result<()> {
// check uncompleted parquet files, need delete those files
wal::check_uncompleted_parquet_files().await?;
src/ingester/src/writer.rs
+139 -24
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -31,7 +31,7 @@ use config::{
use hashbrown::HashSet;
use once_cell::sync::Lazy;
use snafu::ResultExt;
use tokio::sync::{Mutex, RwLock};
use tokio::sync::{mpsc, RwLock};
use wal::Writer as WalWriter;
use crate::{
@@ -40,7 +40,7 @@ use crate::{
immutable::{Immutable, IMMUTABLES},
memtable::MemTable,
rwmap::RwMap,
ReadRecordBatchEntry,
ReadRecordBatchEntry, WriterSignal,
};
static WRITERS: Lazy<Vec<RwMap<WriterKey, Arc<Writer>>>> = Lazy::new(|| {
@@ -56,10 +56,11 @@ static WRITERS: Lazy<Vec<RwMap<WriterKey, Arc<Writer>>>> = Lazy::new(|| {
pub struct Writer {
idx: usize,
key: WriterKey,
wal: Arc<Mutex<WalWriter>>,
wal: Arc<RwLock<WalWriter>>,
memtable: Arc<RwLock<MemTable>>,
next_seq: AtomicU64,
created_at: AtomicI64,
write_queue: Arc<mpsc::Sender<(WriterSignal, Vec<Entry>, bool)>>,
}
// check total memory size
@@ -91,12 +92,34 @@ pub async fn get_writer(
stream_type: &str,
stream_name: &str,
) -> Arc<Writer> {
let start = std::time::Instant::now();
let key = WriterKey::new(org_id, stream_type);
let idx = get_table_idx(thread_id, stream_name);
let r = WRITERS[idx].read().await;
let data = r.get(&key);
if start.elapsed().as_millis() > 500 {
log::warn!(
"get_writer from read cache took: {} ms",
start.elapsed().as_millis()
);
}
if let Some(w) = data {
return w.clone();
}
drop(r);
// slow path
let start = std::time::Instant::now();
let mut rw = WRITERS[idx].write().await;
let w = rw
.entry(key.clone())
.or_insert_with(|| Arc::new(Writer::new(idx, key)));
.or_insert_with(|| Writer::new(idx, key));
if start.elapsed().as_millis() > 500 {
log::warn!(
"get_writer from write cache took: {} ms",
start.elapsed().as_millis()
);
}
w.clone()
}
@@ -141,8 +164,13 @@ pub async fn check_ttl() -> Result<()> {
for w in WRITERS.iter() {
let w = w.read().await;
for r in w.values() {
// check rotation
r.rotate(0, 0).await?;
if let Err(e) = r
.write_queue
.send((WriterSignal::Rotate, vec![], false))
.await
{
log::error!("[INGESTER:MEM] writer queue rotate error: {}", e);
}
}
}
Ok(())
@@ -164,7 +192,7 @@ pub async fn flush_all() -> Result<()> {
}
impl Writer {
pub(crate) fn new(idx: usize, key: WriterKey) -> Self {
pub(crate) fn new(idx: usize, key: WriterKey) -> Arc<Writer> {
let now = Utc::now().timestamp_micros();
let cfg = get_config();
let next_seq = AtomicU64::new(now as u64);
@@ -173,16 +201,18 @@ impl Writer {
.join("logs")
.join(idx.to_string());
log::info!(
"[INGESTER:MEM] create file: {}/{}/{}/{}.wal",
"[INGESTER:MEM:{idx}] create file: {}/{}/{}/{}.wal",
wal_dir.display().to_string(),
&key.org_id,
&key.stream_type,
wal_id
);
Self {
let (tx, mut rx) = mpsc::channel(cfg.limit.wal_write_queue_size);
let writer = Self {
idx,
key: key.clone(),
wal: Arc::new(Mutex::new(
wal: Arc::new(RwLock::new(
WalWriter::new(
wal_dir,
&key.org_id,
@@ -196,7 +226,47 @@ impl Writer {
memtable: Arc::new(RwLock::new(MemTable::new())),
next_seq,
created_at: AtomicI64::new(now),
}
write_queue: Arc::new(tx),
};
let writer = Arc::new(writer);
let writer_clone = writer.clone();
log::info!("[INGESTER:MEM:{idx}] writer queue start consuming");
tokio::spawn(async move {
let mut total: usize = 0;
loop {
match rx.recv().await {
None => break,
Some((sign, entries, fsync)) => match sign {
WriterSignal::Close => break,
WriterSignal::Rotate => {
if let Err(e) = writer.rotate(0, 0).await {
log::error!("[INGESTER:MEM:{idx}] writer rotate error: {}", e);
}
}
WriterSignal::Produce => {
if let Err(e) = writer.consume(entries, fsync).await {
log::error!(
"[INGESTER:MEM:{idx}] writer consume batch error: {}",
e
);
}
}
},
}
total += 1;
if total % 1000 == 0 {
log::info!(
"[INGESTER:MEM:{idx}] writer queue consuming, total: {}, in queue: {}",
total,
rx.len()
);
}
}
log::info!("[INGESTER:MEM:{idx}] writer queue closed");
});
writer_clone
}
pub fn get_key_str(&self) -> String {
@@ -213,7 +283,40 @@ impl Writer {
self.write_batch(vec![entry], fsync).await
}
pub async fn write_batch(&self, mut entries: Vec<Entry>, fsync: bool) -> Result<()> {
pub async fn write_batch(&self, entries: Vec<Entry>, fsync: bool) -> Result<()> {
if entries.is_empty() {
return Ok(());
}
let cfg = get_config();
if !cfg.common.wal_write_queue_enabled {
return self.consume(entries, fsync).await;
}
if cfg.common.wal_write_queue_full_reject {
if let Err(e) = self
.write_queue
.try_send((WriterSignal::Produce, entries, fsync))
{
log::error!(
"[INGESTER:MEM:{}] write queue full, reject write: {}",
self.idx,
e
);
return Err(Error::WalError {
source: wal::Error::WriteQueueFull { idx: self.idx },
});
}
} else {
self.write_queue
.send((WriterSignal::Produce, entries, fsync))
.await
.context(TokioMpscSendEntriesSnafu)?;
}
Ok(())
}
async fn consume(&self, mut entries: Vec<Entry>, fsync: bool) -> Result<()> {
if entries.is_empty() {
return Ok(());
}
@@ -243,7 +346,7 @@ impl Writer {
// write into wal
let start = std::time::Instant::now();
let mut wal = self.wal.lock().await;
let mut wal = self.wal.write().await;
let wal_lock_time = start.elapsed().as_millis() as f64;
metrics::INGEST_WAL_LOCK_TIME
.with_label_values(&[&self.key.org_id])
@@ -273,7 +376,7 @@ impl Writer {
// check fsync
if fsync {
let mut wal = self.wal.lock().await;
let mut wal = self.wal.write().await;
wal.sync().context(WalSnafu)?;
drop(wal);
}
@@ -283,20 +386,22 @@ impl Writer {
// rotate is used to rotate the wal and memtable if the size exceeds the threshold
async fn rotate(&self, entry_bytes_size: usize, entry_batch_size: usize) -> Result<()> {
let start = std::time::Instant::now();
let mut wal = self.wal.lock().await;
let wal_lock_time = start.elapsed().as_millis() as f64;
metrics::INGEST_WAL_LOCK_TIME
.with_label_values(&[&self.key.org_id])
.observe(wal_lock_time);
if !self.check_wal_threshold(wal.size(), entry_bytes_size)
if !self.check_wal_threshold(self.wal.read().await.size(), entry_bytes_size)
&& !self.check_mem_threshold(self.memtable.read().await.size(), entry_batch_size)
{
return Ok(());
}
// rotation wal
let start = std::time::Instant::now();
let mut wal = self.wal.write().await;
let wal_lock_time = start.elapsed().as_millis() as f64;
metrics::INGEST_WAL_LOCK_TIME
.with_label_values(&[&self.key.org_id])
.observe(wal_lock_time);
if !self.check_wal_threshold(wal.size(), entry_bytes_size) {
return Ok(()); // check again to avoid race condition
}
let cfg = get_config();
let wal_id = self.next_seq.fetch_add(1, Ordering::SeqCst);
let wal_dir = PathBuf::from(&cfg.common.data_wal_dir)
@@ -348,8 +453,18 @@ impl Writer {
}
pub async fn close(&self) -> Result<()> {
// wait for all messages to be processed
if let Err(e) = self
.write_queue
.send((WriterSignal::Close, vec![], true))
.await
{
log::error!("[INGESTER:MEM:{}] close writer error: {}", self.idx, e);
}
self.write_queue.closed().await;
// rotation wal
let mut wal = self.wal.lock().await;
let mut wal = self.wal.write().await;
wal.sync().context(WalSnafu)?;
let path = wal.path().clone();
drop(wal);
src/job/files/parquet.rs
+3 -5
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -381,9 +381,7 @@ async fn move_files(
}
// check data retention
let stream_settings = infra::schema::get_settings(&org_id, &stream_name, stream_type)
.await
.unwrap_or_default();
let stream_settings = infra::schema::unwrap_stream_settings(&latest_schema).unwrap_or_default();
let mut stream_data_retention_days = cfg.compact.data_retention_days;
if stream_settings.data_retention > 0 {
stream_data_retention_days = stream_settings.data_retention;
@@ -647,7 +645,7 @@ async fn merge_files(
let file_name = columns[4].to_string();
// get latest version of schema
let stream_settings = infra::schema::get_settings(&org_id, &stream_name, stream_type).await;
let stream_settings = infra::schema::unwrap_stream_settings(&latest_schema);
let bloom_filter_fields = get_stream_setting_bloom_filter_fields(&stream_settings);
let full_text_search_fields = get_stream_setting_fts_fields(&stream_settings);
let index_fields = get_stream_setting_index_fields(&stream_settings);
src/main.rs
+104 -22
View File
@@ -26,6 +26,7 @@ use std::{
};
use actix_web::{dev::ServerHandle, http::KeepAlive, middleware, web, App, HttpServer};
use actix_web_lab::middleware::from_fn;
use actix_web_opentelemetry::RequestTracing;
use arrow_flight::flight_service_server::FlightServiceServer;
use config::get_config;
@@ -86,6 +87,12 @@ use tonic::{
use tracing_appender::non_blocking::WorkerGuard;
use tracing_opentelemetry::OpenTelemetryLayer;
use tracing_subscriber::Registry;
#[cfg(feature = "pyroscope")]
use {
pyroscope::PyroscopeAgent,
pyroscope_pprofrs::{pprof_backend, PprofConfig},
};
#[cfg(feature = "mimalloc")]
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
@@ -117,24 +124,39 @@ async fn main() -> Result<(), anyhow::Error> {
// setup profiling
#[cfg(feature = "profiling")]
let agent = if !cfg.profiling.enabled {
None
} else {
let agent = PyroscopeAgent::builder(&cfg.profiling.server_url, &cfg.profiling.project_name)
.tags(
[
("role", cfg.common.node_role.as_str()),
("instance", cfg.common.instance_name.as_str()),
("version", VERSION),
]
.to_vec(),
)
.backend(pprof_backend(PprofConfig::new().sample_rate(100)))
let pprof_guard = if cfg.profiling.pprof_enabled || cfg.profiling.pprof_protobuf_enabled {
let guard = pprof::ProfilerGuardBuilder::default()
.frequency(1000)
.blocklist(&["libc", "libgcc", "pthread", "vdso"])
.build()
.expect("Failed to setup pyroscope agent");
#[cfg(feature = "profiling")]
.unwrap();
Some(guard)
} else {
None
};
// setup pyroscope
#[cfg(feature = "pyroscope")]
let pyroscope_agent = if cfg.profiling.pyroscope_enabled {
let agent = PyroscopeAgent::builder(
&cfg.profiling.pyroscope_server_url,
&cfg.profiling.pyroscope_project_name,
)
.tags(
[
("role", cfg.common.node_role.as_str()),
("instance", cfg.common.instance_name.as_str()),
("version", VERSION),
]
.to_vec(),
)
.backend(pprof_backend(PprofConfig::new().sample_rate(100)))
.build()
.expect("Failed to setup pyroscope agent");
let agent_running = agent.start().expect("Failed to start pyroscope agent");
Some(agent_running)
} else {
None
};
// cli mode
@@ -391,13 +413,55 @@ async fn main() -> Result<(), anyhow::Error> {
.await;
}
// stop profiling
#[cfg(feature = "profiling")]
if let Some(agent) = agent {
let agent_ready = agent.stop().unwrap();
agent_ready.shutdown();
if let Some(guard) = pprof_guard {
if let Ok(report) = guard.report().build() {
if cfg.profiling.pprof_protobuf_enabled {
let pb_file = format!("{}.pb", cfg.profiling.pprof_flamegraph_path);
match std::fs::File::create(&pb_file) {
Ok(mut file) => {
use std::io::Write;
use pprof::protos::Message;
if let Ok(profile) = report.pprof() {
let mut content = Vec::new();
profile.encode(&mut content).unwrap();
if let Err(e) = file.write_all(&content) {
log::error!("Failed to write flamegraph: {}", e);
}
}
}
Err(e) => {
log::error!("Failed to create flamegraph file: {}", e);
}
}
} else {
match std::fs::File::create(&cfg.profiling.pprof_flamegraph_path) {
Ok(file) => {
if let Err(e) = report.flamegraph(file) {
log::error!("Failed to write flamegraph: {}", e);
}
}
Err(e) => {
log::error!("Failed to create flamegraph file: {}", e);
}
}
}
};
}
// stop pyroscope
#[cfg(feature = "pyroscope")]
if let Some(agent) = pyroscope_agent {
if let Ok(agent_ready) = agent.stop() {
agent_ready.shutdown();
}
}
log::info!("server stopped");
Ok(())
}
@@ -581,6 +645,11 @@ async fn init_http_server() -> Result<(), anyhow::Error> {
.service(
// if `cfg.common.base_uri` is empty, scope("") still works as expected.
web::scope(&cfg.common.base_uri)
.wrap(middlewares::SlowLog::new(
cfg.limit.http_slow_log_threshold,
cfg.limit.circuit_breaker_enabled,
))
.wrap(from_fn(middlewares::check_keep_alive))
.service(router::http::config)
.service(router::http::config_paths)
.service(router::http::api)
@@ -594,6 +663,11 @@ async fn init_http_server() -> Result<(), anyhow::Error> {
} else {
app = app.service({
let scope = web::scope(&cfg.common.base_uri)
.wrap(middlewares::SlowLog::new(
cfg.limit.http_slow_log_threshold,
cfg.limit.circuit_breaker_enabled,
))
.wrap(from_fn(middlewares::check_keep_alive))
.configure(get_config_routes)
.configure(get_service_routes)
.configure(get_other_service_routes)
@@ -601,7 +675,6 @@ async fn init_http_server() -> Result<(), anyhow::Error> {
.configure(get_proxy_routes);
#[cfg(feature = "enterprise")]
let scope = scope.configure(get_script_server_routes);
scope
})
}
@@ -630,7 +703,7 @@ async fn init_http_server() -> Result<(), anyhow::Error> {
let server = server
.workers(cfg.limit.http_worker_num)
.worker_max_blocking_threads(cfg.limit.http_worker_num * cfg.limit.http_worker_max_blocking)
.worker_max_blocking_threads(cfg.limit.http_worker_max_blocking)
.disable_signals()
.run();
let handle = server.handle();
@@ -685,6 +758,11 @@ async fn init_http_server_without_tracing() -> Result<(), anyhow::Error> {
.service(
// if `cfg.common.base_uri` is empty, scope("") still works as expected.
web::scope(&cfg.common.base_uri)
.wrap(middlewares::SlowLog::new(
cfg.limit.http_slow_log_threshold,
cfg.limit.circuit_breaker_enabled,
))
.wrap(from_fn(middlewares::check_keep_alive))
.service(router::http::config)
.service(router::http::config_paths)
.service(router::http::api)
@@ -698,6 +776,11 @@ async fn init_http_server_without_tracing() -> Result<(), anyhow::Error> {
} else {
app = app.service({
let scope = web::scope(&cfg.common.base_uri)
.wrap(middlewares::SlowLog::new(
cfg.limit.http_slow_log_threshold,
cfg.limit.circuit_breaker_enabled,
))
.wrap(from_fn(middlewares::check_keep_alive))
.configure(get_config_routes)
.configure(get_service_routes)
.configure(get_other_service_routes)
@@ -705,7 +788,6 @@ async fn init_http_server_without_tracing() -> Result<(), anyhow::Error> {
.configure(get_proxy_routes);
#[cfg(feature = "enterprise")]
let scope = scope.configure(get_script_server_routes);
scope
})
}
@@ -733,7 +815,7 @@ async fn init_http_server_without_tracing() -> Result<(), anyhow::Error> {
let server = server
.workers(cfg.limit.http_worker_num)
.worker_max_blocking_threads(cfg.limit.http_worker_num * cfg.limit.http_worker_max_blocking)
.worker_max_blocking_threads(cfg.limit.http_worker_max_blocking)
.disable_signals()
.run();
let handle = server.handle();
src/router/http/mod.rs
+32 -15
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -189,6 +189,7 @@ async fn dispatch(
let new_url = get_url(&path).await;
if new_url.is_error {
return Ok(HttpResponse::ServiceUnavailable()
.force_close()
.body(new_url.error.unwrap_or("internal server error".to_string())));
}
@@ -270,13 +271,15 @@ async fn default_proxy(
Ok(resp) => resp,
Err(e) => {
log::error!(
"dispatch: {} to {}, proxy request error: {}, took: {} ms",
"dispatch: {} to {}, proxy request error: {:?}, took: {} ms",
new_url.path,
new_url.node_addr,
e,
start.elapsed().as_millis()
);
return Ok(HttpResponse::ServiceUnavailable().body(e.to_string()));
return Ok(HttpResponse::ServiceUnavailable()
.force_close()
.body(e.to_string()));
}
};
@@ -299,13 +302,15 @@ async fn default_proxy(
Ok(b) => b,
Err(e) => {
log::error!(
"dispatch: {} to {}, proxy response error: {}, took: {} ms",
"dispatch: {} to {}, proxy response error: {:?}, took: {} ms",
new_url.path,
new_url.node_addr,
e,
start.elapsed().as_millis()
);
return Ok(HttpResponse::ServiceUnavailable().body(e.to_string()));
return Ok(HttpResponse::ServiceUnavailable()
.force_close()
.body(e.to_string()));
}
};
Ok(new_resp.body(body))
@@ -341,12 +346,16 @@ async fn proxy_querier_by_body(
// get node name by consistent hash
let Some(node_name) = cluster::get_node_from_consistent_hash(&key, &Role::Querier, None).await
else {
return Ok(HttpResponse::ServiceUnavailable().body("No online querier nodes"));
return Ok(HttpResponse::ServiceUnavailable()
.force_close()
.body("No online querier nodes"));
};
// get node by name
let Some(node) = cluster::get_cached_node_by_name(&node_name).await else {
return Ok(HttpResponse::ServiceUnavailable().body("No online querier nodes"));
return Ok(HttpResponse::ServiceUnavailable()
.force_close()
.body("No online querier nodes"));
};
new_url.full_url = format!("{}{}", node.http_addr, new_url.path);
new_url.node_addr = node
@@ -365,13 +374,15 @@ async fn proxy_querier_by_body(
Ok(resp) => resp,
Err(e) => {
log::error!(
"dispatch: {} to {}, proxy request error: {}, took: {} ms",
"dispatch: {} to {}, proxy request error: {:?}, took: {} ms",
new_url.path,
new_url.node_addr,
e,
start.elapsed().as_millis()
);
return Ok(HttpResponse::ServiceUnavailable().body(e.to_string()));
return Ok(HttpResponse::ServiceUnavailable()
.force_close()
.body(e.to_string()));
}
};
@@ -394,13 +405,15 @@ async fn proxy_querier_by_body(
Ok(b) => b,
Err(e) => {
log::error!(
"dispatch: {} to {}, proxy response error: {}, took: {} ms",
"dispatch: {} to {}, proxy response error: {:?}, took: {} ms",
new_url.path,
new_url.node_addr,
e,
start.elapsed().as_millis()
);
return Ok(HttpResponse::ServiceUnavailable().body(e.to_string()));
return Ok(HttpResponse::ServiceUnavailable()
.force_close()
.body(e.to_string()));
}
};
Ok(new_resp.body(body))
@@ -418,8 +431,10 @@ async fn proxy_ws(
let ws_url = match ws::convert_to_websocket_url(&new_url.full_url) {
Ok(url) => url,
Err(e) => {
log::error!("Error converting URL to WebSocket: {}", e);
return Ok(HttpResponse::BadRequest().body("Invalid WebSocket URL"));
log::error!("Error converting URL to WebSocket: {:?}", e);
return Ok(HttpResponse::BadRequest()
.force_close()
.body("Invalid WebSocket URL"));
}
};
@@ -433,8 +448,10 @@ async fn proxy_ws(
Ok(res)
}
Err(e) => {
log::error!("[WS_ROUTER] failed: {}", e);
Ok(HttpResponse::InternalServerError().body("WebSocket proxy error"))
log::error!("[WS_ROUTER] failed: {:?}", e);
Ok(HttpResponse::InternalServerError()
.force_close()
.body("WebSocket proxy error"))
}
}
} else {
src/service/alerts/alert.rs
+2 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -267,7 +267,7 @@ async fn prepare_alert(
Ok(vrl) => {
let vrl = vrl.trim().to_owned();
if !vrl.is_empty() && !vrl.ends_with('.') {
let vrl = base64::encode_url(&format!("{vrl} \n ."));
let vrl = base64::encode_url(&format!("{vrl}\n."));
alert.query_condition.vrl_function = Some(vrl);
} else if vrl.is_empty() || vrl.eq(".") {
// In case the vrl contains only ".", no need to save it
src/service/circuit_breaker.rs
+188
View File
@@ -0,0 +1,188 @@
// Copyright 2025 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/>.
use std::sync::{
atomic::{AtomicI64, AtomicU64, Ordering},
Arc, LazyLock,
};
use chrono::Utc;
const UPDATE_STATE_RETRY_TIMES: usize = 5;
static CIRCUIT_BREAKER: LazyLock<Arc<CircuitBreaker>> = LazyLock::new(|| {
let cfg = config::get_config();
CircuitBreaker::new(
cfg.limit.circuit_breaker_watching_window,
cfg.limit.circuit_breaker_reset_window_num,
cfg.limit.circuit_breaker_slow_request_threshold,
cfg.limit.http_slow_log_threshold,
)
});
pub fn watch_request(req_took_time: u64) {
CIRCUIT_BREAKER.watch(req_took_time);
}
/// CircuitBreaker is a circuit breaker that can be used to prevent a service from being overloaded.
/// It is a state machine that can be in one of three states: Closed, Open, or HalfOpen.
///
/// The circuit breaker will be in the Closed state when the service is healthy.
/// When the service is unhealthy, the circuit breaker will be in the Open state.
///
/// It watches the slow requests in the current window, and if the number of slow requests exceeds
/// the threshold, the circuit breaker will be in the Open state.
///
/// (TODO) The circuit breaker will be in the HalfOpen state when the service is recovering.
struct CircuitBreaker {
watching_window: i64, // watching window in seconds
reset_window_num: i64, // reset window number
slow_request_threshold: u64, // slow request threshold in requests
http_slow_log_threshold: u64, // http slow log threshold in seconds
current_window: AtomicI64, // current window timestamp
total_requests: AtomicU64, // current window requests
slow_requests: AtomicU64, // current window slow requests
state: AtomicU64, // circuit breaker state
will_reset_at: AtomicI64, // will reset at timestamp
}
enum CircuitBreakerState {
Closed,
Open,
#[allow(dead_code)]
HalfOpen,
}
impl CircuitBreaker {
fn new(
watching_window: i64,
reset_window_num: i64,
slow_request_threshold: u64,
http_slow_log_threshold: u64,
) -> Arc<Self> {
let cb = Self {
watching_window,
reset_window_num,
slow_request_threshold,
http_slow_log_threshold,
current_window: AtomicI64::new(0),
total_requests: AtomicU64::new(0),
slow_requests: AtomicU64::new(0),
state: AtomicU64::new(CircuitBreakerState::Closed as u64),
will_reset_at: AtomicI64::new(0),
};
// reset the current window
cb.reset_current_window();
Arc::new(cb)
}
fn watch(&self, req_took_time: u64) {
// check if need reset current window
self.reset_current_window();
self.total_requests.fetch_add(1, Ordering::Relaxed);
if req_took_time < self.http_slow_log_threshold {
return; // not a slow request
}
let slow_reqs = self.slow_requests.fetch_add(1, Ordering::Relaxed);
// check if need to open the circuit breaker
if slow_reqs >= self.slow_request_threshold
&& self.state.load(Ordering::Relaxed) == CircuitBreakerState::Closed as u64
{
self.open();
}
}
fn open(&self) {
self.state
.store(CircuitBreakerState::Open as u64, Ordering::Relaxed);
self.will_reset_at.store(
Utc::now().timestamp() + self.watching_window * self.reset_window_num,
Ordering::Relaxed,
);
log::warn!(
"[CIRCUIT_BREAKER] circuit breaker is open, total reqs: {}, slow reqs: {}",
self.total_requests.load(Ordering::Relaxed),
self.slow_requests.load(Ordering::Relaxed)
);
// change the cluster node status to unSchedulable
tokio::spawn(async move {
for _ in 0..UPDATE_STATE_RETRY_TIMES {
match crate::common::infra::cluster::set_unschedulable().await {
Ok(_) => {
log::warn!(
"[CIRCUIT_BREAKER] set the cluster node status to unSchedulable"
);
break;
}
Err(e) => {
log::error!(
"[CIRCUIT_BREAKER] failed to set the cluster node status to unSchedulable: {}",
e
);
}
}
}
});
}
fn reset_state(&self) {
if self.state.load(Ordering::Relaxed) == CircuitBreakerState::Open as u64
&& self.will_reset_at.load(Ordering::Relaxed) < Utc::now().timestamp()
&& self.slow_requests.load(Ordering::Relaxed) < self.slow_request_threshold
{
self.state
.store(CircuitBreakerState::Closed as u64, Ordering::Relaxed);
// change the cluster node status to schedulable
tokio::spawn(async move {
for _ in 0..UPDATE_STATE_RETRY_TIMES {
match crate::common::infra::cluster::set_schedulable().await {
Ok(_) => {
log::warn!(
"[CIRCUIT_BREAKER] set the cluster node status to schedulable"
);
break;
}
Err(e) => {
log::error!(
"[CIRCUIT_BREAKER] failed to set the cluster node status to schedulable: {}",
e
);
}
}
}
});
}
}
fn reset_current_window(&self) {
if self.current_window.load(Ordering::Relaxed) == self.get_current_window_timestamp() {
return;
}
self.reset_state();
self.current_window
.store(self.get_current_window_timestamp(), Ordering::Relaxed);
self.total_requests.store(0, Ordering::Relaxed);
self.slow_requests.store(0, Ordering::Relaxed);
}
fn get_current_window_timestamp(&self) -> i64 {
Utc::now().timestamp() / self.watching_window
}
}
src/service/compact/merge.rs
+1 -1
View File
@@ -626,7 +626,7 @@ pub async fn merge_files(
// get latest version of schema
let latest_schema = infra::schema::get(org_id, stream_name, stream_type).await?;
let stream_settings = infra::schema::get_settings(org_id, stream_name, stream_type).await;
let stream_settings = infra::schema::unwrap_stream_settings(&latest_schema);
let bloom_filter_fields = get_stream_setting_bloom_filter_fields(&stream_settings);
let full_text_search_fields = get_stream_setting_fts_fields(&stream_settings);
let index_fields = get_stream_setting_index_fields(&stream_settings);
src/service/db/schema.rs
+2 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -197,6 +197,7 @@ pub async fn list(
drop(r);
return Ok(list_stream_schemas(org_id, stream_type, fetch_schema).await);
}
drop(r);
let db_key = match stream_type {
None => format!("/schema/{org_id}/"),
src/service/enrichment_table/mod.rs
+13 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -206,7 +206,18 @@ pub async fn save_enrichment_data(
stream_name,
)
.await;
let mut req_stats = write_file(&writer, stream_name, buf, !cfg.common.wal_fsync_disabled).await;
let mut req_stats =
match write_file(&writer, stream_name, buf, !cfg.common.wal_fsync_disabled).await {
Ok(stats) => stats,
Err(e) => {
return Ok(
HttpResponse::InternalServerError().json(MetaHttpResponse::error(
http::StatusCode::INTERNAL_SERVER_ERROR.into(),
format!("Error writing enrichment table: {}", e),
)),
);
}
};
// notify update
if stream_schema.has_fields {
src/service/ingestion/mod.rs
+7 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -348,7 +348,7 @@ pub async fn write_file(
stream_name: &str,
buf: HashMap<String, SchemaRecords>,
fsync: bool,
) -> RequestStats {
) -> Result<RequestStats> {
let mut req_stats = RequestStats::default();
let entries = buf
.into_iter()
@@ -380,11 +380,12 @@ pub async fn write_file(
stream_name,
e
);
return Err(e.into());
}
req_stats.size += entries_size as f64 / SIZE_IN_MB;
req_stats.records += entries_records as i64;
req_stats
Ok(req_stats)
}
pub fn check_ingestion_allowed(org_id: &str, stream_name: Option<&str>) -> Result<()> {
@@ -499,6 +500,9 @@ pub async fn get_uds_and_original_data_streams(
) {
let cfg = get_config();
for stream in streams {
if user_defined_schema_map.contains_key(stream.stream_name.as_str()) {
continue;
}
let stream_settings =
infra::schema::get_settings(&stream.org_id, &stream.stream_name, stream.stream_type)
.await
src/service/logs/mod.rs
+14 -13
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -276,22 +276,23 @@ async fn write_logs(
)
.await;
let stream_settings = infra::schema::get_settings(org_id, stream_name, StreamType::Logs)
.await
.unwrap_or_default();
let schema = match stream_schema_map.get(stream_name) {
Some(schema) => schema.schema().clone(),
None => {
return Err(anyhow::anyhow!(
"Schema not found for stream: {}",
stream_name
));
}
};
let stream_settings = infra::schema::unwrap_stream_settings(&schema).unwrap_or_default();
let mut partition_keys: Vec<StreamPartition> = vec![];
let mut partition_time_level = PartitionTimeLevel::from(cfg.limit.logs_file_retention.as_str());
if stream_schema.has_partition_keys {
let partition_det = crate::service::ingestion::get_stream_partition_keys(
org_id,
&StreamType::Logs,
stream_name,
)
.await;
partition_keys = partition_det.partition_keys;
partition_keys = stream_settings.partition_keys;
partition_time_level =
unwrap_partition_time_level(partition_det.partition_time_level, StreamType::Logs);
unwrap_partition_time_level(stream_settings.partition_time_level, StreamType::Logs);
}
// Start get stream alerts
@@ -511,7 +512,7 @@ async fn write_logs(
write_buf,
!cfg.common.wal_fsync_disabled,
)
.await;
.await?;
// send distinct_values
if !distinct_values.is_empty() && !stream_name.starts_with(DISTINCT_STREAM_PREFIX) {
src/service/metadata/distinct_values.rs
+24 -26
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -25,7 +25,7 @@ use arrow_schema::{DataType, Field, Schema};
use config::{
get_config,
meta::stream::StreamType,
utils::{json, schema::infer_json_schema_from_map, schema_ext::SchemaExt},
utils::{json, schema::infer_json_schema_from_map},
FxIndexMap,
};
use infra::{
@@ -42,10 +42,10 @@ use tokio::{
use crate::{
common::meta::stream::SchemaRecords,
service,
service::{
ingestion,
db, ingestion,
metadata::{Metadata, MetadataItem},
schema::get_schema_changes,
},
};
@@ -212,12 +212,13 @@ impl Metadata for DistinctValues {
);
// check for schema
let db_schema =
infra::schema::get(&org_id, &distinct_stream_name, StreamType::Metadata).await?;
let mut _is_new = false;
if db_schema.fields().is_empty() {
_is_new = true;
infra::schema::get_cache(&org_id, &distinct_stream_name, StreamType::Metadata)
.await?;
let mut is_new = false;
if db_schema.fields_map().is_empty() {
is_new = true;
let schema = default_schema.as_ref().clone();
if let Err(e) = service::db::schema::merge(
if let Err(e) = db::schema::merge(
&org_id,
&distinct_stream_name,
StreamType::Metadata,
@@ -227,20 +228,18 @@ impl Metadata for DistinctValues {
.await
{
log::error!("[DISTINCT_VALUES] error while setting schema: {}", e);
return Err(Error::Message(e.to_string()));
}
}
let inferred_schema =
infer_json_schema_from_map(items.iter().map(|(v, _)| v), stream_type)?;
let mut schema_key = db_schema.hash_key();
let mut schema = inferred_schema;
if _is_new || db_schema.fields.ne(&schema.fields) {
match service::db::schema::merge(
let schema = if is_new || get_schema_changes(&db_schema, &inferred_schema).0 {
match db::schema::merge(
&org_id,
&distinct_stream_name,
StreamType::Metadata,
&schema,
&inferred_schema,
Some(timestamp),
)
.await
@@ -249,17 +248,16 @@ impl Metadata for DistinctValues {
log::error!(
"[DISTINCT_VALUES] error while updating schema for {org_id}/{stream_name} : {e}"
);
schema_key = schema.hash_key()
}
Ok(None) => schema_key = schema.hash_key(),
Ok(Some((s, _))) => {
schema_key = s.hash_key();
schema = s;
return Err(Error::Message(e.to_string()));
}
Ok(None) => db_schema.schema().clone(),
Ok(Some((s, _))) => Arc::new(s),
}
}
} else {
db_schema.schema().clone()
};
let schema_key = db_schema.hash_key();
let schema = Arc::new(schema);
let mut buf: HashMap<String, SchemaRecords> = HashMap::new();
for (item, count) in items {
let mut data = json::to_value(item).unwrap();
@@ -274,13 +272,13 @@ impl Metadata for DistinctValues {
&vec![],
unwrap_partition_time_level(None, StreamType::Metadata),
data,
Some(&schema_key),
Some(schema_key),
);
let data = json::Value::Object(data.clone());
let data_size = json::to_vec(&data).unwrap_or_default().len();
let hour_buf = buf.entry(hour_key).or_insert_with(|| SchemaRecords {
schema_key: schema_key.clone(),
schema_key: schema_key.to_string(),
schema: schema.clone(),
records: vec![],
records_size: 0,
@@ -312,7 +310,7 @@ impl Metadata for DistinctValues {
};
// set ownership only in the first time
if _is_new && get_openfga_config().enabled {
if is_new && get_openfga_config().enabled {
set_ownership_if_not_exists(
&org_id,
&format!("{}:{}", StreamType::Metadata, distinct_stream_name),
src/service/metrics/json.rs
+2 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -452,7 +452,7 @@ pub async fn ingest(org_id: &str, body: web::Bytes) -> Result<IngestionResponse>
ingester::get_writer(0, org_id, StreamType::Metrics.as_str(), &stream_name).await;
// for performance issue, we will flush all when the app shutdown
let fsync = false;
let mut req_stats = write_file(&writer, &stream_name, stream_data, fsync).await;
let mut req_stats = write_file(&writer, &stream_name, stream_data, fsync).await?;
req_stats.response_time = start.elapsed().as_secs_f64();
let fns_length: usize =
src/service/metrics/otlp.rs
+2 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -546,7 +546,7 @@ pub async fn handle_otlp_request(
ingester::get_writer(0, org_id, StreamType::Metrics.as_str(), &stream_name).await;
// for performance issue, we will flush all when the app shutdown
let fsync = false;
let mut req_stats = write_file(&writer, &stream_name, stream_data, fsync).await;
let mut req_stats = write_file(&writer, &stream_name, stream_data, fsync).await?;
let fns_length: usize =
stream_executable_pipelines
src/service/metrics/prom.rs
+2 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -534,7 +534,7 @@ pub async fn remote_write(
ingester::get_writer(0, org_id, StreamType::Metrics.as_str(), &stream_name).await;
// for performance issue, we will flush all when the app shutdown
let fsync = false;
let mut req_stats = write_file(&writer, &stream_name, stream_data, fsync).await;
let mut req_stats = write_file(&writer, &stream_name, stream_data, fsync).await?;
let fns_length: usize =
stream_executable_pipelines
src/service/mod.rs
+2 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -16,6 +16,7 @@
use config::{meta::stream::StreamParams, utils::schema::format_stream_name};
use infra::errors::Result;
pub mod alerts;
pub mod circuit_breaker;
pub mod compact;
pub mod dashboards;
pub mod db;
src/service/schema.rs
+13 -15
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -28,8 +28,8 @@ use config::{
use datafusion::arrow::datatypes::{Field, Schema};
use hashbrown::HashSet;
use infra::schema::{
get_settings, unwrap_stream_settings, SchemaCache, STREAM_RECORD_ID_GENERATOR,
STREAM_SCHEMAS_LATEST, STREAM_SETTINGS,
unwrap_stream_settings, SchemaCache, STREAM_RECORD_ID_GENERATOR, STREAM_SCHEMAS_LATEST,
STREAM_SETTINGS,
};
use serde_json::{Map, Value};
@@ -108,7 +108,7 @@ pub async fn check_for_schema(
get_schema_changes(schema, &inferred_schema);
if !is_schema_changed {
// check defined_schema_fields
let stream_setting = get_settings(org_id, stream_name, stream_type).await;
let stream_setting = unwrap_stream_settings(schema.schema());
let (defined_schema_fields, need_original) = match stream_setting {
Some(s) => (
s.defined_schema_fields.unwrap_or_default(),
@@ -425,7 +425,7 @@ pub fn generate_schema_for_defined_schema_fields(
))
}
fn get_schema_changes(schema: &SchemaCache, inferred_schema: &Schema) -> (bool, Vec<Field>) {
pub fn get_schema_changes(schema: &SchemaCache, inferred_schema: &Schema) -> (bool, Vec<Field>) {
let mut is_schema_changed = false;
let mut field_datatype_delta: Vec<Field> = vec![];
@@ -483,23 +483,21 @@ pub async fn stream_schema_exists(
let schema = match stream_schema_map.get(stream_name) {
Some(schema) => schema.schema().clone(),
None => {
let schema = infra::schema::get(org_id, stream_name, stream_type)
let schema_cache = infra::schema::get_cache(org_id, stream_name, stream_type)
.await
.unwrap();
let schema = Arc::new(schema);
stream_schema_map.insert(
stream_name.to_string(),
SchemaCache::new_from_arc(schema.clone()),
);
schema
let db_schema = schema_cache.schema().clone();
stream_schema_map.insert(stream_name.to_string(), schema_cache);
db_schema
}
};
if !schema.fields().is_empty() {
schema_chk.has_fields = true;
}
if let Some(value) = schema.metadata().get("settings") {
let settings: json::Value = json::from_slice(value.as_bytes()).unwrap();
if settings.get("partition_keys").is_some() {
let settings = unwrap_stream_settings(&schema);
if let Some(stream_setting) = settings {
if !stream_setting.partition_keys.is_empty() {
schema_chk.has_partition_keys = true;
}
}
src/service/search/cluster/flight.rs
+8 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -450,7 +450,13 @@ pub async fn get_online_querier_nodes(
node_group: Option<RoleGroup>,
) -> Result<Vec<Node>> {
// get nodes from cluster
let mut nodes = match infra_cluster::get_cached_online_query_nodes(node_group).await {
let cfg = get_config();
let nodes = if cfg.common.feature_query_skip_wal {
infra_cluster::get_cached_online_querier_nodes(node_group).await
} else {
infra_cluster::get_cached_online_query_nodes(node_group).await
};
let mut nodes = match nodes {
Some(nodes) => nodes,
None => {
log::error!("[trace_id {trace_id}] flight->search: no querier node online");
src/service/traces/mod.rs
+6 -2
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -961,7 +961,11 @@ async fn write_traces(
data_buf,
!cfg.common.wal_fsync_disabled,
)
.await;
.await
.map_err(|e| {
log::error!("Error while writing traces: {}", e);
std::io::Error::new(std::io::ErrorKind::Other, e.to_string())
})?;
// send distinct_values
if !distinct_values.is_empty() && !stream_name.starts_with(DISTINCT_STREAM_PREFIX) {
src/wal/src/errors.rs
+4 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 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
@@ -83,4 +83,7 @@ pub enum Error {
expected: u32,
actual: u32,
},
WriteQueueFull {
idx: usize,
},
}
web/src/plugins/traces/SyntaxGuide.vue
+1 -6
View File
@@ -1,4 +1,4 @@
<!-- Copyright 2023 OpenObserve Inc.
<!-- Copyright 2025 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
@@ -67,11 +67,6 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
To search value 'stderr' for stream column use
<span class="bg-highlight">stream='stderr'</span>
</li>
<li>
To search and use query function <i>extract_ip</i> on column
log use
<span class="bg-highlight">extract_ip(log) | code=200</span>
</li>
<li>
For additional examples,
<a