mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 16:09:05 +00:00
fix: need delete merged files from ingester cache db (#2677)
This commit is contained in:
Hengfei Yang
Generated
+59
-29
@@ -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",
|
||||
|
||||
+4
-1
@@ -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"
|
||||
|
||||
@@ -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<Str
|
||||
if nodes.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let hash = config::utils::hash::default::new().sum64(key);
|
||||
let hash = config::utils::hash::gxhash::new().sum64(key);
|
||||
let mut iter = nodes.lower_bound(Bound::Included(&hash));
|
||||
loop {
|
||||
if let Some(uuid) = iter.value() {
|
||||
@@ -533,11 +533,50 @@ mod tests {
|
||||
add_node_to_consistent_hash(&node_c, &Role::Compactor).await;
|
||||
}
|
||||
|
||||
for key in vec![
|
||||
["test", "node-q-8", "node-c-6"],
|
||||
["test1", "node-q-3", "node-c-9"],
|
||||
["test2", "node-q-3", "node-c-0"],
|
||||
] {
|
||||
for key in vec!["test", "test1", "test2", "test3"] {
|
||||
println!(
|
||||
"{key}-q: {}",
|
||||
get_node_from_consistent_hash(key, &Role::Querier)
|
||||
.await
|
||||
.unwrap()
|
||||
);
|
||||
println!(
|
||||
"{key}-c: {}",
|
||||
get_node_from_consistent_hash(key, &Role::Compactor)
|
||||
.await
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
// fnv hash
|
||||
let _data = vec![
|
||||
["test", "node-q-8", "node-c-8"],
|
||||
["test1", "node-q-8", "node-c-8"],
|
||||
["test2", "node-q-8", "node-c-8"],
|
||||
["test3", "node-q-8", "node-c-8"],
|
||||
];
|
||||
// murmur3 hash
|
||||
let _data = vec![
|
||||
["test", "node-q-2", "node-c-3"],
|
||||
["test1", "node-q-5", "node-c-6"],
|
||||
["test2", "node-q-4", "node-c-2"],
|
||||
["test3", "node-q-0", "node-c-3"],
|
||||
];
|
||||
// cityhash hash
|
||||
let _data = vec![
|
||||
["test", "node-q-6", "node-c-7"],
|
||||
["test1", "node-q-5", "node-c-2"],
|
||||
["test2", "node-q-2", "node-c-4"],
|
||||
["test3", "node-q-2", "node-c-1"],
|
||||
];
|
||||
// gxhash hash
|
||||
let data = vec![
|
||||
["test", "node-q-8", "node-c-0"],
|
||||
["test1", "node-q-9", "node-c-1"],
|
||||
["test2", "node-q-9", "node-c-8"],
|
||||
["test3", "node-q-3", "node-c-7"],
|
||||
];
|
||||
for key in data {
|
||||
assert_eq!(
|
||||
get_node_from_consistent_hash(key.get(0).unwrap(), &Role::Querier).await,
|
||||
Some(key.get(1).unwrap().to_string())
|
||||
|
||||
@@ -19,7 +19,7 @@ use arrow_schema::Field;
|
||||
use config::{
|
||||
meta::stream::{PartitionTimeLevel, StreamStats, StreamType},
|
||||
utils::{
|
||||
hash::{fnv, Sum64},
|
||||
hash::{gxhash, Sum64},
|
||||
json,
|
||||
},
|
||||
};
|
||||
@@ -195,7 +195,7 @@ impl StreamPartition {
|
||||
match &self.types {
|
||||
StreamPartitionType::Value => 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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -13,18 +13,17 @@
|
||||
// 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::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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -13,8 +13,10 @@
|
||||
// You should have received a copy of the GNU Affero General Public License
|
||||
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
pub mod default;
|
||||
pub mod cityhash;
|
||||
pub mod fnv;
|
||||
pub mod gxhash;
|
||||
pub mod murmur3;
|
||||
|
||||
pub trait Sum64 {
|
||||
fn sum64(&mut self, key: &str) -> u64;
|
||||
|
||||
@@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -14,7 +14,7 @@
|
||||
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
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::<Vec<_>>();
|
||||
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
|
||||
|
||||
@@ -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()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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<i64, anyhow::Error> {
|
||||
async fn update_stats_lock_node() -> Result<Option<i64>, 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<i64, anyhow::Error> {
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String>) -> 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<String>) -> 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();
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
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,
|
||||
|
||||
Reference in New Issue
Block a user