diff --git a/Cargo.lock b/Cargo.lock index 47b9034ebf..eaff985674 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1874,6 +1874,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "cityhasher" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ceab37c9e94f42414cccae77e930232c517f1bb190947018cffb0ab41fc40992" + [[package]] name = "clap" version = "3.2.25" @@ -1886,31 +1892,31 @@ dependencies = [ "clap_lex 0.2.4", "indexmap 1.9.3", "once_cell", - "strsim", + "strsim 0.10.0", "termcolor", "textwrap", ] [[package]] name = "clap" -version = "4.4.18" +version = "4.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e578d6ec4194633722ccf9544794b71b1385c3c027efe0c55db226fc880865c" +checksum = "80c21025abd42669a92efc996ef13cfb2c5c627858421ea58d5c3b331a6c134f" dependencies = [ "clap_builder", - "clap_derive 4.4.7", + "clap_derive 4.5.0", ] [[package]] name = "clap_builder" -version = "4.4.18" +version = "4.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4df4df40ec50c46000231c914968278b1eb05098cf8f1b3a518a95030e71d1c7" +checksum = "458bf1f341769dfcf849846f65dffdf9146daa56bcd2a47cb4e1de9915567c99" dependencies = [ "anstream", "anstyle", - "clap_lex 0.6.0", - "strsim", + "clap_lex 0.7.0", + "strsim 0.11.0", ] [[package]] @@ -1928,9 +1934,9 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.4.7" +version = "4.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cf9804afaaf59a91e75b022a30fb7229a7901f60c755489cc61c9b423b836442" +checksum = "307bc0538d5f0f83b8248db3087aa92fe504e4691294d0c96c0eabc33f47ba47" dependencies = [ "heck", "proc-macro2", @@ -1949,9 +1955,9 @@ dependencies = [ [[package]] name = "clap_lex" -version = "0.6.0" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "702fc72eb24e5a1e48ce58027a675bc24edd52096d5397d4aea7c6dd9eca0bd1" +checksum = "98cc8fbded0c607b7ba9dd60cd98df59af97e84d24e49c8557331cfc26d301ce" [[package]] name = "clipboard-win" @@ -2055,17 +2061,20 @@ dependencies = [ "byteorder", "bytes", "chrono", + "cityhasher", "dashmap", "dotenv_config", "dotenvy", "get_if_addrs", "getrandom", + "gxhash", "hashbrown 0.14.3", "hex", "indexmap 2.1.0", "itertools 0.12.1", "log", "memchr", + "murmur3", "once_cell", "parking_lot 0.12.1", "parquet", @@ -2210,7 +2219,7 @@ dependencies = [ "anes", "cast", "ciborium", - "clap 4.4.18", + "clap 4.5.0", "criterion-plot", "is-terminal", "itertools 0.10.5", @@ -2361,7 +2370,7 @@ dependencies = [ "ident_case", "proc-macro2", "quote", - "strsim", + "strsim 0.10.0", "syn 2.0.48", ] @@ -3283,6 +3292,16 @@ dependencies = [ "onig", ] +[[package]] +name = "gxhash" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fcc9a192659d9fd88d8bd8b8ccdec491225e3623083c1251a1a406c47934415c" +dependencies = [ + "rand", + "rustc_version", +] + [[package]] name = "h2" version = "0.3.24" @@ -3720,12 +3739,12 @@ dependencies = [ [[package]] name = "is-terminal" -version = "0.4.10" +version = "0.4.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0bad00257d07be169d870ab665980b06cdb366d792ad690bf2e76876dc503455" +checksum = "fe8f25ce1159c7740ff0b9b2f5cdf4a8428742ba7c112b9f20f22cd5219c7dab" dependencies = [ "hermit-abi 0.3.5", - "rustix 0.38.31", + "libc", "windows-sys 0.52.0", ] @@ -3786,9 +3805,9 @@ checksum = "8eaf4bc02d17cbdd7ff4c7438cafcdf7fb9a4613313ad11b4f8fefe7d3fa0130" [[package]] name = "jobserver" -version = "0.1.27" +version = "0.1.28" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8c37f63953c4c63420ed5fd3d6d398c719489b9f872b9fa683262f8edd363c7d" +checksum = "ab46a6e9526ddef3ae7f787c06f0f2600639ba80ea3eade3d8e670a2230f51d6" dependencies = [ "libc", ] @@ -4288,6 +4307,12 @@ version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a" +[[package]] +name = "murmur3" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9252111cf132ba0929b6f8e030cac2a24b507f3a4d6db6fb2896f27b354c714b" + [[package]] name = "names" version = "0.14.0" @@ -4430,19 +4455,18 @@ checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9" [[package]] name = "num-integer" -version = "0.1.45" +version = "0.1.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "225d3389fb3509a24c93f5c29eb6bde2586b98d9f016636dff58d7c6f7569cd9" +checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f" dependencies = [ - "autocfg", "num-traits", ] [[package]] name = "num-iter" -version = "0.1.43" +version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7d03e6c028c5dc5cac6e2dec0efda81fc887605bb3d884578bb6d6bf7514e252" +checksum = "d869c01cc0c455284163fd0092f1f93835385ccab5a98a0dcc497b2f8bf055a9" dependencies = [ "autocfg", "num-integer", @@ -4463,9 +4487,9 @@ dependencies = [ [[package]] name = "num-traits" -version = "0.2.17" +version = "0.2.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "39e3200413f237f41ab11ad6d161bc7239c84dcb631773ccd7de3dfe4b5c267c" +checksum = "da0df0e5185db44f69b44f26786fe401b6c293d1907744beaa7fa62b2e5a517a" dependencies = [ "autocfg", "libm", @@ -4601,7 +4625,7 @@ checksum = "624a8340c38c1b80fd549087862da4ba43e08858af025b236e509b6649fc13d5" [[package]] name = "openobserve" -version = "0.8.0" +version = "0.8.1" dependencies = [ "actix-cors", "actix-multipart", @@ -4623,7 +4647,7 @@ dependencies = [ "blake3", "bytes", "chrono", - "clap 4.4.18", + "clap 4.5.0", "cloudevents-sdk", "config", "csv", @@ -6875,6 +6899,12 @@ version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623" +[[package]] +name = "strsim" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ee073c9e4cd00e28217186dbe12796d692868f432bf2e97ee73bed0c56dfa01" + [[package]] name = "strum" version = "0.25.0" @@ -7822,7 +7852,7 @@ dependencies = [ "chrono", "chrono-tz", "cidr-utils", - "clap 4.4.18", + "clap 4.5.0", "codespan-reporting", "community-id", "crypto_secretbox", diff --git a/Cargo.toml b/Cargo.toml index 50520c7f67..2e517702d1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,7 @@ keywords = [ license = "AGPL-3.0-only" name = "openobserve" repository = "https://github.com/openobserve/openobserve/" -version = "0.8.0" +version = "0.8.1" publish = false # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -190,6 +190,7 @@ base64 = "0.21" bytes = "1.4" byteorder = "1.4.3" chrono = { version = "0.4", default-features = false, features = ["clock"] } +cityhasher = { version = "0.1", default-features = false } dashmap = { version = "5.5", features = ["serde"] } datafusion = "35" datafusion-expr = "35" @@ -207,6 +208,7 @@ flate2 = { version = "1.0", features = ["zlib"] } futures = "0.3" get_if_addrs = "0.5" getrandom = "0.2.11" +gxhash = "~3.0.0" hashlink = "0.8.4" hashbrown = { version = "0.14.3", features = ["serde"] } hex = "0.4" @@ -215,6 +217,7 @@ ipnetwork = "0.20" itertools = "0.12" log = "0.4" memchr = "2.5" +murmur3 = "0.5" once_cell = "1.17" parking_lot = "0.12" prometheus = "0.13" diff --git a/src/common/infra/cluster.rs b/src/common/infra/cluster.rs index 4db141c1d5..9647bc528d 100644 --- a/src/common/infra/cluster.rs +++ b/src/common/infra/cluster.rs @@ -42,7 +42,7 @@ pub async fn add_node_to_consistent_hash(node: &Node, role: &Role) { Role::Compactor => COMPACTOR_CONSISTENT_HASH.write().await, _ => return, }; - let mut h = config::utils::hash::default::new(); + let mut h = config::utils::hash::gxhash::new(); for i in 0..CONSISTENT_HASH_VNODES { let key = format!("{}{}", node.uuid, i); let hash = h.sum64(&key); @@ -59,7 +59,7 @@ pub async fn remove_node_from_consistent_hash(node: &Node, role: &Role) { Role::Compactor => COMPACTOR_CONSISTENT_HASH.write().await, _ => return, }; - let mut h = config::utils::hash::default::new(); + let mut h = config::utils::hash::gxhash::new(); for i in 0..CONSISTENT_HASH_VNODES { let key = format!("{}{}", node.uuid, i); let hash = h.sum64(&key); @@ -79,7 +79,7 @@ pub async fn get_node_from_consistent_hash(key: &str, role: &Role) -> Option value.to_string(), StreamPartitionType::Hash(n) => { - let h = fnv::new().sum64(value); + let h = gxhash::new().sum64(value); let bucket = h % n; bucket.to_string() } @@ -325,9 +325,19 @@ mod tests { json::to_string(&part).unwrap(), r#"{"field":"field","types":{"hash":32},"disabled":false}"# ); - assert_eq!(part.get_partition_key("hello"), "field=11"); - assert_eq!(part.get_partition_key("world"), "field=19"); - assert_eq!(part.get_partition_key("foo"), "field=23"); - assert_eq!(part.get_partition_key("bar"), "field=26"); + + for key in &[ + "hello", "world", "foo", "bar", "test", "test1", "test2", "test3", + ] { + println!("{}: {}", key, part.get_partition_key(key)); + } + assert_eq!(part.get_partition_key("hello"), "field=8"); + assert_eq!(part.get_partition_key("world"), "field=14"); + assert_eq!(part.get_partition_key("foo"), "field=3"); + assert_eq!(part.get_partition_key("bar"), "field=23"); + assert_eq!(part.get_partition_key("test"), "field=18"); + assert_eq!(part.get_partition_key("test1"), "field=3"); + assert_eq!(part.get_partition_key("test2"), "field=15"); + assert_eq!(part.get_partition_key("test3"), "field=2"); } } diff --git a/src/config/Cargo.toml b/src/config/Cargo.toml index 9d6251a0bc..043c5386ab 100644 --- a/src/config/Cargo.toml +++ b/src/config/Cargo.toml @@ -16,16 +16,19 @@ base64.workspace = true bytes.workspace = true byteorder.workspace = true chrono.workspace = true +cityhasher.workspace = true dashmap.workspace = true dotenv_config.workspace = true dotenvy.workspace = true get_if_addrs.workspace = true +gxhash.workspace = true hashbrown.workspace = true hex.workspace = true indexmap.workspace = true itertools.workspace = true log.workspace = true memchr.workspace = true +murmur3.workspace = true once_cell.workspace = true parking_lot.workspace = true parquet.workspace = true diff --git a/src/config/src/utils/hash/default.rs b/src/config/src/utils/hash/cityhash.rs similarity index 51% rename from src/config/src/utils/hash/default.rs rename to src/config/src/utils/hash/cityhash.rs index ee89e18968..f41ca9ffdc 100644 --- a/src/config/src/utils/hash/default.rs +++ b/src/config/src/utils/hash/cityhash.rs @@ -13,18 +13,17 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::hash::{DefaultHasher, Hasher}; - use super::Sum64; -pub fn new() -> DefaultHasher { - DefaultHasher::new() +pub struct CityHash {} + +pub fn new() -> CityHash { + CityHash {} } -impl Sum64 for DefaultHasher { +impl Sum64 for CityHash { fn sum64(&mut self, key: &str) -> u64 { - self.write(key.as_bytes()); - self.finish() + cityhasher::hash(key) } } @@ -33,14 +32,21 @@ mod tests { use super::*; #[test] - fn test_default_hasher_sum64() { + fn test_cityhash_sum64() { let mut h = new(); - assert_eq!(h.sum64("hello"), 16350172494705860510); - assert_eq!(h.sum64("world"), 1348462810646499051); - assert_eq!(h.sum64("foo"), 13658606619662280602); - assert_eq!(h.sum64("bar"), 11254443135816450233); - assert_eq!(h.sum64("test"), 18199023029103496422); - assert_eq!(h.sum64("test1"), 5073490808167294919); - assert_eq!(h.sum64("test2"), 14030261930256505646); + for key in &[ + "hello", "world", "foo", "bar", "test", "test1", "test2", "test3", + ] { + let sum = h.sum64(key); + println!("{}: {}", key, sum); + } + assert_eq!(h.sum64("hello"), 13009744463427800296); + assert_eq!(h.sum64("world"), 16436542438370751598); + assert_eq!(h.sum64("foo"), 6150913649986995171); + assert_eq!(h.sum64("bar"), 18347965944623823639); + assert_eq!(h.sum64("test"), 8581389452482819506); + assert_eq!(h.sum64("test1"), 10258151040114700067); + assert_eq!(h.sum64("test2"), 13211586869705959535); + assert_eq!(h.sum64("test3"), 3708989183841402594); } } diff --git a/src/config/src/utils/hash/fnv.rs b/src/config/src/utils/hash/fnv.rs index 829ffdf387..6e8ee31dd2 100644 --- a/src/config/src/utils/hash/fnv.rs +++ b/src/config/src/utils/hash/fnv.rs @@ -52,6 +52,12 @@ mod tests { #[test] fn test_fnv64a() { let mut h = new(); + for key in &[ + "hello", "world", "foo", "bar", "test", "test1", "test2", "test3", + ] { + let sum = h.sum64(key); + println!("{}: {}", key, sum); + } assert_eq!(h.sum64("hello"), 11831194018420276491); assert_eq!(h.sum64("world"), 5717881983045765875); assert_eq!(h.sum64("foo"), 15902901984413996407); @@ -59,5 +65,6 @@ mod tests { assert_eq!(h.sum64("test"), 18007334074686647077); assert_eq!(h.sum64("test1"), 2271358237066212092); assert_eq!(h.sum64("test2"), 2271361535601096725); + assert_eq!(h.sum64("test3"), 2271360436089468514); } } diff --git a/src/config/src/utils/hash/gxhash.rs b/src/config/src/utils/hash/gxhash.rs new file mode 100644 index 0000000000..588c89dddf --- /dev/null +++ b/src/config/src/utils/hash/gxhash.rs @@ -0,0 +1,52 @@ +// Copyright 2023 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 +// 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 . + +use super::Sum64; + +pub struct GxHash {} + +pub fn new() -> GxHash { + GxHash {} +} + +impl Sum64 for GxHash { + fn sum64(&mut self, key: &str) -> u64 { + gxhash::gxhash64(key.as_bytes(), 0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_gxhash_sum64() { + let mut h = new(); + for key in &[ + "hello", "world", "foo", "bar", "test", "test1", "test2", "test3", + ] { + let sum = h.sum64(key); + println!("{}: {}", key, sum); + } + assert_eq!(h.sum64("hello"), 17199510979973968020); + assert_eq!(h.sum64("world"), 16083628800851799373); + assert_eq!(h.sum64("foo"), 12351526755496507957); + assert_eq!(h.sum64("bar"), 2492955946775841796); + assert_eq!(h.sum64("test"), 687545438460047850); + assert_eq!(h.sum64("test1"), 13065486829486102133); + assert_eq!(h.sum64("test2"), 16870625056057693394); + assert_eq!(h.sum64("test3"), 14341735574462002086); + } +} diff --git a/src/config/src/utils/hash/mod.rs b/src/config/src/utils/hash/mod.rs index 4dc2225c87..63517f7d18 100644 --- a/src/config/src/utils/hash/mod.rs +++ b/src/config/src/utils/hash/mod.rs @@ -13,8 +13,10 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -pub mod default; +pub mod cityhash; pub mod fnv; +pub mod gxhash; +pub mod murmur3; pub trait Sum64 { fn sum64(&mut self, key: &str) -> u64; diff --git a/src/config/src/utils/hash/murmur3.rs b/src/config/src/utils/hash/murmur3.rs new file mode 100644 index 0000000000..00f8f529e2 --- /dev/null +++ b/src/config/src/utils/hash/murmur3.rs @@ -0,0 +1,58 @@ +// Copyright 2023 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 +// 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 . + +use std::io::Cursor; + +use murmur3::murmur3_x64_128; + +use super::Sum64; + +pub struct Mr3 {} + +pub fn new() -> Mr3 { + Mr3 {} +} + +impl Sum64 for Mr3 { + fn sum64(&mut self, key: &str) -> u64 { + let mut r = Cursor::new(key); + let ret = murmur3_x64_128(&mut r, 0).unwrap(); + ret as u64 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_murmur3_sum64() { + let mut h = new(); + for key in &[ + "hello", "world", "foo", "bar", "test", "test1", "test2", "test3", + ] { + let sum = h.sum64(key); + println!("{}: {}", key, sum); + } + assert_eq!(h.sum64("hello"), 14688674573012802306); + assert_eq!(h.sum64("world"), 8198091784597505258); + assert_eq!(h.sum64("foo"), 16316970633193145697); + assert_eq!(h.sum64("bar"), 10535706080149431812); + assert_eq!(h.sum64("test"), 12429135405209477533); + assert_eq!(h.sum64("test1"), 271003567416724429); + assert_eq!(h.sum64("test2"), 17808217402673344406); + assert_eq!(h.sum64("test3"), 11977005607778562912); + } +} diff --git a/src/handler/grpc/request/event.rs b/src/handler/grpc/request/event.rs index 425000af86..efc9b6be44 100644 --- a/src/handler/grpc/request/event.rs +++ b/src/handler/grpc/request/event.rs @@ -14,7 +14,7 @@ // along with this program. If not, see . use config::{ - cluster::{is_querier, LOCAL_NODE_ROLE, LOCAL_NODE_UUID}, + cluster::{is_compactor, is_querier, LOCAL_NODE_ROLE, LOCAL_NODE_UUID}, meta::{cluster::Role, stream::FileKey}, metrics, CONFIG, }; @@ -55,16 +55,20 @@ impl Event for Eventer { .filter(|v| v.deleted) .map(|v| v.key.clone()) .collect::>(); - if let Err(e) = infra_file_list::batch_add(&put_items).await { - // metrics - let time = start.elapsed().as_secs_f64(); - metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/event/send_file_list", "500", "", "", ""]) - .observe(time); - metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/event/send_file_list", "500", "", "", ""]) - .inc(); - return Err(Status::internal(e.to_string())); + // querier and compactor can accept add new files + // ingester only accept remove old files + if is_querier(&LOCAL_NODE_ROLE) || is_compactor(&LOCAL_NODE_ROLE) { + if let Err(e) = infra_file_list::batch_add(&put_items).await { + // metrics + let time = start.elapsed().as_secs_f64(); + metrics::GRPC_RESPONSE_TIME + .with_label_values(&["/event/send_file_list", "500", "", "", ""]) + .observe(time); + metrics::GRPC_INCOMING_REQUESTS + .with_label_values(&["/event/send_file_list", "500", "", "", ""]) + .inc(); + return Err(Status::internal(e.to_string())); + } } if let Err(e) = infra_file_list::batch_remove(&del_items).await { // metrics @@ -78,7 +82,7 @@ impl Event for Eventer { return Err(Status::internal(e.to_string())); } - // cache latest files + // cache latest files for querier if CONFIG.memory_cache.cache_latest_files && is_querier(&LOCAL_NODE_ROLE) { for item in put_items { let Some(node) = get_node_from_consistent_hash(&item.key, &Role::Querier).await diff --git a/src/infra/src/errors/mod.rs b/src/infra/src/errors/mod.rs index 59bbf0e0a8..14c4b29958 100644 --- a/src/infra/src/errors/mod.rs +++ b/src/infra/src/errors/mod.rs @@ -222,15 +222,5 @@ mod tests { "DbError# key /another/shrubbery does not exist", &err.to_string() ); - assert_eq!( - r#" - DbError( - KeyNotExists( - "/another/shrubbery", - ), - ) - "#, - &err.to_string() - ); } } diff --git a/src/service/compact/file_list.rs b/src/service/compact/file_list.rs index f3883e86f7..e47743728f 100644 --- a/src/service/compact/file_list.rs +++ b/src/service/compact/file_list.rs @@ -197,7 +197,7 @@ async fn merge_file_list(offset: i64) -> Result<(), anyhow::Error> { let locker = dist_lock::lock(&lock_key, CONFIG.etcd.command_timeout).await?; let node = db::compact::file_list::get_process(offset).await; if !node.is_empty() && LOCAL_NODE_UUID.ne(&node) && get_node_by_uuid(&node).is_some() { - log::error!("[COMPACT] list_list offset [{offset}] is merging by {node}"); + log::debug!("[COMPACT] list_list offset [{offset}] is processing by {node}"); dist_lock::unlock(&locker).await?; return Ok(()); // not this node, just skip } diff --git a/src/service/compact/stats.rs b/src/service/compact/stats.rs index 6750fff43b..8f281398d4 100644 --- a/src/service/compact/stats.rs +++ b/src/service/compact/stats.rs @@ -23,13 +23,17 @@ pub async fn update_stats_from_file_list() -> Result<(), anyhow::Error> { // get last offset let (mut offset, node) = db::compact::stats::get_offset().await; if !node.is_empty() && LOCAL_NODE_UUID.ne(&node) && get_node_by_uuid(&node).is_some() { - log::error!("[COMPACT] update stats from file_list is merging by {node}"); + log::debug!("[COMPACT] update stats from file_list is processing by {node}"); return Ok(()); } // before starting, set current node to lock the job if node.is_empty() || LOCAL_NODE_UUID.ne(&node) { - offset = update_stats_lock_node().await?; + offset = match update_stats_lock_node().await { + Ok(Some(offset)) => offset, + Ok(None) => return Ok(()), + Err(e) => return Err(e), + } } // get latest offset @@ -56,7 +60,7 @@ pub async fn update_stats_from_file_list() -> Result<(), anyhow::Error> { Ok(()) } -async fn update_stats_lock_node() -> Result { +async fn update_stats_lock_node() -> Result, anyhow::Error> { let lock_key = "compact/stream_stats/offset".to_string(); let locker = dist_lock::lock(&lock_key, CONFIG.etcd.command_timeout).await?; // check the working node for the organization again, maybe other node locked it @@ -64,14 +68,17 @@ async fn update_stats_lock_node() -> Result { let (offset, node) = db::compact::stats::get_offset().await; if !node.is_empty() && LOCAL_NODE_UUID.ne(&node) && get_node_by_uuid(&node).is_some() { dist_lock::unlock(&locker).await?; - return Err(anyhow::anyhow!( - "[COMPACT] update stats from file_list is merging by {node}" - )); + log::debug!("[COMPACT] update stats from file_list is processing by {node}"); + return Ok(None); } // bind the job to this node let ret = db::compact::stats::set_offset(offset, Some(&LOCAL_NODE_UUID.clone())).await; // already bind to this node, we can unlock now dist_lock::unlock(&locker).await?; - if let Err(e) = ret { Err(e) } else { Ok(offset) } + if let Err(e) = ret { + Err(e) + } else { + Ok(Some(offset)) + } } diff --git a/src/service/db/file_list/broadcast.rs b/src/service/db/file_list/broadcast.rs index 8e696f266b..81d9ad9ab9 100644 --- a/src/service/db/file_list/broadcast.rs +++ b/src/service/db/file_list/broadcast.rs @@ -16,7 +16,7 @@ use std::sync::Arc; use config::{ - cluster::{is_compactor, is_querier, LOCAL_NODE_UUID}, + cluster::{is_compactor, is_ingester, is_querier, LOCAL_NODE_UUID}, meta::{ cluster::{Node, NodeStatus}, stream::FileKey, @@ -48,7 +48,7 @@ pub async fn send(items: &[FileKey], node_uuid: Option) -> Result<(), an cluster::get_cached_nodes(|node| { node.scheduled && (node.status == NodeStatus::Prepare || node.status == NodeStatus::Online) - && (is_querier(&node.role) || is_compactor(&node.role)) + && (is_querier(&node.role) || is_compactor(&node.role) || is_ingester(&node.role)) }) .unwrap() }; @@ -58,7 +58,7 @@ pub async fn send(items: &[FileKey], node_uuid: Option) -> Result<(), an if node.uuid.eq(&local_node_uuid) { continue; } - if !is_querier(&node.role) && !is_compactor(&node.role) { + if !is_querier(&node.role) && !is_compactor(&node.role) && !is_ingester(&node.role) { continue; } let node_uuid = node.uuid.clone(); diff --git a/src/wal/src/writer.rs b/src/wal/src/writer.rs index a5a5fbeac9..aa7ce88aa7 100644 --- a/src/wal/src/writer.rs +++ b/src/wal/src/writer.rs @@ -14,7 +14,7 @@ // along with this program. If not, see . use std::{ - fs::{create_dir_all, File, OpenOptions}, + fs::{create_dir_all, remove_file, File, OpenOptions}, io::{self, Seek, SeekFrom, Write}, path::PathBuf, }; @@ -56,11 +56,16 @@ impl Writer { .context(FileReadSnafu { path: path.clone() })?; } - f.write_all(super::FILE_TYPE_IDENTIFIER) - .context(WriteFileTypeSnafu)?; + if let Err(e) = f.write_all(super::FILE_TYPE_IDENTIFIER) { + _ = remove_file(&path); + return Err(Error::WriteFileType { source: e }); + } let bytes_written = super::FILE_TYPE_IDENTIFIER.len(); - f.sync_all().expect("fsync failure"); + if let Err(e) = f.sync_all() { + _ = remove_file(&path); + return Err(Error::WriteFileType { source: e }); + } Ok(Self { path,