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

feat: support reloading config vars using an API (#3623)

Browse Source 
The new endpoint `/config/reload` will refresh the
environment variables.

There are a bunch of env-vars which will not be reloaded
for e.g. HTTP_PORT etc. these will need a restart.

---------

Co-authored-by: Hengfei Yang <hengfei.yang@gmail.com>
This commit is contained in:
Ankur Srivastava
2024-05-30 08:23:27 +08:00
committed by GitHub
co-authored by Hengfei Yang
parent d502edce96
commit 8f3df9e7a1
319 changed files with 2043 additions and 1651 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3 -2
View File
@@ -2265,6 +2265,7 @@ dependencies = [
"actix-web-prometheus",
"ahash 0.8.11",
"anyhow",
"arc-swap",
"arrow",
"arrow-json",
"arrow-schema",
@@ -5328,7 +5329,7 @@ dependencies = [
"rayon",
"regex",
"regex-syntax 0.8.3",
"reportserver",
"report_server",
"reqwest 0.12.4",
"rust-embed-for-web",
"segment",
@@ -6356,7 +6357,7 @@ dependencies = [
]
[[package]]
name = "reportserver"
name = "report_server"
version = "0.1.0"
dependencies = [
"actix-web",
Cargo.toml
+4 -3
View File
@@ -159,7 +159,7 @@ zstd.workspace = true
config.workspace = true
infra.workspace = true
ingester.workspace = true
reportserver.workspace = true
report_server.workspace = true
chromiumoxide.workspace = true
lettre.workspace = true
@@ -173,7 +173,7 @@ float-cmp = "0.9"
async-walkdir = "1.0.0"
[workspace]
members = ["src/config", "src/infra", "src/ingester", "src/wal", "src/proto", "src/reportserver"]
members = ["src/config", "src/infra", "src/ingester", "src/wal", "src/proto", "src/report_server"]
resolver = "2"
[workspace.package]
@@ -187,12 +187,13 @@ infra = { path = "src/infra" }
ingester = { path = "src/ingester" }
wal = { path = "src/wal" }
proto = { path = "src/proto" }
reportserver = { path = "src/reportserver" }
report_server = { path = "src/report_server" }
ahash = { version = "0.8", features = ["serde"] }
actix-web = "4.5"
actix-web-prometheus = { version = "0.1", features = ["process"] }
anyhow = "1.0"
arc-swap = "1.7.1"
argon2 = { version = "0.5", features = ["alloc", "password-hash"] }
async-trait = "0.1"
async-recursion = "1.0"
build.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/cli/basic/cli.rs
+5 -5
View File
@@ -13,7 +13,7 @@
// 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 config::{utils::file::set_permission, CONFIG};
use config::utils::file::set_permission;
use infra::file_list as infra_file_list;
use crate::{
@@ -115,7 +115,7 @@ pub async fn cli() -> Result<bool, anyhow::Error> {
if app.subcommand().is_none() {
return Ok(false);
}
let cfg = config::get_config();
#[cfg(not(feature = "tokio-console"))]
env_logger::init_from_env(env_logger::Env::new().default_filter_or("INFO"));
@@ -142,13 +142,13 @@ pub async fn cli() -> Result<bool, anyhow::Error> {
"root" => {
let _ = users::update_user(
meta::organization::DEFAULT_ORG,
CONFIG.auth.root_user_email.as_str(),
cfg.auth.root_user_email.as_str(),
false,
CONFIG.auth.root_user_email.as_str(),
cfg.auth.root_user_email.as_str(),
meta::user::UpdateUser {
change_password: true,
old_password: None,
new_password: Some(CONFIG.auth.root_user_password.clone()),
new_password: Some(cfg.auth.root_user_password.clone()),
role: Some(meta::user::UserRole::Root),
first_name: Some("root".to_owned()),
last_name: Some("".to_owned()),
src/cli/basic/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/cli/data/cli.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/cli/data/export.rs
+2 -1
View File
@@ -37,6 +37,7 @@ impl Context for Export {
Err(_) => return Ok(false),
};
let cfg = config::get_config();
let table = c.stream_name;
let query = search::Query {
sql: format!("select * from {}", table),
@@ -47,7 +48,7 @@ impl Context for Export {
query_type: "".to_owned(),
start_time: c.start_time,
end_time: c.end_time,
sort_by: Some(format!("{} ASC", config::CONFIG.common.column_timestamp)),
sort_by: Some(format!("{} ASC", cfg.common.column_timestamp)),
track_total_hits: false,
query_context: None,
uses_zo_fn: false,
src/cli/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/infra/cluster/etcd.rs
+25 -23
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -15,9 +15,9 @@
use config::{
cluster::*,
get_config,
meta::cluster::{Node, NodeStatus, Role},
utils::json,
CONFIG,
};
use etcd_client::PutOptions;
use infra::{
@@ -49,9 +49,12 @@ pub(crate) async fn register_and_keepalive() -> Result<()> {
}
let lease_id = unsafe { LOCAL_NODE_KEY_LEASE_ID };
let ret =
etcd::keepalive_lease_id(lease_id, CONFIG.limit.node_heartbeat_ttl, is_offline)
.await;
let ret = etcd::keepalive_lease_id(
lease_id,
get_config().limit.node_heartbeat_ttl,
is_offline,
)
.await;
if ret.is_ok() {
break;
}
@@ -77,8 +80,9 @@ pub(crate) async fn register_and_keepalive() -> Result<()> {
/// Register to cluster
async fn register() -> Result<()> {
let cfg = get_config();
// 1. create a cluster lock for node register
let locker = dist_lock::lock("/nodes/register", CONFIG.limit.node_heartbeat_ttl as u64).await?;
let locker = dist_lock::lock("/nodes/register", cfg.limit.node_heartbeat_ttl as u64).await?;
// 2. watch node list
tokio::task::spawn(async move { super::watch_node_list().await });
@@ -126,15 +130,15 @@ async fn register() -> Result<()> {
}
// 4. join the cluster
let key = format!("{}nodes/{}", &CONFIG.etcd.prefix, *LOCAL_NODE_UUID);
let key = format!("{}nodes/{}", &cfg.etcd.prefix, *LOCAL_NODE_UUID);
let node = Node {
id: new_node_id,
uuid: LOCAL_NODE_UUID.clone(),
name: CONFIG.common.instance_name.clone(),
http_addr: format!("http://{}:{}", get_local_http_ip(), CONFIG.http.port),
grpc_addr: format!("http://{}:{}", get_local_grpc_ip(), CONFIG.grpc.port),
name: cfg.common.instance_name.clone(),
http_addr: format!("http://{}:{}", get_local_http_ip(), cfg.http.port),
grpc_addr: format!("http://{}:{}", get_local_grpc_ip(), cfg.grpc.port),
role: LOCAL_NODE_ROLE.clone(),
cpu_num: CONFIG.limit.cpu_num as u64,
cpu_num: cfg.limit.cpu_num as u64,
status: NodeStatus::Prepare,
scheduled: true,
broadcasted: false,
@@ -155,10 +159,7 @@ async fn register() -> Result<()> {
// register node to cluster
let mut client = etcd::get_etcd_client().await.clone();
let resp = match client
.lease_grant(CONFIG.limit.node_heartbeat_ttl, None)
.await
{
let resp = match client.lease_grant(cfg.limit.node_heartbeat_ttl, None).await {
Ok(v) => v,
Err(e) => {
dist_lock::unlock(&locker).await?;
@@ -193,6 +194,7 @@ pub(crate) async fn set_offline(new_lease_id: bool) -> Result<()> {
/// set online to cluster
pub(crate) async fn set_status(status: NodeStatus, new_lease_id: bool) -> Result<()> {
let cfg = get_config();
// set node status to online
let node = match super::NODES.read().await.get(LOCAL_NODE_UUID.as_str()) {
Some(node) => {
@@ -203,11 +205,11 @@ pub(crate) async fn set_status(status: NodeStatus, new_lease_id: bool) -> Result
None => Node {
id: unsafe { LOCAL_NODE_ID },
uuid: LOCAL_NODE_UUID.clone(),
name: CONFIG.common.instance_name.clone(),
http_addr: format!("http://{}:{}", get_local_node_ip(), CONFIG.http.port),
grpc_addr: format!("http://{}:{}", get_local_node_ip(), CONFIG.grpc.port),
name: cfg.common.instance_name.clone(),
http_addr: format!("http://{}:{}", get_local_node_ip(), cfg.http.port),
grpc_addr: format!("http://{}:{}", get_local_node_ip(), cfg.grpc.port),
role: LOCAL_NODE_ROLE.clone(),
cpu_num: CONFIG.limit.cpu_num as u64,
cpu_num: cfg.limit.cpu_num as u64,
status: status.clone(),
scheduled: true,
broadcasted: false,
@@ -223,7 +225,7 @@ pub(crate) async fn set_status(status: NodeStatus, new_lease_id: bool) -> Result
// get new lease id
let mut client = etcd::get_etcd_client().await.clone();
let resp = client
.lease_grant(CONFIG.limit.node_heartbeat_ttl, None)
.lease_grant(cfg.limit.node_heartbeat_ttl, None)
.await?;
let lease_id = resp.id();
// update local node key lease id
@@ -232,7 +234,7 @@ pub(crate) async fn set_status(status: NodeStatus, new_lease_id: bool) -> Result
}
}
let key = format!("{}nodes/{}", &CONFIG.etcd.prefix, *LOCAL_NODE_UUID);
let key = format!("{}nodes/{}", &cfg.etcd.prefix, *LOCAL_NODE_UUID);
let opt = PutOptions::new().with_lease(unsafe { LOCAL_NODE_KEY_LEASE_ID });
let mut client = etcd::get_etcd_client().await.clone();
if let Err(e) = client.put(key, val, Some(opt)).await {
@@ -244,7 +246,7 @@ pub(crate) async fn set_status(status: NodeStatus, new_lease_id: bool) -> Result
/// Leave cluster
pub(crate) async fn leave() -> Result<()> {
let key = format!("{}nodes/{}", &CONFIG.etcd.prefix, *LOCAL_NODE_UUID);
let key = format!("{}nodes/{}", get_config().etcd.prefix, *LOCAL_NODE_UUID);
let mut client = etcd::get_etcd_client().await.clone();
if let Err(e) = client.delete(key, None).await {
return Err(Error::Message(format!("leave node error: {}", e)));
@@ -254,7 +256,7 @@ pub(crate) async fn leave() -> Result<()> {
}
pub(crate) async fn update_local_node(node: &Node) -> Result<()> {
let key = format!("{}nodes/{}", &CONFIG.etcd.prefix, *LOCAL_NODE_UUID);
let key = format!("{}nodes/{}", get_config().etcd.prefix, *LOCAL_NODE_UUID);
let opt = PutOptions::new().with_lease(unsafe { LOCAL_NODE_KEY_LEASE_ID });
let val = json::to_string(&node).unwrap();
let mut client = etcd::get_etcd_client().await.clone();
src/common/infra/cluster/mod.rs
+35 -25
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -17,12 +17,13 @@ use std::{cmp::min, ops::Bound, sync::Arc, time::Duration};
use config::{
cluster::*,
get_config, get_instance_id,
meta::{
cluster::{Node, NodeStatus, Role},
meta_store::MetaStore,
},
utils::{hash::Sum64, json},
RwAHashMap, RwBTreeMap, CONFIG, INSTANCE_ID,
RwAHashMap, RwBTreeMap,
};
use infra::{
db::{get_coordinator, Event},
@@ -51,7 +52,7 @@ pub async fn add_node_to_consistent_hash(node: &Node, role: &Role) {
_ => return,
};
let mut h = config::utils::hash::gxhash::new();
for i in 0..CONFIG.limit.consistent_hash_vnodes {
for i in 0..get_config().limit.consistent_hash_vnodes {
let key = format!("{}{}", node.uuid, i);
let hash = h.sum64(&key);
nodes.insert(hash, node.uuid.clone());
@@ -65,7 +66,7 @@ pub async fn remove_node_from_consistent_hash(node: &Node, role: &Role) {
_ => return,
};
let mut h = config::utils::hash::gxhash::new();
for i in 0..CONFIG.limit.consistent_hash_vnodes {
for i in 0..get_config().limit.consistent_hash_vnodes {
let key = format!("{}{}", node.uuid, i);
let hash = h.sum64(&key);
nodes.remove(&hash);
@@ -94,16 +95,18 @@ pub async fn get_node_from_consistent_hash(key: &str, role: &Role) -> Option<Str
#[inline]
pub fn get_internal_grpc_token() -> String {
if CONFIG.grpc.internal_grpc_token.is_empty() {
INSTANCE_ID.get("instance_id").unwrap().to_string()
let cfg = get_config();
if cfg.grpc.internal_grpc_token.is_empty() {
get_instance_id()
} else {
CONFIG.grpc.internal_grpc_token.clone()
cfg.grpc.internal_grpc_token.clone()
}
}
/// Register and keepalive the node to cluster
pub async fn register_and_keepalive() -> Result<()> {
if CONFIG.common.local_mode {
let cfg = get_config();
if cfg.common.local_mode {
let roles = load_local_node_role();
if !is_single_node(&roles) {
panic!("Local mode only support NODE_ROLE=all");
@@ -116,14 +119,14 @@ pub async fn register_and_keepalive() -> Result<()> {
return Ok(());
}
match CONFIG.common.cluster_coordinator.as_str().into() {
match cfg.common.cluster_coordinator.as_str().into() {
MetaStore::Nats => nats::register_and_keepalive().await?,
_ => etcd::register_and_keepalive().await?,
};
// check node heatbeat
tokio::task::spawn(async move {
let ttl_keep_alive = min(10, (CONFIG.limit.node_heartbeat_ttl / 2) as u64);
let ttl_keep_alive = min(10, (cfg.limit.node_heartbeat_ttl / 2) as u64);
let client = reqwest::Client::new();
loop {
time::sleep(time::Duration::from_secs(ttl_keep_alive)).await;
@@ -137,33 +140,36 @@ pub async fn register_and_keepalive() -> Result<()> {
}
pub async fn set_online(new_lease_id: bool) -> Result<()> {
if CONFIG.common.local_mode {
let cfg = get_config();
if cfg.common.local_mode {
return Ok(());
}
match CONFIG.common.cluster_coordinator.as_str().into() {
match cfg.common.cluster_coordinator.as_str().into() {
MetaStore::Nats => nats::set_online().await,
_ => etcd::set_online(new_lease_id).await,
}
}
pub async fn set_offline(new_lease_id: bool) -> Result<()> {
if CONFIG.common.local_mode {
let cfg = get_config();
if cfg.common.local_mode {
return Ok(());
}
match CONFIG.common.cluster_coordinator.as_str().into() {
match cfg.common.cluster_coordinator.as_str().into() {
MetaStore::Nats => nats::set_offline().await,
_ => etcd::set_offline(new_lease_id).await,
}
}
pub async fn update_local_node(node: &Node) -> Result<()> {
if CONFIG.common.local_mode {
let cfg = get_config();
if cfg.common.local_mode {
return Ok(());
}
match CONFIG.common.cluster_coordinator.as_str().into() {
match cfg.common.cluster_coordinator.as_str().into() {
MetaStore::Nats => nats::update_local_node(node).await,
_ => etcd::update_local_node(node).await,
}
@@ -173,12 +179,12 @@ pub async fn leave() -> Result<()> {
unsafe {
LOCAL_NODE_STATUS = NodeStatus::Offline;
}
if CONFIG.common.local_mode {
let cfg = get_config();
if cfg.common.local_mode {
return Ok(());
}
match CONFIG.common.cluster_coordinator.as_str().into() {
match cfg.common.cluster_coordinator.as_str().into() {
MetaStore::Nats => nats::leave().await,
_ => etcd::leave().await,
}
@@ -207,7 +213,9 @@ async fn watch_node_list() -> Result<()> {
let mut events = client.watch(key).await?;
let events = Arc::get_mut(&mut events).unwrap();
log::info!("Start watching node_list");
loop {
let cfg = get_config();
let ev = match events.recv().await {
Some(ev) => ev,
None => {
@@ -238,7 +246,7 @@ async fn watch_node_list() -> Result<()> {
continue;
}
log::info!("[CLUSTER] join {:?}", item_value);
if !CONFIG.common.meta_store_external && !broadcasted {
if !cfg.common.meta_store_external && !broadcasted {
// The ingester need broadcast local file list to the new node
if is_ingester(&LOCAL_NODE_ROLE)
&& (item_value.status.eq(&NodeStatus::Prepare)
@@ -297,12 +305,13 @@ async fn watch_node_list() -> Result<()> {
}
async fn check_nodes_status(client: &reqwest::Client) -> Result<()> {
let cfg = get_config();
let nodes = get_cached_online_nodes().await.unwrap_or_default();
for node in nodes {
if node.uuid.eq(LOCAL_NODE_UUID.as_str()) {
continue;
}
let url = format!("{}{}/healthz", node.http_addr, CONFIG.common.base_uri);
let url = format!("{}{}/healthz", node.http_addr, cfg.common.base_uri);
let resp = client.get(url).timeout(HEALTH_CHECK_TIMEOUT).send().await;
if resp.is_err() || !resp.unwrap().status().is_success() {
log::error!("[CLUSTER] node {} health check failed", node.name);
@@ -350,14 +359,15 @@ pub async fn get_cached_nodes(cond: fn(&Node) -> bool) -> Option<Vec<Node>> {
#[inline(always)]
pub fn load_local_mode_node() -> Node {
let cfg = get_config();
Node {
id: 1,
uuid: LOCAL_NODE_UUID.clone(),
name: CONFIG.common.instance_name.clone(),
http_addr: format!("http://127.0.0.1:{}", CONFIG.http.port),
grpc_addr: format!("http://127.0.0.1:{}", CONFIG.grpc.port),
name: cfg.common.instance_name.clone(),
http_addr: format!("http://127.0.0.1:{}", cfg.http.port),
grpc_addr: format!("http://127.0.0.1:{}", cfg.grpc.port),
role: [Role::All].to_vec(),
cpu_num: CONFIG.limit.cpu_num as u64,
cpu_num: cfg.limit.cpu_num as u64,
status: NodeStatus::Online,
scheduled: true,
broadcasted: false,
src/common/infra/cluster/nats.rs
+14 -12
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -17,9 +17,9 @@ use core::cmp::min;
use config::{
cluster::*,
get_config,
meta::cluster::{Node, NodeStatus, Role},
utils::json,
CONFIG,
};
use infra::{
db::{get_coordinator, NEED_WATCH},
@@ -46,7 +46,7 @@ pub(crate) async fn register_and_keepalive() -> Result<()> {
}
}
// after the node is online, keepalive
let ttl_keep_alive = min(10, (CONFIG.limit.node_heartbeat_ttl / 2) as u64);
let ttl_keep_alive = min(10, (get_config().limit.node_heartbeat_ttl / 2) as u64);
loop {
time::sleep(time::Duration::from_secs(ttl_keep_alive)).await;
loop {
@@ -72,8 +72,9 @@ pub(crate) async fn register_and_keepalive() -> Result<()> {
/// Register to cluster
async fn register() -> Result<()> {
let cfg = get_config();
// 1. create a cluster lock for node register
let locker = dist_lock::lock("/nodes/register", CONFIG.limit.node_heartbeat_ttl as u64).await?;
let locker = dist_lock::lock("/nodes/register", cfg.limit.node_heartbeat_ttl as u64).await?;
// 2. watch node list
task::spawn(async move { super::watch_node_list().await });
@@ -125,11 +126,11 @@ async fn register() -> Result<()> {
let node = Node {
id: new_node_id,
uuid: LOCAL_NODE_UUID.clone(),
name: CONFIG.common.instance_name.clone(),
http_addr: format!("http://{}:{}", get_local_http_ip(), CONFIG.http.port),
grpc_addr: format!("http://{}:{}", get_local_grpc_ip(), CONFIG.grpc.port),
name: cfg.common.instance_name.clone(),
http_addr: format!("http://{}:{}", get_local_http_ip(), cfg.http.port),
grpc_addr: format!("http://{}:{}", get_local_grpc_ip(), cfg.grpc.port),
role: LOCAL_NODE_ROLE.clone(),
cpu_num: CONFIG.limit.cpu_num as u64,
cpu_num: cfg.limit.cpu_num as u64,
status: NodeStatus::Prepare,
scheduled: true,
broadcasted: false,
@@ -172,6 +173,7 @@ pub(crate) async fn set_offline() -> Result<()> {
/// set online to cluster
pub(crate) async fn set_status(status: NodeStatus) -> Result<()> {
let cfg = get_config();
// set node status to online
let node = match super::NODES.read().await.get(LOCAL_NODE_UUID.as_str()) {
Some(node) => {
@@ -182,11 +184,11 @@ pub(crate) async fn set_status(status: NodeStatus) -> Result<()> {
None => Node {
id: unsafe { LOCAL_NODE_ID },
uuid: LOCAL_NODE_UUID.clone(),
name: CONFIG.common.instance_name.clone(),
http_addr: format!("http://{}:{}", get_local_node_ip(), CONFIG.http.port),
grpc_addr: format!("http://{}:{}", get_local_node_ip(), CONFIG.grpc.port),
name: cfg.common.instance_name.clone(),
http_addr: format!("http://{}:{}", get_local_node_ip(), cfg.http.port),
grpc_addr: format!("http://{}:{}", get_local_node_ip(), cfg.grpc.port),
role: LOCAL_NODE_ROLE.clone(),
cpu_num: CONFIG.limit.cpu_num as u64,
cpu_num: cfg.limit.cpu_num as u64,
status: status.clone(),
scheduled: true,
broadcasted: false,
src/common/infra/mod.rs
+3 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -13,7 +13,7 @@
// 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 ::config::{ider, INSTANCE_ID};
use ::config::{cache_instance_id, ider};
use crate::service::db::instance;
@@ -32,7 +32,7 @@ pub async fn init() -> Result<(), anyhow::Error> {
id
}
};
INSTANCE_ID.insert("instance_id".to_owned(), instance_id);
cache_instance_id(&instance_id);
wal::init().await?;
// because of asynchronous, we need to wait for a while
src/common/infra/wal.rs
+10 -10
View File
@@ -19,11 +19,11 @@ use arrow::{ipc::writer::StreamWriter, record_batch::RecordBatch};
use arrow_schema::Schema;
use chrono::{DateTime, Datelike, TimeZone, Utc};
use config::{
ider,
get_config, ider,
meta::stream::{PartitionTimeLevel, StreamType},
metrics,
utils::asynchronism::file::get_file_contents,
CONFIG, FILE_EXT_ARROW, FILE_EXT_JSON,
FILE_EXT_ARROW, FILE_EXT_JSON,
};
use hashbrown::HashMap;
use once_cell::sync::Lazy;
@@ -147,7 +147,7 @@ impl Default for Manager {
impl Manager {
pub fn new() -> Manager {
let size = CONFIG.limit.cpu_num;
let size = get_config().limit.cpu_num;
let mut data = Vec::with_capacity(size);
for _ in 0..size {
data.push(RwLock::new(HashMap::<String, Arc<RwFile>>::default()));
@@ -178,7 +178,7 @@ impl Manager {
drop(manager);
// check size & ttl
if file.size().await >= (CONFIG.limit.max_file_size_on_disk as i64)
if file.size().await >= (get_config().limit.max_file_size_on_disk as i64)
|| file.expired() <= Utc::now().timestamp()
{
let mut manager = locker.write().await;
@@ -258,16 +258,16 @@ impl RwFile {
schema: Option<Schema>,
) -> RwFile {
let use_arrow = schema.is_some();
let cfg = get_config();
let mut dir_path = if use_arrow {
format!(
"{}files/{}/{}/{}/",
&CONFIG.common.data_idx_dir, stream.org_id, stream.stream_type, stream.stream_name
&cfg.common.data_idx_dir, stream.org_id, stream.stream_type, stream.stream_name
)
} else {
format!(
"{}files/{}/{}/{}/",
&CONFIG.common.data_wal_dir, stream.org_id, stream.stream_type, stream.stream_name
&cfg.common.data_wal_dir, stream.org_id, stream.stream_type, stream.stream_name
)
};
// Hack for file_list
@@ -308,7 +308,7 @@ impl RwFile {
let time_now: DateTime<Utc> = Utc::now();
let level_duration = partition_time_level.unwrap_or_default().duration();
let ttl = if !CONFIG.limit.ignore_file_retention_by_stream && level_duration > 0 {
let ttl = if !cfg.limit.ignore_file_retention_by_stream && level_duration > 0 {
let time_end_day = Utc
.with_ymd_and_hms(
time_now.year(),
@@ -324,12 +324,12 @@ impl RwFile {
if expired > time_end_day {
// if the file expired time is tomorrow, it should be deleted at 23:59:59 +
// 10min
time_end_day + CONFIG.limit.max_file_retention_time as i64
time_end_day + cfg.limit.max_file_retention_time as i64
} else {
expired
}
} else {
time_now.timestamp() + CONFIG.limit.max_file_retention_time as i64
time_now.timestamp() + cfg.limit.max_file_retention_time as i64
};
RwFile {
src/common/meta/alerts/destinations.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/alerts/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/alerts/templates.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/authz.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/dashboards/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/dashboards/reports.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/dashboards/v1/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/dashboards/v2/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/dashboards/v3/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/functions.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/ingestion.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/maxmind/client.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/maxmind/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/middleware_data.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/organization.rs
+2 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -13,7 +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 config::CONFIG;
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
@@ -99,7 +98,7 @@ pub struct RumIngestionResponse {
}
fn default_scrape_interval() -> u32 {
CONFIG.common.default_scrape_interval
config::get_config().common.default_scrape_interval
}
#[derive(Serialize, ToSchema, Deserialize, Debug, Clone)]
src/common/meta/prom.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/saved_view.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/service.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/syslog.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/telemetry.rs
+10 -11
View File
@@ -17,8 +17,9 @@ use std::collections::HashMap;
use config::{
cluster::{is_single_node, load_local_node_role},
get_config, get_instance_id,
utils::json,
CONFIG, INSTANCE_ID, SIZE_IN_MB, TELEMETRY_CLIENT,
SIZE_IN_MB, TELEMETRY_CLIENT,
};
use hashbrown::HashSet;
use infra::{cache::stats, db as infra_db, schema::STREAM_SCHEMAS_LATEST};
@@ -53,7 +54,7 @@ impl Telemetry {
data: Option<HashMap<String, json::Value>>,
send_zo_data: bool,
) {
if !CONFIG.common.telemetry_enabled {
if !get_config().common.telemetry_enabled {
return;
}
log::info!("sending event {}", event);
@@ -69,7 +70,7 @@ impl Telemetry {
self.add_event(Track {
user: segment::message::User::UserId {
user_id: segment::message::User::AnonymousId {
anonymous_id: INSTANCE_ID.get("instance_id").unwrap().to_string(),
anonymous_id: get_instance_id(),
}
.to_string(),
},
@@ -80,10 +81,7 @@ impl Telemetry {
});
let res = TELEMETRY_CLIENT
.send(
INSTANCE_ID.get("instance_id").unwrap().to_string(),
Message::from(self.event.clone()),
)
.send(get_instance_id(), Message::from(self.event.clone()))
.await;
if res.is_err() {
@@ -109,7 +107,7 @@ pub fn get_base_info(data: &mut HashMap<String, json::Value>) -> HashMap<String,
);
data.insert(
"host_name".to_string(),
CONFIG.common.instance_name.clone().into(),
get_config().common.instance_name.clone().into(),
);
data.insert("zo_version".to_string(), VERSION.to_owned().into());
@@ -138,6 +136,7 @@ pub async fn add_zo_info(mut data: HashMap<String, json::Value>) -> HashMap<Stri
}
drop(r);
let cfg = get_config();
data.insert("num_org".to_string(), orgs.len().into());
data.insert("num_streams".to_string(), num_streams.into());
data.insert("num_logs_streams".to_string(), logs_streams.into());
@@ -145,12 +144,12 @@ pub async fn add_zo_info(mut data: HashMap<String, json::Value>) -> HashMap<Stri
data.insert("num_users".to_string(), USERS.len().into());
data.insert(
"is_local_mode".to_string(),
json::Value::Bool(CONFIG.common.local_mode),
json::Value::Bool(cfg.common.local_mode),
);
if CONFIG.common.local_mode {
if cfg.common.local_mode {
data.insert(
"local_mode_storage".to_string(),
CONFIG.common.local_mode_storage.clone().into(),
cfg.common.local_mode_storage.clone().into(),
);
}
let db = infra_db::get_db().await;
src/common/meta/traces.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/meta/user.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/migration/file_list.rs
+2 -2
View File
@@ -14,9 +14,9 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use config::{
get_config,
meta::stream::{FileKey, PartitionTimeLevel, StreamType},
utils::{file::get_file_meta, time::BASE_TIME},
CONFIG,
};
use infra::file_list as infra_file_list;
@@ -26,7 +26,7 @@ use crate::{
};
pub async fn run(prefix: &str, from: &str, to: &str) -> Result<(), anyhow::Error> {
if get_file_meta(&CONFIG.common.data_wal_dir).is_err() {
if get_file_meta(&get_config().common.data_wal_dir).is_err() {
// there is no local wal files, no need upgrade
return Ok(());
}
src/common/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/utils/auth.rs
+7 -8
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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,8 +16,6 @@
use actix_web::{dev::Payload, Error, FromRequest, HttpRequest};
use argon2::{password_hash::SaltString, Algorithm, Argon2, Params, PasswordHasher, Version};
use config::utils::json;
#[cfg(feature = "enterprise")]
use config::CONFIG;
use futures::future::{ready, Ready};
#[cfg(feature = "enterprise")]
@@ -177,11 +175,12 @@ impl FromRequest for AuthExtractor {
let mut method = req.method().to_string();
let local_path = req.path().to_string();
let path =
match local_path.strip_prefix(format!("{}/api/", CONFIG.common.base_uri).as_str()) {
Some(path) => path,
None => &local_path,
};
let path = match local_path
.strip_prefix(format!("{}/api/", config::get_config().common.base_uri).as_str())
{
Some(path) => path,
None => &local_path,
};
let path_columns = path.split('/').collect::<Vec<&str>>();
let url_len = path_columns.len();
src/common/utils/functions.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/utils/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/utils/record_batch_parquet_ext.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/common/utils/stream.rs
+4 -3
View File
@@ -20,9 +20,10 @@ use std::{
use actix_web::HttpResponse;
use config::{
get_config,
meta::stream::{FileMeta, StreamType},
utils::{arrow::record_batches_to_json_rows, json},
CONFIG, FILE_EXT_JSON,
FILE_EXT_JSON,
};
use datafusion::{
arrow::{datatypes::Schema, record_batch::RecordBatch},
@@ -60,7 +61,7 @@ pub fn get_file_name_v1(org_id: &str, stream_name: &str, suffix: u32) -> String
// "./data/openobserve/olympics/olympics#2022#09#13#13_1.json"
format!(
"{}{}/{}/{}/{}_{}{}",
&CONFIG.common.data_wal_dir,
get_config().common.data_wal_dir,
org_id,
StreamType::Logs,
stream_name,
@@ -84,7 +85,7 @@ pub async fn populate_file_meta(
let sql = format!(
"SELECT min({0}) as min, max({0}) as max, count({0}) as num_records FROM temp;",
CONFIG.common.column_timestamp
get_config().common.column_timestamp
);
let df = ctx.sql(sql.as_str()).await?;
let batches = df.collect().await?;
src/common/utils/zo_logger.rs
+8 -7
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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,7 +16,7 @@
use std::sync::Arc;
use cloudevents::{Event, EventBuilder, EventBuilderV10};
use config::{ider, utils::json, CONFIG};
use config::{get_config, ider, utils::json};
use log::{Metadata, Record};
use once_cell::sync::Lazy;
use reqwest::Client;
@@ -35,7 +35,7 @@ pub struct ZoLogger {
impl log::Log for ZoLogger {
fn enabled(&self, _metadata: &Metadata) -> bool {
CONFIG.log.events_enabled
get_config().log.events_enabled
}
fn log(self: &ZoLogger, record: &Record) {
@@ -59,7 +59,7 @@ impl log::Log for ZoLogger {
let event = EventBuilderV10::new()
.id(ider::generate())
.ty("debug_log")
.source(CONFIG.common.instance_name.clone())
.source(get_config().common.instance_name.clone())
.data("application/json", payload)
.build()
.unwrap();
@@ -78,10 +78,11 @@ pub async fn send_logs() {
let mut logs = LOGS.write().await;
logs.push(val);
if logs.len() >= CONFIG.log.events_batch_size {
let cfg = get_config();
if logs.len() >= cfg.log.events_batch_size {
let old_logs = std::mem::take(&mut *logs);
let url = url::Url::parse(&CONFIG.log.events_url).unwrap();
let auth = format!("Basic {}", &CONFIG.log.events_auth);
let url = url::Url::parse(&cfg.log.events_url).unwrap();
let auth = format!("Basic {}", &cfg.log.events_auth);
let cl = Arc::clone(&cl);
tokio::task::spawn(async move {
let _ = cl
src/config/Cargo.toml
+1
View File
@@ -8,6 +8,7 @@ license.workspace = true
actix-web-prometheus.workspace = true
ahash.workspace = true
anyhow.workspace = true
arc-swap.workspace = true
arrow.workspace = true
arrow-json.workspace = true
arrow-schema.workspace = true
src/config/src/cluster.rs
+9 -8
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -18,9 +18,8 @@ use std::net::IpAddr;
use once_cell::sync::Lazy;
use crate::{
ider,
get_config, ider,
meta::cluster::{NodeStatus, Role},
CONFIG,
};
pub static mut LOCAL_NODE_ID: i32 = 0;
@@ -36,8 +35,9 @@ pub fn load_local_node_uuid() -> String {
#[inline(always)]
pub fn get_local_http_ip() -> String {
if !CONFIG.http.addr.is_empty() {
CONFIG.http.addr.clone()
let cfg = get_config();
if !cfg.http.addr.is_empty() {
cfg.http.addr.clone()
} else {
get_local_node_ip()
}
@@ -45,8 +45,9 @@ pub fn get_local_http_ip() -> String {
#[inline(always)]
pub fn get_local_grpc_ip() -> String {
if !CONFIG.grpc.addr.is_empty() {
CONFIG.grpc.addr.clone()
let cfg = get_config();
if !cfg.grpc.addr.is_empty() {
cfg.grpc.addr.clone()
} else {
get_local_node_ip()
}
@@ -64,7 +65,7 @@ pub fn get_local_node_ip() -> String {
#[inline(always)]
pub fn load_local_node_role() -> Vec<Role> {
CONFIG
get_config()
.common
.node_role
.clone()
src/config/src/config.rs
+59 -30
View File
@@ -13,8 +13,9 @@
// 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::{cmp::max, collections::BTreeMap, path::Path, time::Duration};
use std::{cmp::max, collections::BTreeMap, path::Path, sync::Arc, time::Duration};
use arc_swap::ArcSwap;
use chromiumoxide::{browser::BrowserConfig, handler::viewport::Viewport};
use dotenv_config::EnvConfig;
use dotenvy::dotenv;
@@ -69,7 +70,7 @@ pub static SQL_FULL_TEXT_SEARCH_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
_DEFAULT_SQL_FULL_TEXT_SEARCH_FIELDS
.iter()
.map(|s| s.to_string()),
CONFIG
get_config()
.common
.feature_fulltext_extra_fields
.split(',')
@@ -89,7 +90,7 @@ pub static SQL_FULL_TEXT_SEARCH_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
});
pub static QUICK_MODEL_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
let mut fields = CONFIG
let mut fields = get_config()
.common
.feature_quick_mode_fields
.split(',')
@@ -111,7 +112,7 @@ const _DEFAULT_DISTINCT_FIELDS: [&str; 2] = ["service_name", "operation_name"];
pub static DISTINCT_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
let mut fields = chain(
_DEFAULT_DISTINCT_FIELDS.iter().map(|s| s.to_string()),
CONFIG
get_config()
.common
.feature_distinct_extra_fields
.split(',')
@@ -134,7 +135,7 @@ const _DEFAULT_BLOOM_FILTER_FIELDS: [&str; 1] = ["trace_id"];
pub static BLOOM_FILTER_DEFAULT_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
let mut fields = chain(
_DEFAULT_BLOOM_FILTER_FIELDS.iter().map(|s| s.to_string()),
CONFIG
get_config()
.common
.bloom_filter_default_fields
.split(',')
@@ -153,8 +154,8 @@ pub static BLOOM_FILTER_DEFAULT_FIELDS: Lazy<Vec<String>> = Lazy::new(|| {
fields
});
pub static CONFIG: Lazy<Config> = Lazy::new(init);
pub static INSTANCE_ID: Lazy<RwHashMap<String, String>> = Lazy::new(Default::default);
static CONFIG: Lazy<ArcSwap<Config>> = Lazy::new(|| ArcSwap::from(Arc::new(init())));
static INSTANCE_ID: Lazy<RwHashMap<String, String>> = Lazy::new(Default::default);
pub static TELEMETRY_CLIENT: Lazy<segment::HttpClient> = Lazy::new(|| {
segment::HttpClient::new(
@@ -162,10 +163,27 @@ pub static TELEMETRY_CLIENT: Lazy<segment::HttpClient> = Lazy::new(|| {
.connect_timeout(Duration::new(10, 0))
.build()
.unwrap(),
CONFIG.common.telemetry_url.clone(),
CONFIG.load().common.telemetry_url.clone(),
)
});
pub fn get_config() -> Arc<Config> {
CONFIG.load().clone()
}
pub fn refresh_config() -> Result<(), anyhow::Error> {
CONFIG.store(Arc::new(init()));
Ok(())
}
pub fn cache_instance_id(instance_id: &str) {
INSTANCE_ID.insert("instance_id".to_owned(), instance_id.to_owned());
}
pub fn get_instance_id() -> String {
INSTANCE_ID.get("instance_id").unwrap().clone()
}
static CHROME_LAUNCHER_OPTIONS: tokio::sync::OnceCell<Option<BrowserConfig>> =
tokio::sync::OnceCell::const_new();
@@ -176,31 +194,32 @@ pub async fn get_chrome_launch_options() -> &'static Option<BrowserConfig> {
}
async fn init_chrome_launch_options() -> Option<BrowserConfig> {
if !CONFIG.chrome.chrome_enabled || !CONFIG.common.report_server_url.is_empty() {
let cfg = get_config();
if !cfg.chrome.chrome_enabled || !cfg.common.report_server_url.is_empty() {
None
} else {
let mut browser_config = BrowserConfig::builder()
.window_size(
CONFIG.chrome.chrome_window_width,
CONFIG.chrome.chrome_window_height,
cfg.chrome.chrome_window_width,
cfg.chrome.chrome_window_height,
)
.viewport(Viewport {
width: CONFIG.chrome.chrome_window_width,
height: CONFIG.chrome.chrome_window_height,
width: cfg.chrome.chrome_window_width,
height: cfg.chrome.chrome_window_height,
device_scale_factor: Some(1.0),
..Viewport::default()
});
if CONFIG.chrome.chrome_with_head {
if cfg.chrome.chrome_with_head {
browser_config = browser_config.with_head();
}
if CONFIG.chrome.chrome_no_sandbox {
if cfg.chrome.chrome_no_sandbox {
browser_config = browser_config.no_sandbox();
}
if !CONFIG.chrome.chrome_path.is_empty() {
browser_config = browser_config.chrome_executable(CONFIG.chrome.chrome_path.as_str());
if !cfg.chrome.chrome_path.is_empty() {
browser_config = browser_config.chrome_executable(cfg.chrome.chrome_path.as_str());
} else {
panic!("Chrome path must be specified");
}
@@ -209,15 +228,16 @@ async fn init_chrome_launch_options() -> Option<BrowserConfig> {
}
pub static SMTP_CLIENT: Lazy<Option<AsyncSmtpTransport<Tokio1Executor>>> = Lazy::new(|| {
if !CONFIG.smtp.smtp_enabled {
let cfg = get_config();
if !cfg.smtp.smtp_enabled {
None
} else {
let tls_parameters = TlsParameters::new(CONFIG.smtp.smtp_host.clone()).unwrap();
let tls_parameters = TlsParameters::new(cfg.smtp.smtp_host.clone()).unwrap();
let mut transport_builder =
AsyncSmtpTransport::<Tokio1Executor>::builder_dangerous(&CONFIG.smtp.smtp_host)
.port(CONFIG.smtp.smtp_port);
AsyncSmtpTransport::<Tokio1Executor>::builder_dangerous(&cfg.smtp.smtp_host)
.port(cfg.smtp.smtp_port);
let option = &CONFIG.smtp.smtp_encryption;
let option = &cfg.smtp.smtp_encryption;
transport_builder = if option == "starttls" {
transport_builder.tls(Tls::Required(tls_parameters))
} else if option == "ssltls" {
@@ -226,18 +246,25 @@ pub static SMTP_CLIENT: Lazy<Option<AsyncSmtpTransport<Tokio1Executor>>> = Lazy:
transport_builder
};
if !CONFIG.smtp.smtp_username.is_empty() && !CONFIG.smtp.smtp_password.is_empty() {
if !cfg.smtp.smtp_username.is_empty() && !cfg.smtp.smtp_password.is_empty() {
transport_builder = transport_builder.credentials(Credentials::new(
CONFIG.smtp.smtp_username.clone(),
CONFIG.smtp.smtp_password.clone(),
cfg.smtp.smtp_username.clone(),
cfg.smtp.smtp_password.clone(),
));
}
Some(transport_builder.build())
}
});
pub static BLOCKED_STREAMS: Lazy<Vec<&str>> =
Lazy::new(|| CONFIG.common.blocked_streams.split(',').collect());
pub static BLOCKED_STREAMS: Lazy<Vec<String>> = Lazy::new(|| {
let blocked_streams = get_config()
.common
.blocked_streams
.split(',')
.map(|x| x.to_string())
.collect();
blocked_streams
});
#[derive(EnvConfig)]
pub struct Config {
@@ -1448,13 +1475,15 @@ fn check_s3_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
#[inline]
pub fn is_local_disk_storage() -> bool {
CONFIG.common.local_mode && CONFIG.common.local_mode_storage.eq("disk")
let cfg = get_config();
cfg.common.local_mode && cfg.common.local_mode_storage.eq("disk")
}
#[inline]
pub fn get_cluster_name() -> String {
if !CONFIG.common.cluster_name.is_empty() {
CONFIG.common.cluster_name.to_string()
let cfg = get_config();
if !cfg.common.cluster_name.is_empty() {
cfg.common.cluster_name.to_string()
} else {
INSTANCE_ID.get("instance_id").unwrap().to_string()
}
src/config/src/ider.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/lib.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/meta/cluster.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/meta/logger.rs
+4 -5
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -27,16 +27,15 @@ use tracing_subscriber::{
registry::LookupSpan,
};
use crate::CONFIG;
pub struct CustomTimeFormat;
impl FormatTime for CustomTimeFormat {
fn format_time(&self, w: &mut Writer<'_>) -> std::fmt::Result {
if CONFIG.log.local_time_format.is_empty() {
let cfg = crate::get_config();
if cfg.log.local_time_format.is_empty() {
write!(w, "{}", Utc::now().to_rfc3339())
} else {
write!(w, "{}", Local::now().format(&CONFIG.log.local_time_format))
write!(w, "{}", Local::now().format(&cfg.log.local_time_format))
}
}
}
src/config/src/meta/meta_store.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/meta/sql.rs
+3 -3
View File
@@ -25,7 +25,7 @@ use sqlparser::{
parser::Parser,
};
use crate::CONFIG;
use crate::get_config;
const MAX_LIMIT: usize = 100000;
const MAX_OFFSET: usize = 100000;
@@ -313,7 +313,7 @@ impl<'a> TryFrom<Timerange<'a>> for Option<(i64, i64)> {
Some(expr) => parse_expr_for_field(
expr,
&SqlOperator::And,
&CONFIG.common.column_timestamp,
&get_config().common.column_timestamp,
&mut fields,
)?,
None => {}
@@ -544,7 +544,7 @@ fn parse_expr_check_field_name(s: &str, field: &str) -> bool {
if s == field {
return true;
}
if field == "*" && s != "_all" && s != CONFIG.common.column_timestamp.clone() {
if field == "*" && s != "_all" && s != get_config().common.column_timestamp.clone() {
return true;
}
src/config/src/meta/stream.rs
+3 -4
View File
@@ -24,12 +24,11 @@ use utoipa::ToSchema;
use super::usage::Stats;
use crate::{
get_config,
utils::{
hash::{gxhash, Sum64},
json,
json::{Map, Value},
json::{self, Map, Value},
},
CONFIG,
};
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize, ToSchema, Hash)]
@@ -223,7 +222,7 @@ impl StreamStats {
fn time_range(&self) -> (i64, i64) {
assert!(self.doc_time_min <= self.doc_time_max);
let file_push_interval = Duration::try_seconds(CONFIG.limit.file_push_interval as _)
let file_push_interval = Duration::try_seconds(get_config().limit.file_push_interval as _)
.unwrap()
.num_microseconds()
.unwrap();
src/config/src/meta/usage.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/metrics.rs
+6 -7
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -21,8 +21,6 @@ use prometheus::{
CounterVec, HistogramOpts, HistogramVec, IntCounterVec, IntGaugeVec, Opts, Registry,
};
use crate::CONFIG;
pub const NAMESPACE: &str = "zo";
const HELP_SUFFIX: &str =
"Please include 'organization, 'stream type', and 'stream' labels for this metric.";
@@ -682,10 +680,11 @@ fn register_metrics(registry: &Registry) {
}
fn create_const_labels() -> HashMap<String, String> {
let cfg = crate::config::get_config();
let mut labels = HashMap::new();
labels.insert("cluster".to_string(), CONFIG.common.cluster_name.clone());
labels.insert("instance".to_string(), CONFIG.common.instance_name.clone());
labels.insert("role".to_string(), CONFIG.common.node_role.clone());
labels.insert("cluster".to_string(), cfg.common.cluster_name.clone());
labels.insert("instance".to_string(), cfg.common.instance_name.clone());
labels.insert("role".to_string(), cfg.common.node_role.clone());
labels
}
@@ -694,7 +693,7 @@ pub fn create_prometheus_handler() -> PrometheusMetrics {
register_metrics(&registry);
PrometheusMetricsBuilder::new(NAMESPACE)
.endpoint(format!("{}/metrics", CONFIG.common.base_uri).as_str())
.endpoint(format!("{}/metrics", crate::config::get_config().common.base_uri).as_str())
.const_labels(create_const_labels())
.registry(registry)
.build()
src/config/src/utils/asynchronism/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/cgroup.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/flatten.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/hash/cityhash.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/hash/fnv.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/hash/gxhash.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/hash/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/hash/murmur3.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/parquet.rs
+7 -6
View File
@@ -33,8 +33,9 @@ pub fn new_parquet_writer<'a>(
full_text_search_fields: &'a [String],
metadata: &'a FileMeta,
) -> AsyncArrowWriter<&'a mut Vec<u8>> {
let row_group_size = if CONFIG.limit.parquet_max_row_group_size > 0 {
CONFIG.limit.parquet_max_row_group_size
let cfg = get_config();
let row_group_size = if cfg.limit.parquet_max_row_group_size > 0 {
cfg.limit.parquet_max_row_group_size
} else {
PARQUET_MAX_ROW_GROUP_SIZE
};
@@ -46,11 +47,11 @@ pub fn new_parquet_writer<'a>(
.set_dictionary_enabled(true)
.set_encoding(Encoding::PLAIN)
.set_column_dictionary_enabled(
CONFIG.common.column_timestamp.as_str().into(),
cfg.common.column_timestamp.as_str().into(),
false,
)
.set_column_encoding(
CONFIG.common.column_timestamp.as_str().into(),
cfg.common.column_timestamp.as_str().into(),
Encoding::DELTA_BINARY_PACKED,
)
.set_key_value_metadata(Some(vec![
@@ -70,9 +71,9 @@ pub fn new_parquet_writer<'a>(
}
// Bloom filter stored by row_group, so if the num_rows can limit to row_group_size
let num_rows = min(metadata.records as u64, row_group_size as u64);
if CONFIG.common.bloom_filter_enabled {
if cfg.common.bloom_filter_enabled {
// if bloom_filter_on_all_fields is true, then use all string fields
let fields = if CONFIG.common.bloom_filter_on_all_fields {
let fields = if cfg.common.bloom_filter_on_all_fields {
schema
.fields()
.iter()
src/config/src/utils/rand.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/schema.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/schema_ext.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/str.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/config/src/utils/time.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/grpc/auth/mod.rs
+9 -11
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -13,7 +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 config::CONFIG;
use http_auth_basic::Credentials;
use tonic::{metadata::MetadataValue, Request, Status};
@@ -26,10 +25,9 @@ use crate::common::{
};
pub fn check_auth(req: Request<()>) -> Result<Request<()>, Status> {
let cfg = config::get_config();
let metadata = req.metadata();
if !metadata.contains_key(&CONFIG.grpc.org_header_key)
&& !metadata.contains_key("authorization")
{
if !metadata.contains_key(&cfg.grpc.org_header_key) && !metadata.contains_key("authorization") {
return Err(Status::unauthenticated("No valid auth token"));
}
@@ -43,11 +41,11 @@ pub fn check_auth(req: Request<()>) -> Result<Request<()>, Status> {
if token.eq(get_internal_grpc_token().as_str()) {
Ok(req)
} else {
let org_id = metadata.get(&CONFIG.grpc.org_header_key);
let org_id = metadata.get(&cfg.grpc.org_header_key);
if org_id.is_none() {
return Err(Status::invalid_argument(format!(
"Please specify organization id with header key '{}' ",
&CONFIG.grpc.org_header_key
&cfg.grpc.org_header_key
)));
}
@@ -92,14 +90,14 @@ pub fn check_auth(req: Request<()>) -> Result<Request<()>, Status> {
#[cfg(test)]
mod tests {
use config::INSTANCE_ID;
use config::cache_instance_id;
use super::*;
use crate::common::meta::user::User;
#[tokio::test]
async fn test_check_no_auth() {
INSTANCE_ID.insert("instance_id".to_owned(), "instance".to_string());
cache_instance_id("instance");
ROOT_USER.insert(
"root".to_string(),
User {
@@ -128,7 +126,7 @@ mod tests {
#[tokio::test]
async fn test_check_auth() {
INSTANCE_ID.insert("instance_id".to_owned(), "instance".to_string());
cache_instance_id("instance");
ROOT_USER.insert(
"root".to_string(),
User {
@@ -155,7 +153,7 @@ mod tests {
#[tokio::test]
async fn test_check_err_auth() {
INSTANCE_ID.insert("instance_id".to_owned(), "instance".to_string());
cache_instance_id("instance");
ROOT_USER.insert(
"root".to_string(),
User {
src/handler/grpc/request/event.rs
+6 -6
View File
@@ -16,10 +16,10 @@
use chrono::{Duration, Utc};
use config::{
cluster::{is_compactor, is_querier, LOCAL_NODE_ROLE, LOCAL_NODE_UUID},
get_config,
meta::{cluster::Role, stream::FileKey},
metrics,
utils::parquet::read_recordbatch_from_bytes,
CONFIG,
};
use hashbrown::HashSet;
use infra::{file_list as infra_file_list, schema::STREAM_SCHEMAS_FIELDS};
@@ -57,11 +57,11 @@ impl Event for Eventer {
.filter(|v| v.deleted)
.map(|v| v.key.clone())
.collect::<Vec<_>>();
let cfg = get_config();
// Warning: external meta store should not accept any file list
// querier and compactor can accept add new files
// ingester only accept remove old files
if !CONFIG.common.meta_store_external {
if !cfg.common.meta_store_external {
if is_querier(&LOCAL_NODE_ROLE) || is_compactor(&LOCAL_NODE_ROLE) {
if let Err(e) = infra_file_list::batch_add(&put_items).await {
// metrics
@@ -89,7 +89,7 @@ impl Event for Eventer {
}
// cache latest files for querier
if CONFIG.memory_cache.cache_latest_files && is_querier(&LOCAL_NODE_ROLE) {
if cfg.memory_cache.cache_latest_files && is_querier(&LOCAL_NODE_ROLE) {
let mut cached_field_stream = HashSet::new();
for item in put_items.iter() {
let Some(node) = get_node_from_consistent_hash(&item.key, &Role::Querier).await
@@ -102,7 +102,7 @@ impl Event for Eventer {
if infra::cache::file_data::download("download", &item.key)
.await
.is_ok()
&& CONFIG.limit.quick_mode_file_list_enabled
&& cfg.limit.quick_mode_file_list_enabled
{
let columns = item.key.split('/').collect::<Vec<&str>>();
if columns[2] != "logs" {
@@ -138,7 +138,7 @@ async fn cache_latest_fields(stream: &str, file: &str) -> Result<(), anyhow::Err
drop(fr);
if field_cache_time
+ Duration::try_seconds(CONFIG.limit.quick_mode_file_list_interval)
+ Duration::try_seconds(get_config().limit.quick_mode_file_list_interval)
.unwrap()
.num_microseconds()
.unwrap()
src/handler/grpc/request/file_list.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/grpc/request/logs.rs
+6 -6
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -14,7 +14,6 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use async_trait::async_trait;
use config::CONFIG;
use opentelemetry_proto::tonic::collector::logs::v1::{
logs_service_server::LogsService, ExportLogsServiceRequest, ExportLogsServiceResponse,
};
@@ -29,21 +28,22 @@ impl LogsService for LogsServer {
&self,
request: tonic::Request<ExportLogsServiceRequest>,
) -> Result<tonic::Response<ExportLogsServiceResponse>, tonic::Status> {
let cfg = config::get_config();
let metadata = request.metadata().clone();
let msg = format!(
"Please specify organization id with header key '{}' ",
&CONFIG.grpc.org_header_key
&cfg.grpc.org_header_key
);
if !metadata.contains_key(&CONFIG.grpc.org_header_key) {
if !metadata.contains_key(&cfg.grpc.org_header_key) {
return Err(Status::invalid_argument(msg));
}
let in_req = request.into_inner();
let org_id = metadata.get(&CONFIG.grpc.org_header_key);
let org_id = metadata.get(&cfg.grpc.org_header_key);
if org_id.is_none() {
return Err(Status::invalid_argument(msg));
}
let stream_name = metadata.get(&CONFIG.grpc.stream_header_key);
let stream_name = metadata.get(&cfg.grpc.stream_header_key);
let mut in_stream_name: Option<&str> = None;
if let Some(stream_name) = stream_name {
in_stream_name = Some(stream_name.to_str().unwrap());
src/handler/grpc/request/metrics/ingester.rs
+5 -5
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -14,7 +14,6 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use async_trait::async_trait;
use config::CONFIG;
use opentelemetry_proto::tonic::collector::metrics::v1::{
metrics_service_server::MetricsService, ExportMetricsServiceRequest,
ExportMetricsServiceResponse,
@@ -30,17 +29,18 @@ impl MetricsService for Ingester {
&self,
request: tonic::Request<ExportMetricsServiceRequest>,
) -> Result<tonic::Response<ExportMetricsServiceResponse>, tonic::Status> {
let cfg = config::get_config();
let metadata = request.metadata().clone();
let msg = format!(
"Please specify organization id with header key '{}' ",
&CONFIG.grpc.org_header_key
&cfg.grpc.org_header_key
);
if !metadata.contains_key(&CONFIG.grpc.org_header_key) {
if !metadata.contains_key(&cfg.grpc.org_header_key) {
return Err(Status::invalid_argument(msg));
}
let in_req = request.into_inner();
let org_id = metadata.get(&CONFIG.grpc.org_header_key);
let org_id = metadata.get(&cfg.grpc.org_header_key);
if org_id.is_none() {
return Err(Status::invalid_argument(msg));
}
src/handler/grpc/request/metrics/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/grpc/request/metrics/querier.rs
+14 -14
View File
@@ -23,7 +23,6 @@ use config::{
parquet::{parse_time_range_from_filename, read_metadata_from_bytes},
schema_ext::SchemaExt,
},
CONFIG,
};
use infra::errors;
use opentelemetry::global;
@@ -150,20 +149,21 @@ impl Metrics for Querier {
}
// get parquet files
let wal_dir = match std::path::Path::new(&CONFIG.common.data_wal_dir).canonicalize() {
Ok(path) => {
let mut path = path.to_str().unwrap().to_string();
// Hack for windows
if path.starts_with("\\\\?\\") {
path = path[4..].to_string();
path = path.replace('\\', "/");
let wal_dir =
match std::path::Path::new(&config::get_config().common.data_wal_dir).canonicalize() {
Ok(path) => {
let mut path = path.to_str().unwrap().to_string();
// Hack for windows
if path.starts_with("\\\\?\\") {
path = path[4..].to_string();
path = path.replace('\\', "/");
}
path
}
path
}
Err(_) => {
return Ok(Response::new(resp));
}
};
Err(_) => {
return Ok(Response::new(resp));
}
};
let pattern = format!("{wal_dir}/files/{org_id}/metrics/{stream_name}/");
let files = scan_files(&pattern, "parquet", None).unwrap_or_default();
src/handler/grpc/request/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/grpc/request/traces.rs
+5 -5
View File
@@ -13,7 +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 config::CONFIG;
use opentelemetry_proto::tonic::collector::trace::v1::{
trace_service_server::TraceService, ExportTraceServiceRequest, ExportTraceServiceResponse,
};
@@ -30,22 +29,23 @@ impl TraceService for TraceServer {
&self,
request: tonic::Request<ExportTraceServiceRequest>,
) -> Result<tonic::Response<ExportTraceServiceResponse>, tonic::Status> {
let cfg = config::get_config();
let metadata = request.metadata().clone();
let msg = format!(
"Please specify organization id with header key '{}' ",
&CONFIG.grpc.org_header_key
&cfg.grpc.org_header_key
);
if !metadata.contains_key(&CONFIG.grpc.org_header_key) {
if !metadata.contains_key(&cfg.grpc.org_header_key) {
return Err(Status::invalid_argument(msg));
}
let in_req = request.into_inner();
let org_id = metadata.get(&CONFIG.grpc.org_header_key);
let org_id = metadata.get(&cfg.grpc.org_header_key);
if org_id.is_none() {
return Err(Status::invalid_argument(msg));
}
let stream_name = metadata.get(&CONFIG.grpc.stream_header_key);
let stream_name = metadata.get(&cfg.grpc.stream_header_key);
let mut in_stream_name: Option<&str> = None;
if let Some(stream_name) = stream_name {
in_stream_name = Some(stream_name.to_str().unwrap());
src/handler/grpc/request/usage.rs
+2 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -14,7 +14,6 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use async_trait::async_trait;
use config::CONFIG;
use proto::cluster_rpc::{usage_server::Usage, UsageRequest, UsageResponse};
use tonic::{Request, Response, Status};
@@ -30,7 +29,7 @@ impl Usage for UsageServerImpl {
let metadata = request.metadata().clone();
let req = request.into_inner();
let report_to_stream = req.stream_name;
let report_to_org_id = metadata.get(&CONFIG.grpc.org_header_key);
let report_to_org_id = metadata.get(&config::get_config().grpc.org_header_key);
let in_data = req.data.unwrap_or_default();
let resp = crate::service::logs::otlp_grpc::usage_ingest(
report_to_org_id.unwrap().to_str().unwrap(),
src/handler/http/auth/jwt.rs
+10 -6
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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,7 +16,7 @@
use std::str::FromStr;
#[cfg(feature = "enterprise")]
use config::CONFIG;
use config::get_config;
#[cfg(feature = "enterprise")]
use jsonwebtoken::TokenData;
#[cfg(feature = "enterprise")]
@@ -220,7 +220,7 @@ pub async fn process_token(
&org.name,
&user_email,
org.role.clone(),
&CONFIG.auth.root_user_email,
&get_config().auth.root_user_email,
)
.await
{
@@ -239,8 +239,12 @@ pub async fn process_token(
}
for org in orgs_removed {
match users::remove_user_from_org(&org.name, &user_email, &CONFIG.auth.root_user_email)
.await
match users::remove_user_from_org(
&org.name,
&user_email,
&get_config().auth.root_user_email,
)
.await
{
Ok(_) => {
log::info!("User removed from the organization {}", org.name);
@@ -264,7 +268,7 @@ pub async fn process_token(
&org.name,
&user_email,
false,
&CONFIG.auth.root_user_email,
&get_config().auth.root_user_email,
crate::common::meta::user::UpdateUser {
role: Some(org.role.clone()),
..Default::default()
src/handler/http/auth/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/auth/token.rs
+2 -4
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -20,8 +20,6 @@ use actix_web::{
http::{header, Method},
};
#[cfg(feature = "enterprise")]
use config::CONFIG;
#[cfg(feature = "enterprise")]
use o2_enterprise::enterprise::{common::infra::config::O2_CONFIG, dex::service::auth::get_jwks};
use crate::common::utils::auth::AuthExtractor;
@@ -44,7 +42,7 @@ pub async fn token_validator(
let path = match req
.request()
.path()
.strip_prefix(format!("{}/api/", CONFIG.common.base_uri).as_str())
.strip_prefix(format!("{}/api/", config::get_config().common.base_uri).as_str())
{
Some(path) => path,
None => req.request().path(),
src/handler/http/auth/validator.rs
+9 -6
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -20,7 +20,7 @@ use actix_web::{
web, Error,
};
use actix_web_httpauth::extractors::basic::BasicAuth;
use config::{utils::base64, CONFIG};
use config::{get_config, utils::base64};
use crate::{
common::{
@@ -45,10 +45,11 @@ pub async fn validator(
auth_info: AuthExtractor,
path_prefix: &str,
) -> Result<ServiceRequest, (Error, ServiceRequest)> {
let cfg = get_config();
let path = match req
.request()
.path()
.strip_prefix(format!("{}{}", CONFIG.common.base_uri, path_prefix).as_str())
.strip_prefix(format!("{}{}", cfg.common.base_uri, path_prefix).as_str())
{
Some(path) => path,
None => req.request().path(),
@@ -244,10 +245,11 @@ pub async fn validator_aws(
req: ServiceRequest,
_credentials: Option<BasicAuth>,
) -> Result<ServiceRequest, (Error, ServiceRequest)> {
let cfg = get_config();
let path = req
.request()
.path()
.strip_prefix(format!("{}/aws/", CONFIG.common.base_uri).as_str())
.strip_prefix(format!("{}/aws/", cfg.common.base_uri).as_str())
.unwrap_or(req.request().path());
match req.headers().get("X-Amz-Firehose-Access-Key") {
@@ -288,10 +290,11 @@ pub async fn validator_gcp(
req: ServiceRequest,
_credentials: Option<BasicAuth>,
) -> Result<ServiceRequest, (Error, ServiceRequest)> {
let cfg = get_config();
let path = req
.request()
.path()
.strip_prefix(format!("{}/gcp/", CONFIG.common.base_uri).as_str())
.strip_prefix(format!("{}/gcp/", cfg.common.base_uri).as_str())
.unwrap_or(req.request().path());
let query =
@@ -332,7 +335,7 @@ pub async fn validator_rum(
let path = req
.request()
.path()
.strip_prefix(format!("{}/rum/v1/", CONFIG.common.base_uri).as_str())
.strip_prefix(format!("{}/rum/v1/", get_config().common.base_uri).as_str())
.unwrap_or(req.request().path());
// After this previous path clean we should get only the
src/handler/http/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/authz/fga.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/authz/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/enrichment_table/mod.rs
+5 -4
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
@@ -17,7 +17,7 @@ use std::io::Error;
use actix_multipart::Multipart;
use actix_web::{post, web, HttpRequest, HttpResponse};
use config::{CONFIG, SIZE_IN_MB};
use config::SIZE_IN_MB;
use hashbrown::HashMap;
use crate::{
@@ -62,10 +62,11 @@ pub async fn save_enrichment_table(
/ SIZE_IN_MB
}
};
if content_length > CONFIG.limit.enrichment_table_limit as f64 {
let cfg = config::get_config();
if content_length > cfg.limit.enrichment_table_limit as f64 {
return Ok(MetaHttpResponse::bad_request(format!(
"exceeds allowed limit of {} mb",
CONFIG.limit.enrichment_table_limit
cfg.limit.enrichment_table_limit
)));
}
match content_type {
src/handler/http/request/functions/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/kv/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/logs/ingest.rs
+2 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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,7 +16,6 @@
use std::io::Error;
use actix_web::{http, post, web, HttpRequest, HttpResponse};
use config::CONFIG;
use crate::{
common::meta::{
@@ -288,7 +287,7 @@ pub async fn otlp_logs_write(
let user_email = req.headers().get("user_id").unwrap().to_str().unwrap();
let in_stream_name = req
.headers()
.get(&CONFIG.grpc.stream_header_key)
.get(&config::get_config().grpc.stream_header_key)
.map(|header| header.to_str().unwrap());
if content_type.eq(CONTENT_TYPE_PROTO) {
// log::info!("otlp::logs_proto_handler");
src/handler/http/request/logs/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/metrics/ingest.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/metrics/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/organization/es.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/organization/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/organization/settings.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/prom/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/rum/ingest.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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
src/handler/http/request/rum/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2023 Zinc Labs Inc.
// Copyright 2024 Zinc Labs 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

Some files were not shown because too many files have changed in this diff Show More