diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index 0ae082435e..5b46237472 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -5,8 +5,8 @@ RUN apt-get update && \ apt-get install -y --no-install-recommends \ protobuf-compiler \ cmake curl && \ - rustup toolchain install nightly-2025-11-11 && \ - rustup default nightly-2025-11-11 && \ + rustup toolchain install nightly-2026-05-20 && \ + rustup default nightly-2026-05-20 && \ rustup target add x86_64-unknown-linux-gnu && \ apt-get clean && \ rm -rf /var/lib/apt/lists/* diff --git a/.github/workflows/api-testing.yml b/.github/workflows/api-testing.yml index a8719771c0..3c59e0d2f5 100644 --- a/.github/workflows/api-testing.yml +++ b/.github/workflows/api-testing.yml @@ -75,7 +75,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 - name: Setup Rust Cache uses: Swatinem/rust-cache@v2 @@ -185,7 +185,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 - name: Setup Rust Cache uses: Swatinem/rust-cache@v2 @@ -275,7 +275,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 - name: Setup Rust Cache uses: Swatinem/rust-cache@v2 diff --git a/.github/workflows/build-fork-pr-image.yml b/.github/workflows/build-fork-pr-image.yml index 7753ade788..5a73d7c690 100644 --- a/.github/workflows/build-fork-pr-image.yml +++ b/.github/workflows/build-fork-pr-image.yml @@ -57,7 +57,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 - uses: Swatinem/rust-cache@v2 with: diff --git a/.github/workflows/build-pr-image.yml b/.github/workflows/build-pr-image.yml index 14505b6f18..a20f44192d 100644 --- a/.github/workflows/build-pr-image.yml +++ b/.github/workflows/build-pr-image.yml @@ -78,7 +78,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 - uses: Swatinem/rust-cache@v2 with: @@ -172,7 +172,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 - uses: Swatinem/rust-cache@v2 with: diff --git a/.github/workflows/db-testing.yml b/.github/workflows/db-testing.yml index 5ac8a80da3..d1d0e6161b 100644 --- a/.github/workflows/db-testing.yml +++ b/.github/workflows/db-testing.yml @@ -80,7 +80,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 - name: Setup Rust Cache uses: Swatinem/rust-cache@v2 diff --git a/.github/workflows/playwright.yml b/.github/workflows/playwright.yml index 0631d422b8..4e282739fd 100644 --- a/.github/workflows/playwright.yml +++ b/.github/workflows/playwright.yml @@ -96,7 +96,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 targets: x86_64-unknown-linux-gnu - uses: Swatinem/rust-cache@v2 with: diff --git a/.github/workflows/playwright_regression.yml b/.github/workflows/playwright_regression.yml index 50b44f3cb5..62a8afbc60 100644 --- a/.github/workflows/playwright_regression.yml +++ b/.github/workflows/playwright_regression.yml @@ -67,7 +67,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 targets: x86_64-unknown-linux-gnu - uses: Swatinem/rust-cache@v2 with: diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 5805e2eb3b..5612f0d591 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -4,7 +4,7 @@ on: - "v*.*.*" name: Release env: - RUST_TOOLCHAIN: nightly-2025-11-11 + RUST_TOOLCHAIN: nightly-2026-05-20 jobs: build: name: Build binary diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml index 33c1ac013b..7deefe0ead 100644 --- a/.github/workflows/unit-tests.yml +++ b/.github/workflows/unit-tests.yml @@ -74,7 +74,7 @@ jobs: - name: Setup Rust Toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2025-11-11 + toolchain: nightly-2026-05-20 targets: x86_64-unknown-linux-gnu - uses: taiki-e/install-action@v2 with: diff --git a/Cargo.lock b/Cargo.lock index 4f933c4d95..6baa6fd23e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6047,6 +6047,17 @@ dependencies = [ "sha2", ] +[[package]] +name = "mac_address" +version = "1.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0aeb26bf5e836cc1c341c8106051b573f1766dfa05aa87f0b98be5e51b02303" +dependencies = [ + "nix 0.29.0", + "serde", + "winapi", +] + [[package]] name = "macro_rules_attribute" version = "0.1.3" @@ -6144,6 +6155,15 @@ dependencies = [ "libc", ] +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "mimalloc" version = "0.1.48" @@ -6381,6 +6401,19 @@ dependencies = [ "libc", ] +[[package]] +name = "nix" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46" +dependencies = [ + "bitflags 2.11.0", + "cfg-if", + "cfg_aliases", + "libc", + "memoffset", +] + [[package]] name = "nix" version = "0.30.1" @@ -7704,8 +7737,7 @@ dependencies = [ [[package]] name = "proc-macro-error-attr2" version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96de42df36bb9bba5542fe9f1a054b8cc87e172759a1868aa05c1f3acc89dfc5" +source = "git+https://github.com/hengfeiyang/proc-macro-error-2?branch=main#3aec7aa1c9c4c9de7019f1b9f919e19380342bbf" dependencies = [ "proc-macro2", "quote", @@ -7714,8 +7746,7 @@ dependencies = [ [[package]] name = "proc-macro-error2" version = "2.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11ec05c52be0a07b08061f7dd003e7d7092e0472bc731b4af7bb1ef876109802" +source = "git+https://github.com/hengfeiyang/proc-macro-error-2?branch=main#3aec7aa1c9c4c9de7019f1b9f919e19380342bbf" dependencies = [ "proc-macro-error-attr2", "proc-macro2", @@ -9318,9 +9349,9 @@ dependencies = [ [[package]] name = "sea-orm" -version = "1.1.19" +version = "1.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d945f62558fac19e5988680d2fdf747b734c2dbc6ce2cb81ba33ed8dde5b103" +checksum = "2dc312fedd460a47ea563911761d254a84e7b51d8cc73ec92c929e78f33fa957" dependencies = [ "async-stream", "async-trait", @@ -9329,6 +9360,7 @@ dependencies = [ "derive_more 2.1.1", "futures-util", "log", + "mac_address", "ouroboros", "pgvector", "rust_decimal", @@ -9348,9 +9380,9 @@ dependencies = [ [[package]] name = "sea-orm-cli" -version = "1.1.19" +version = "1.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c94492e2ab6c045b4cc38013809ce255d14c3d352c9f0d11e6b920e2adc948ad" +checksum = "da80ebcdb44571e86f03a2bdcb5532136a87397f366f38bbce64673fc5e6a450" dependencies = [ "chrono", "clap", @@ -9367,9 +9399,9 @@ dependencies = [ [[package]] name = "sea-orm-macros" -version = "1.1.19" +version = "1.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "84c2e64a50a9cc8339f10a27577e10062c7f995488e469f2c95762c5ee847832" +checksum = "9b9a3f90e336ec74803e8eb98c61bc98754c1adfba3b4f84d946237b752b1c88" dependencies = [ "heck 0.5.0", "proc-macro2", @@ -9381,9 +9413,9 @@ dependencies = [ [[package]] name = "sea-orm-migration" -version = "1.1.19" +version = "1.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7315c0cadb7e60fb17ee2bb282aa27d01911fc2a7e5836ec1d4ac37d19250bb4" +checksum = "07c577f2959277e936c1d08109acd1e08fc36a95ef29ec028190ba82cad8f96e" dependencies = [ "async-trait", "clap", @@ -12582,7 +12614,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 81bc32739f..90976407f9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -409,13 +409,13 @@ rustls = { version = "0.23", default-features = false, features = [ ] } rustls-native-certs = "0.8" webpki-roots = "1.0" -sea-orm = { version = "1.1.19", features = [ +sea-orm = { version = "1.1.20", features = [ "sqlx-postgres", "sqlx-sqlite", "runtime-tokio-rustls", "macros", ] } -sea-orm-migration = { version = "1.1.19", features = [ +sea-orm-migration = { version = "1.1.20", features = [ "sqlx-postgres", "sqlx-sqlite", "runtime-tokio-rustls", @@ -482,3 +482,4 @@ zstd = "0.13" [patch.crates-io] object_store = { git = "https://github.com/openobserve/arrow-rs-object-store", branch = "openobserve-v0.13.2" } +proc-macro-error2 = { git = "https://github.com/hengfeiyang/proc-macro-error-2", branch = "main" } diff --git a/deploy/build/Dockerfile b/deploy/build/Dockerfile index ea5b9a5095..e6c1686d4c 100644 --- a/deploy/build/Dockerfile +++ b/deploy/build/Dockerfile @@ -8,8 +8,8 @@ RUN NODE_OPTIONS="--max-old-space-size=8192" npm run build FROM public.ecr.aws/zinclabs/rust:trixie-sccache AS builder -# RUN rustup toolchain install nightly-2025-11-11 -# RUN rustup default nightly-2025-11-11 +# RUN rustup toolchain install nightly-2026-05-20 +# RUN rustup default nightly-2026-05-20 # RUN rustup target add x86_64-unknown-linux-gnu # RUN diff -u <(rustc --print cfg) <(rustc -C target-cpu=native --print cfg) RUN rustc --version && sccache --version diff --git a/deploy/build/Dockerfile.sccache.aarch64 b/deploy/build/Dockerfile.sccache.aarch64 index ce393b1df7..d75046de43 100644 --- a/deploy/build/Dockerfile.sccache.aarch64 +++ b/deploy/build/Dockerfile.sccache.aarch64 @@ -1,7 +1,7 @@ FROM public.ecr.aws/docker/library/rust:trixie -RUN rustup toolchain install nightly-2025-11-11 -RUN rustup default nightly-2025-11-11 +RUN rustup toolchain install nightly-2026-05-20 +RUN rustup default nightly-2026-05-20 RUN rustup target add aarch64-unknown-linux-gnu RUN rustup component add rustfmt clippy llvm-tools RUN cargo install --locked sccache diff --git a/deploy/build/Dockerfile.sccache.amd64 b/deploy/build/Dockerfile.sccache.amd64 index 02b919d86f..94c3aa05fa 100644 --- a/deploy/build/Dockerfile.sccache.amd64 +++ b/deploy/build/Dockerfile.sccache.amd64 @@ -1,7 +1,7 @@ FROM public.ecr.aws/docker/library/rust:trixie -RUN rustup toolchain install nightly-2025-11-11 -RUN rustup default nightly-2025-11-11 +RUN rustup toolchain install nightly-2026-05-20 +RUN rustup default nightly-2026-05-20 RUN rustup target add x86_64-unknown-linux-gnu RUN rustup component add rustfmt clippy llvm-tools RUN cargo install --locked sccache diff --git a/deploy/build/Dockerfile.tag-debug.aarch64 b/deploy/build/Dockerfile.tag-debug.aarch64 index 90e5c497ee..63969d11c6 100644 --- a/deploy/build/Dockerfile.tag-debug.aarch64 +++ b/deploy/build/Dockerfile.tag-debug.aarch64 @@ -11,8 +11,8 @@ FROM public.ecr.aws/zinclabs/rust:trixie-sccache AS builder # RUN apt-get install -y protobuf-compiler # RUN apt-get install -y cmake curl -# RUN rustup toolchain install nightly-2025-11-11 -# RUN rustup default nightly-2025-11-11 +# RUN rustup toolchain install nightly-2026-05-20 +# RUN rustup default nightly-2026-05-20 # RUN rustup target add aarch64-unknown-linux-gnu WORKDIR /openobserve diff --git a/deploy/build/Dockerfile.tag-debug.amd64 b/deploy/build/Dockerfile.tag-debug.amd64 index 0a8a20e8ca..749508a15f 100644 --- a/deploy/build/Dockerfile.tag-debug.amd64 +++ b/deploy/build/Dockerfile.tag-debug.amd64 @@ -11,8 +11,8 @@ FROM public.ecr.aws/zinclabs/rust:trixie-sccache AS builder # RUN apt-get install -y protobuf-compiler # RUN apt-get install -y cmake curl -# RUN rustup toolchain install nightly-2025-11-11 -# RUN rustup default nightly-2025-11-11 +# RUN rustup toolchain install nightly-2026-05-20 +# RUN rustup default nightly-2026-05-20 # RUN rustup target add x86_64-unknown-linux-gnu WORKDIR /openobserve diff --git a/deploy/build/Dockerfile.tag-simd.amd64 b/deploy/build/Dockerfile.tag-simd.amd64 index 9db8db0c81..1bf00d62db 100644 --- a/deploy/build/Dockerfile.tag-simd.amd64 +++ b/deploy/build/Dockerfile.tag-simd.amd64 @@ -11,8 +11,8 @@ FROM public.ecr.aws/zinclabs/rust:trixie-sccache AS builder # RUN apt-get install -y protobuf-compiler # RUN apt-get install -y cmake curl -# RUN rustup toolchain install nightly-2025-11-11 -# RUN rustup default nightly-2025-11-11 +# RUN rustup toolchain install nightly-2026-05-20 +# RUN rustup default nightly-2026-05-20 # RUN rustup target add x86_64-unknown-linux-gnu WORKDIR /openobserve diff --git a/deploy/build/Dockerfile.tag.aarch64 b/deploy/build/Dockerfile.tag.aarch64 index 51f23cad26..b9bee1a58a 100644 --- a/deploy/build/Dockerfile.tag.aarch64 +++ b/deploy/build/Dockerfile.tag.aarch64 @@ -11,8 +11,8 @@ FROM public.ecr.aws/zinclabs/rust:trixie-sccache AS builder # RUN apt-get install -y protobuf-compiler # RUN apt-get install -y cmake curl -# RUN rustup toolchain install nightly-2025-11-11 -# RUN rustup default nightly-2025-11-11 +# RUN rustup toolchain install nightly-2026-05-20 +# RUN rustup default nightly-2026-05-20 # RUN rustup target add aarch64-unknown-linux-gnu WORKDIR /openobserve diff --git a/deploy/build/Dockerfile.tag.amd64 b/deploy/build/Dockerfile.tag.amd64 index 3498c1b83b..be76e92f43 100644 --- a/deploy/build/Dockerfile.tag.amd64 +++ b/deploy/build/Dockerfile.tag.amd64 @@ -11,8 +11,8 @@ FROM public.ecr.aws/zinclabs/rust:trixie-sccache AS builder # RUN apt-get install -y protobuf-compiler # RUN apt-get install -y cmake curl -# RUN rustup toolchain install nightly-2025-11-11 -# RUN rustup default nightly-2025-11-11 +# RUN rustup toolchain install nightly-2026-05-20 +# RUN rustup default nightly-2026-05-20 # RUN rustup target add x86_64-unknown-linux-gnu WORKDIR /openobserve diff --git a/deploy/build/buildspec-test.yml b/deploy/build/buildspec-test.yml index f9a5ad7b4b..86910d4312 100644 --- a/deploy/build/buildspec-test.yml +++ b/deploy/build/buildspec-test.yml @@ -10,9 +10,9 @@ phases: - unzip protoc-21.12-linux-x86_64.zip -d protoc && cp protoc/bin/protoc /usr/local/bin/ && cp -r protoc/include/google /usr/local/include/ - curl https://sh.rustup.rs -sSf | sh -s -- -y - . "$HOME/.cargo/env" - - rustup toolchain install nightly-2025-11-11 + - rustup toolchain install nightly-2026-05-20 - rustup toolchain list - - rustup default nightly-2025-11-11 + - rustup default nightly-2026-05-20 - rustup component add clippy llvm-tools - export RUSTFLAGS='-C target-cpu=native' - rm -rf ~/.cargo/registry diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 876a94158d..a1ce5273c7 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,3 +1,3 @@ [toolchain] -channel = "nightly-2025-11-11" +channel = "nightly-2026-05-20" components = ["rustfmt", "clippy", "llvm-tools"] diff --git a/src/common/utils/auth.rs b/src/common/utils/auth.rs index fc08dc5933..71162df00e 100644 --- a/src/common/utils/auth.rs +++ b/src/common/utils/auth.rs @@ -187,11 +187,11 @@ pub async fn set_ownership(org_id: &str, obj_type: &str, obj: Authz) { // So we can return here log::debug!( "folder tuples already exists for org: {org_id}; folder: {}", - &obj.obj_id + obj.obj_id ); return; } else if obj.parent_type.eq("folders") { - log::debug!("checking parent folder tuples for folder: {}", &obj.parent); + log::debug!("checking parent folder tuples for folder: {}", obj.parent); // In case of dashboard, we need to check if the tuples for its folder exist // If not, the below function creates the proper tuples for the folder authorizer::authz::check_folder_exists(org_id, &obj.parent).await; @@ -615,8 +615,8 @@ pub fn generate_presigned_url( ) -> String { // let time = chrono::Utc::now().timestamp(); let stage1 = get_hash(password, salt); - let stage2 = get_hash(&format!("{}{}", &stage1, time), salt); - let stage3 = get_hash(&format!("{}{}", &stage2, exp_in), salt); + let stage2 = get_hash(&format!("{}{}", stage1, time), salt); + let stage3 = get_hash(&format!("{}{}", stage2, exp_in), salt); let user_pass = format!("{username}:{stage3}"); let auth = base64::engine::general_purpose::STANDARD.encode(user_pass); diff --git a/src/common/utils/http.rs b/src/common/utils/http.rs index 256f9664ec..27070d5a8c 100644 --- a/src/common/utils/http.rs +++ b/src/common/utils/http.rs @@ -218,7 +218,7 @@ pub fn parse_ip_addr(ip_address: &str) -> Result<(IpAddr, Option), AddrPars sock_addr.ip() }) .map_err(|e| { - log::error!("Error parsing IP address: {}, {}", &ip_address, e); + log::error!("Error parsing IP address: {}, {}", ip_address, e); e }) })?; diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 37639e9cf2..456fb30db8 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -376,7 +376,7 @@ pub fn calculate_config_file_hash(path: &PathBuf) -> Result Result<(), anyhow::Error> { match crate::config_path_manager::get_config_file_path() { Some(path) => { - log::info!("Loading config from file {:?}", &path); + log::info!("Loading config from file {:?}", path); if dotenvy::from_path_override(&path).is_err() { return Err(anyhow::anyhow!("Config loading from file failed")); } diff --git a/src/config/src/meta/promql/value.rs b/src/config/src/meta/promql/value.rs index 105c0e5401..7381288efa 100644 --- a/src/config/src/meta/promql/value.rs +++ b/src/config/src/meta/promql/value.rs @@ -484,26 +484,29 @@ impl Serialize for RangeValue { where S: Serializer, { - if self.exemplars.is_none() { - let mut seq = serializer.serialize_struct("range_value", 2)?; - let labels_map = self - .labels - .iter() - .map(|l| (l.name.as_str(), l.value.as_str())) - .collect::>(); - seq.serialize_field("metric", &labels_map)?; - seq.serialize_field("values", &self.samples)?; - seq.end() - } else { - let mut seq = serializer.serialize_struct("range_value", 2)?; - let labels_map = self - .labels - .iter() - .map(|l| (l.name.as_str(), l.value.as_str())) - .collect::>(); - seq.serialize_field("seriesLabels", &labels_map)?; - seq.serialize_field("exemplars", &self.exemplars.as_ref().unwrap())?; - seq.end() + match &self.exemplars { + Some(exemplars) => { + let mut seq = serializer.serialize_struct("range_value", 2)?; + let labels_map = self + .labels + .iter() + .map(|l| (l.name.as_str(), l.value.as_str())) + .collect::>(); + seq.serialize_field("seriesLabels", &labels_map)?; + seq.serialize_field("exemplars", &exemplars)?; + seq.end() + } + None => { + let mut seq = serializer.serialize_struct("range_value", 2)?; + let labels_map = self + .labels + .iter() + .map(|l| (l.name.as_str(), l.value.as_str())) + .collect::>(); + seq.serialize_field("metric", &labels_map)?; + seq.serialize_field("values", &self.samples)?; + seq.end() + } } } } diff --git a/src/config/src/meta/stream.rs b/src/config/src/meta/stream.rs index 6210e95925..5d71d17013 100644 --- a/src/config/src/meta/stream.rs +++ b/src/config/src/meta/stream.rs @@ -915,7 +915,7 @@ impl TimeRange { return ranges; } let mut ranges = ranges; - ranges.sort_by(|a, b| a.start.cmp(&b.start)); + ranges.sort_by_key(|k| k.start); let mut result = Vec::new(); let mut current = ranges[0].clone(); for range in ranges.iter().skip(1) { diff --git a/src/config/src/utils/json.rs b/src/config/src/utils/json.rs index bfad8baed2..13201ee665 100644 --- a/src/config/src/utils/json.rs +++ b/src/config/src/utils/json.rs @@ -23,13 +23,7 @@ pub fn get_float_value(val: &Value) -> f64 { Value::String(v) => v.parse::().unwrap_or(0.0), // f64, i64, u64 both can be converted to f64 Value::Number(v) => v.as_f64().unwrap_or(0.0), - Value::Bool(v) => { - if *v { - 1.0 - } else { - 0.0 - } - } + Value::Bool(v) if *v => 1.0, _ => 0.0, } } @@ -42,13 +36,7 @@ pub fn get_int_value(val: &Value) -> i64 { Some(v) => v, None => v.as_f64().unwrap_or(0.0) as i64, }, - Value::Bool(v) => { - if *v { - 1 - } else { - 0 - } - } + Value::Bool(v) if *v => 1, _ => 0, } } @@ -61,13 +49,7 @@ pub fn get_uint_value(val: &Value) -> u64 { Some(v) => v, None => v.as_f64().unwrap_or(0.0) as u64, }, - Value::Bool(v) => { - if *v { - 1 - } else { - 0 - } - } + Value::Bool(v) if *v => 1, _ => 0, } } @@ -179,13 +161,9 @@ pub fn get_value_from_path(value: &Value, path: &str) -> Option { if key.is_empty() { continue; } - match temp.as_object() { - Some(map) => match map.get(key) { - Some(v) => temp = v, - None => return None, - }, - None => return None, - } + let map = temp.as_object()?; + let val = map.get(key)?; + temp = val; } match rest { None => Some(temp.to_owned()), diff --git a/src/config/src/utils/prom_json_encoder.rs b/src/config/src/utils/prom_json_encoder.rs index 1465143866..298da1640a 100644 --- a/src/config/src/utils/prom_json_encoder.rs +++ b/src/config/src/utils/prom_json_encoder.rs @@ -105,9 +105,7 @@ impl JsonEncoder { } // individual buckets let timestamp = crate::utils::time::now_micros(); - for (bound, value) in - upper_bounds.into_iter().zip(cumulative_counts.into_iter()) - { + for (bound, value) in upper_bounds.into_iter().zip(cumulative_counts) { let mut row = Map::new(); row.insert("_timestamp".to_string(), json!(timestamp)); row.insert("__name__".to_string(), json!(format!("{}_bucket", name))); @@ -122,8 +120,7 @@ impl JsonEncoder { let count = json!(h.get_sample_count()); let sum = json!(h.get_sample_sum()); - for (ty, val) in ["count", "sum"].into_iter().zip([count, sum].into_iter()) - { + for (ty, val) in ["count", "sum"].into_iter().zip([count, sum]) { let mut row = Map::new(); row.insert("_timestamp".to_string(), json!(timestamp)); row.insert("__name__".to_string(), json!(format!("{}_{}", name, ty))); @@ -150,7 +147,7 @@ impl JsonEncoder { // individual buckets let timestamp = crate::utils::time::now_micros(); - for (quantile, value) in quantiles.into_iter().zip(values.into_iter()) { + for (quantile, value) in quantiles.into_iter().zip(values) { let mut row = Map::new(); row.insert("_timestamp".to_string(), json!(timestamp)); row.insert("__name__".to_string(), json!(format!("{}_bucket", name))); @@ -164,7 +161,7 @@ impl JsonEncoder { let names = ["sum".to_string(), "count".to_string()]; let values = [json!(s.sample_sum()), json!(s.sample_count())]; - for (key, value) in names.into_iter().zip(values.into_iter()) { + for (key, value) in names.into_iter().zip(values) { let mut row = Map::new(); row.insert("_timestamp".to_string(), json!(timestamp)); row.insert("__name__".to_string(), json!(format!("{name}_{key}"))); diff --git a/src/config/src/utils/record_batch_ext.rs b/src/config/src/utils/record_batch_ext.rs index 02dd0ee4d6..1c78a25ee1 100644 --- a/src/config/src/utils/record_batch_ext.rs +++ b/src/config/src/utils/record_batch_ext.rs @@ -1078,13 +1078,7 @@ fn append_value_optimized( } } Value::String(s) => s.parse::().unwrap_or(0), - Value::Bool(bo) => { - if *bo { - 1 - } else { - 0 - } - } + Value::Bool(bo) if *bo => 1, _ => 0, }; b.append_value(val); @@ -1106,13 +1100,7 @@ fn append_value_optimized( } } Value::String(s) => s.parse::().unwrap_or(0), - Value::Bool(bo) => { - if *bo { - 1 - } else { - 0 - } - } + Value::Bool(bo) if *bo => 1, _ => 0, }; b.append_value(val); @@ -1126,13 +1114,7 @@ fn append_value_optimized( let val = match value { Value::Number(n) => n.as_f64().unwrap_or(0.0), Value::String(s) => s.parse::().unwrap_or(0.0), - Value::Bool(bo) => { - if *bo { - 1.0 - } else { - 0.0 - } - } + Value::Bool(bo) if *bo => 1.0, _ => 0.0, }; b.append_value(val); diff --git a/src/handler/grpc/auth/mod.rs b/src/handler/grpc/auth/mod.rs index 5354602541..7e734374e0 100644 --- a/src/handler/grpc/auth/mod.rs +++ b/src/handler/grpc/auth/mod.rs @@ -61,7 +61,7 @@ pub fn check_auth(req: Request<()>) -> Result, Status> { if org_id.is_none() { return Err(Status::invalid_argument(format!( "Please specify organization id with header key '{}' ", - &cfg.grpc.org_header_key + cfg.grpc.org_header_key ))); } diff --git a/src/handler/grpc/request/logs.rs b/src/handler/grpc/request/logs.rs index 098c254812..8d98ec9e80 100644 --- a/src/handler/grpc/request/logs.rs +++ b/src/handler/grpc/request/logs.rs @@ -34,7 +34,7 @@ impl LogsService for LogsServer { let metadata = request.metadata().clone(); let msg = format!( "Please specify organization id with header key '{}' ", - &cfg.grpc.org_header_key + cfg.grpc.org_header_key ); if !metadata.contains_key(&cfg.grpc.org_header_key) { return Err(Status::invalid_argument(msg)); diff --git a/src/handler/grpc/request/metrics/ingester.rs b/src/handler/grpc/request/metrics/ingester.rs index 5e779fded9..79367fa8c9 100644 --- a/src/handler/grpc/request/metrics/ingester.rs +++ b/src/handler/grpc/request/metrics/ingester.rs @@ -37,7 +37,7 @@ impl MetricsService for MetricsIngester { let metadata = request.metadata().clone(); let msg = format!( "Please specify organization id with header key '{}' ", - &cfg.grpc.org_header_key + cfg.grpc.org_header_key ); if !metadata.contains_key(&cfg.grpc.org_header_key) { return Err(Status::invalid_argument(msg)); diff --git a/src/handler/grpc/request/traces.rs b/src/handler/grpc/request/traces.rs index 2b7e362412..8a2f18c42e 100644 --- a/src/handler/grpc/request/traces.rs +++ b/src/handler/grpc/request/traces.rs @@ -36,7 +36,7 @@ impl TraceService for TraceServer { let metadata = request.metadata().clone(); let msg = format!( "Please specify organization id with header key '{}' ", - &cfg.grpc.org_header_key + cfg.grpc.org_header_key ); if !metadata.contains_key(&cfg.grpc.org_header_key) { return Err(Status::invalid_argument(msg)); diff --git a/src/handler/http/request/enrichment_table/mod.rs b/src/handler/http/request/enrichment_table/mod.rs index 040949ddfb..eeb523adcd 100644 --- a/src/handler/http/request/enrichment_table/mod.rs +++ b/src/handler/http/request/enrichment_table/mod.rs @@ -124,38 +124,30 @@ pub async fn save_enrichment_table( } match content_type { - Some(content_type) => { + Some(content_type) if content_type .to_str() .unwrap_or("") - .starts_with("multipart/form-data") - { - let append_data = match query.get("append") { - Some(append_data) => append_data.parse::().unwrap_or(false), - None => false, - }; + .starts_with("multipart/form-data") => + { + let append_data = match query.get("append") { + Some(append_data) => append_data.parse::().unwrap_or(false), + None => false, + }; - // Extract multipart data using axum's Multipart - match extract_multipart(payload, append_data).await { - Ok(json_record) => { - match save_enrichment_data(&org_id, &table_name, json_record, append_data) - .await - { - Ok(resp) => resp, - Err(e) => MetaHttpResponse::internal_error(e.to_string()), - } + // Extract multipart data using axum's Multipart + match extract_multipart(payload, append_data).await { + Ok(json_record) => { + match save_enrichment_data(&org_id, &table_name, json_record, append_data).await + { + Ok(resp) => resp, + Err(e) => MetaHttpResponse::internal_error(e.to_string()), } - Err(e) => MetaHttpResponse::bad_request(e), } - } else { - MetaHttpResponse::bad_request( - "Bad Request, content-type must be multipart/form-data", - ) + Err(e) => MetaHttpResponse::bad_request(e), } } - None => { - MetaHttpResponse::bad_request("Bad Request, content-type must be multipart/form-data") - } + _ => MetaHttpResponse::bad_request("Bad Request, content-type must be multipart/form-data"), } } diff --git a/src/handler/http/request/organization/ingestion_tokens.rs b/src/handler/http/request/organization/ingestion_tokens.rs index d6deff9b3d..1960958ddd 100644 --- a/src/handler/http/request/organization/ingestion_tokens.rs +++ b/src/handler/http/request/organization/ingestion_tokens.rs @@ -14,7 +14,6 @@ // along with this program. If not, see . use axum::{Json, extract::Path, response::Response}; -use config::meta::user::UserRole; use serde_json::json; use crate::{ @@ -26,10 +25,10 @@ use crate::{ OrgIngestionTokenListResponse, OrgIngestionTokenResponse, }, }, - utils::auth::{UserEmail, is_root_user}, + utils::auth::UserEmail, }, handler::http::extractors::Headers, - service::{ingestion_tokens, users}, + service::ingestion_tokens, }; /// List all org-level ingestion tokens (masked). @@ -102,10 +101,11 @@ pub async fn create_ingestion_token( #[cfg(not(feature = "enterprise"))] { - if !is_root_user(user_id) { - match users::get_user(Some(&org_id), user_id).await { + if !crate::common::utils::auth::is_root_user(user_id) { + match crate::service::users::get_user(Some(&org_id), user_id).await { Some(initiator) - if initiator.role == UserRole::Admin || initiator.role == UserRole::Root => {} + if initiator.role == config::meta::user::UserRole::Admin + || initiator.role == config::meta::user::UserRole::Root => {} _ => { return MetaHttpResponse::forbidden( "Admin or Root role required to create ingestion tokens", @@ -156,14 +156,15 @@ pub async fn enable_disable_ingestion_token( Path((org_id, name)): Path<(String, String)>, Json(body): Json, ) -> Response { - let user_id = user_email.user_id.as_str(); + let _user_id = user_email.user_id.as_str(); #[cfg(not(feature = "enterprise"))] { - if !is_root_user(user_id) { - match users::get_user(Some(&org_id), user_id).await { + if !crate::common::utils::auth::is_root_user(_user_id) { + match crate::service::users::get_user(Some(&org_id), _user_id).await { Some(initiator) - if initiator.role == UserRole::Admin || initiator.role == UserRole::Root => {} + if initiator.role == config::meta::user::UserRole::Admin + || initiator.role == config::meta::user::UserRole::Root => {} _ => { return MetaHttpResponse::forbidden( "Admin or Root role required to manage ingestion tokens", diff --git a/src/handler/http/request/search/multi_streams.rs b/src/handler/http/request/search/multi_streams.rs index 9a9a06cb72..3ee9118813 100644 --- a/src/handler/http/request/search/multi_streams.rs +++ b/src/handler/http/request/search/multi_streams.rs @@ -255,7 +255,7 @@ pub async fn search_multi( req.query.start_time = req.query.end_time - max_query_range * 3600 * 1_000_000; range_error = format!( "{} Query duration for stream {} is modified due to query range restriction of {} hours", - range_error, &stream_name, max_query_range + range_error, stream_name, max_query_range ); if multi_res.new_start_time.is_none() { diff --git a/src/handler/http/request/search/utils.rs b/src/handler/http/request/search/utils.rs index 67ad8ba43e..10f9199ee0 100644 --- a/src/handler/http/request/search/utils.rs +++ b/src/handler/http/request/search/utils.rs @@ -72,10 +72,8 @@ pub async fn check_stream_permissions( o2_model_type = LOGS_PATTERN_KEY; } } - StreamPermissionResourceType::Insights => { - if config.logs_pattern_rbac_enabled { - o2_model_type = LOGS_INSIGHTS_KEY; - } + StreamPermissionResourceType::Insights if config.logs_pattern_rbac_enabled => { + o2_model_type = LOGS_INSIGHTS_KEY; } _ => {} }; diff --git a/src/handler/http/request/short_url/mod.rs b/src/handler/http/request/short_url/mod.rs index 80a2db97e0..e1a1a353bb 100644 --- a/src/handler/http/request/short_url/mod.rs +++ b/src/handler/http/request/short_url/mod.rs @@ -140,7 +140,7 @@ pub async fn retrieve( .redirect_http() } else { let redirect = RedirectResponseBuilder::default().build(); - log::error!("Short URL not found, {}", &redirect); + log::error!("Short URL not found, {}", redirect); redirect.redirect_http() } } diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 38641959e2..bc945eb510 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -824,16 +824,16 @@ pub async fn get_latest_traces( match sort_by.as_str() { "duration" => { if sort_order == "ASC" { - traces_data.sort_by(|a, b| a.duration.cmp(&b.duration)); + traces_data.sort_by_key(|k| k.duration); } else { - traces_data.sort_by(|a, b| b.duration.cmp(&a.duration)); + traces_data.sort_by_key(|k| std::cmp::Reverse(k.duration)); } } _ => { if sort_order == "ASC" { - traces_data.sort_by(|a, b| a.start_time.cmp(&b.start_time)); + traces_data.sort_by_key(|k| k.start_time); } else { - traces_data.sort_by(|a, b| b.start_time.cmp(&a.start_time)); + traces_data.sort_by_key(|k| std::cmp::Reverse(k.start_time)); } } } @@ -1572,9 +1572,9 @@ async fn process_latest_traces_stream( let mut partition_hits: Vec<&TraceResponseItem> = traces_data.values().collect(); if sort_order == "ASC" { - partition_hits.sort_by(|a, b| a.start_time.cmp(&b.start_time)); + partition_hits.sort_by_key(|k| k.start_time); } else { - partition_hits.sort_by(|a, b| b.start_time.cmp(&a.start_time)); + partition_hits.sort_by_key(|k| std::cmp::Reverse(k.start_time)); } let deliverable: Vec = partition_hits diff --git a/src/handler/http/request/traces/session.rs b/src/handler/http/request/traces/session.rs index cae9291535..a99bd648ff 100644 --- a/src/handler/http/request/traces/session.rs +++ b/src/handler/http/request/traces/session.rs @@ -564,11 +564,10 @@ fn extract_first_user_message( let parsed: json::Value; let msgs_val: &json::Value = if val.is_array() { val - } else if let Some(s) = val.as_str() { + } else { + let s = val.as_str()?; parsed = json::from_str(s).ok()?; &parsed - } else { - return None; }; // Resolve the actual messages array — either the top-level value, or a nested @@ -577,10 +576,8 @@ fn extract_first_user_message( a } else if let Some(a) = msgs_val.get("messages").and_then(|v| v.as_array()) { a - } else if let Some(a) = msgs_val.get("contents").and_then(|v| v.as_array()) { - a } else { - return None; + msgs_val.get("contents").and_then(|v| v.as_array())? }; for msg in arr { @@ -737,7 +734,7 @@ fn aggregate_sessions( &details, )); } - sessions_data.sort_by(|a, b| b.start_time.cmp(&a.start_time)); + sessions_data.sort_by_key(|k| std::cmp::Reverse(k.start_time)); sessions_data } diff --git a/src/handler/http/request/traces/user.rs b/src/handler/http/request/traces/user.rs index 50226c4e39..71ef49d094 100644 --- a/src/handler/http/request/traces/user.rs +++ b/src/handler/http/request/traces/user.rs @@ -539,7 +539,7 @@ fn aggregate_users( &details, )); } - users_data.sort_by(|a, b| b.last_event.cmp(&a.last_event)); + users_data.sort_by_key(|k| std::cmp::Reverse(k.last_event)); users_data } diff --git a/src/handler/http/request/users/mod.rs b/src/handler/http/request/users/mod.rs index 8cc847a5cd..9a017b760e 100644 --- a/src/handler/http/request/users/mod.rs +++ b/src/handler/http/request/users/mod.rs @@ -901,10 +901,8 @@ pub async fn get_auth( }); let cookie_name = "auth_tokens"; let auth_cookie = if req_ts == 0 { - let access_token = format!( - "Basic {}", - base64::encode(&format!("{}:{}", &name, &password)) - ); + let access_token = + format!("Basic {}", base64::encode(&format!("{name}:{password}"))); let tokens = AuthTokens { access_token, refresh_token: "".to_string(), @@ -918,7 +916,7 @@ pub async fn get_auth( let auth_ext = format!( "{} {}", cookie_name, - base64::encode(&format!("{}:{}", &name, &password)) + base64::encode(&format!("{name}:{password}")) ); let tokens = AuthTokensExt { diff --git a/src/infra/src/cache/file_data/mod.rs b/src/infra/src/cache/file_data/mod.rs index a5b1ccff0a..ba25921087 100644 --- a/src/infra/src/cache/file_data/mod.rs +++ b/src/infra/src/cache/file_data/mod.rs @@ -109,7 +109,7 @@ impl CacheStrategy { return None; } let mut idx = None; - for (_, val) in map.iter() { + for val in map.values() { if !cache[*val].is_empty() { idx = Some(*val); break; diff --git a/src/infra/src/client/grpc.rs b/src/infra/src/client/grpc.rs index cbeb1eb6b1..cf705ec5f7 100644 --- a/src/infra/src/client/grpc.rs +++ b/src/infra/src/client/grpc.rs @@ -116,7 +116,7 @@ pub async fn get_cached_channel(grpc_addr: &str) -> Result Result { let cfg = config::get_config(); let mut channel = Channel::from_shared(grpc_addr.to_string()).map_err(|err| { - log::error!("gRPC node: {}, parse err: {:?}", &grpc_addr, err); + log::error!("gRPC node: {}, parse err: {:?}", grpc_addr, err); Status::internal("parse gRPC node error".to_string()) })?; if cfg.grpc.tls_enabled { @@ -135,7 +135,7 @@ pub async fn create_channel(grpc_addr: &str) -> Result { } } channel = channel.tls_config(tls).map_err(|err| { - log::error!("gRPC node: {}, tls err: {:?}", &grpc_addr, err); + log::error!("gRPC node: {}, tls err: {:?}", grpc_addr, err); Status::internal("tls gRPC node error".to_string()) })?; } @@ -146,7 +146,7 @@ pub async fn create_channel(grpc_addr: &str) -> Result { .connect() .await .map_err(|err| { - log::error!("gRPC node: {}, connect err: {:?}", &grpc_addr, err); + log::error!("gRPC node: {}, connect err: {:?}", grpc_addr, err); Status::internal("connect to gRPC node error".to_string()) })?; Ok(channel) @@ -188,7 +188,7 @@ pub async fn make_grpc_search_client( .map_err(|err| { log::error!( "[trace_id {trace_id}] search->grpc: node: {}, connect err: {:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), err ); let err = ErrorCodes::from_json(err.message()) @@ -254,7 +254,7 @@ pub async fn make_grpc_metrics_client( .map_err(|err| { log::error!( "[trace_id {trace_id}] promql->search->grpc: node: {}, connect err: {:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), err ); let err = ErrorCodes::from_json(err.message()) @@ -308,7 +308,7 @@ pub async fn make_grpc_node_client( .map_err(|err| { log::error!( "[trace_id {trace_id}] node->grpc: node: {}, connect err: {:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), err ); let err = ErrorCodes::from_json(err.message()) @@ -359,7 +359,7 @@ pub async fn make_grpc_cluster_info_client( .map_err(|err| { log::error!( "[trace_id {trace_id}] cluster_info->grpc: node: {}, connect err: {:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), err ); let err = ErrorCodes::from_json(err.message()) diff --git a/src/infra/src/cluster/mod.rs b/src/infra/src/cluster/mod.rs index a4954bde35..4922b6b0e2 100644 --- a/src/infra/src/cluster/mod.rs +++ b/src/infra/src/cluster/mod.rs @@ -162,7 +162,7 @@ pub async fn get_node_from_consistent_hash_within( } } // Wrap around to the beginning of the ring. - for (_, name) in nodes.iter() { + for name in nodes.values() { if allowed.contains(name) { return Some(name.clone()); } @@ -209,7 +209,7 @@ pub async fn get_nodes_from_consistent_hash( } } // Wrap around to the beginning of the ring. - for (_, name) in nodes.iter() { + for name in nodes.values() { result.insert(name.to_string()); if result.len() >= n { break; diff --git a/src/infra/src/coordinator/alerts.rs b/src/infra/src/coordinator/alerts.rs index 9869326814..2e2bf02150 100644 --- a/src/infra/src/coordinator/alerts.rs +++ b/src/infra/src/coordinator/alerts.rs @@ -77,7 +77,7 @@ where match ev { Event::Put(ev) => { let Some((org, alert_id)) = parse_alert_key(&ev.key) else { - log::error!("watch_alerts: failed to parse event key {}", &ev.key); + log::error!("watch_alerts: failed to parse event key {}", ev.key); continue; }; let folder_id = ev.value.map(|v| String::from_utf8_lossy(&v).to_string()); @@ -87,7 +87,7 @@ where } Event::Delete(ev) => { let Some((org, alert_id)) = parse_alert_key(&ev.key) else { - log::error!("watch_alerts: failed to parse event key {}", &ev.key); + log::error!("watch_alerts: failed to parse event key {}", ev.key); continue; }; let _ = (on_delete)(org, alert_id).await; diff --git a/src/infra/src/scheduler/postgres.rs b/src/infra/src/scheduler/postgres.rs index 40e38cd2dc..6de51ca404 100644 --- a/src/infra/src/scheduler/postgres.rs +++ b/src/infra/src/scheduler/postgres.rs @@ -180,7 +180,7 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s if trigger.module == TriggerModule::Alert && trigger.is_realtime { let key = format!( "{TRIGGERS_KEY}{}/{}/{}", - trigger.module, &trigger.org, &trigger.module_key + trigger.module, trigger.org, trigger.module_key ); let cluster_coordinator = db::get_coordinator().await; cluster_coordinator @@ -308,7 +308,7 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s if trigger.module == TriggerModule::Alert && trigger.is_realtime { let key = format!( "{TRIGGERS_KEY}{}/{}/{}", - trigger.module, &trigger.org, &trigger.module_key + trigger.module, trigger.org, trigger.module_key ); let cluster_coordinator = db::get_coordinator().await; cluster_coordinator @@ -385,7 +385,7 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s if trigger.module == TriggerModule::Alert && trigger.is_realtime { let key = format!( "{TRIGGERS_KEY}{}/{}/{}", - trigger.module, &trigger.org, &trigger.module_key + trigger.module, trigger.org, trigger.module_key ); let cluster_coordinator = db::get_coordinator().await; if let Err(e) = cluster_coordinator @@ -393,9 +393,7 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s .await { log::error!( - "Error updating cluster coordinator for trigger {}: {}", - key, - e + "Error updating cluster coordinator for trigger {key}: {e}", ); } } @@ -403,10 +401,7 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s Ok(()) } Err(e) => { - log::warn!( - "Bulk update failed, falling back to individual updates: {}", - e - ); + log::warn!("Bulk update failed, falling back to individual updates: {e}"); // Fallback to individual updates for trigger in triggers { self.update_trigger(trigger, false).await?; @@ -483,8 +478,7 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s if let Err(e) = query.execute(&pool).await { log::warn!( - "Bulk status update with data failed, falling back to individual updates: {}", - e + "Bulk status update with data failed, falling back to individual updates: {e}" ); for (org, module, module_key, status, retries, data) in updates_with_data { self.update_status(&org, module, &module_key, status, retries, data.as_deref()) diff --git a/src/infra/src/scheduler/sqlite.rs b/src/infra/src/scheduler/sqlite.rs index 4c208d9103..8304643c77 100644 --- a/src/infra/src/scheduler/sqlite.rs +++ b/src/infra/src/scheduler/sqlite.rs @@ -178,7 +178,7 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s if trigger.module == TriggerModule::Alert && trigger.is_realtime { let key = format!( "{TRIGGERS_KEY}{}/{}/{}", - trigger.module, &trigger.org, &trigger.module_key + trigger.module, trigger.org, trigger.module_key ); // TODO: For sqlite cluster coordinator, the alert triggers are put @@ -311,7 +311,7 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s if trigger.module == TriggerModule::Alert && trigger.is_realtime { let key = format!( "{TRIGGERS_KEY}{}/{}/{}", - trigger.module, &trigger.org, &trigger.module_key + trigger.module, trigger.org, trigger.module_key ); let cluster_coordinator = db::get_coordinator().await; cluster_coordinator diff --git a/src/infra/src/schema/mod.rs b/src/infra/src/schema/mod.rs index c6a5919cfe..a7e24eea73 100644 --- a/src/infra/src/schema/mod.rs +++ b/src/infra/src/schema/mod.rs @@ -799,7 +799,7 @@ impl SchemaCache { pub fn size(&self) -> usize { let mut size = std::mem::size_of::() + self.schema.size(); size += std::mem::size_of::>(); - for (key, _val) in self.fields_map.iter() { + for key in self.fields_map.keys() { size += std::mem::size_of::() + key.len(); size += std::mem::size_of::(); } diff --git a/src/infra/src/table/dashboards.rs b/src/infra/src/table/dashboards.rs index 3492578969..33c727b7c4 100644 --- a/src/infra/src/table/dashboards.rs +++ b/src/infra/src/table/dashboards.rs @@ -192,26 +192,26 @@ pub async fn put( // Get the fields that will be inserted into or updated in the database. let dashboard_id = dashboard .dashboard_id() - .and_then(|s| if s.is_empty() { None } else { Some(s) }) + .filter(|&s| !s.is_empty()) .ok_or(errors::PutDashboardError::MissingDashboardId)? .to_owned(); let owner = dashboard .owner() - .and_then(|s| if s.is_empty() { None } else { Some(s) }) + .filter(|&s| !s.is_empty()) .ok_or(errors::PutDashboardError::MissingOwner)? .to_owned(); let title = dashboard .title() - .and_then(|s| if s.is_empty() { None } else { Some(s) }) + .filter(|&s| !s.is_empty()) .ok_or(errors::PutDashboardError::MissingTitle)? .to_owned(); let role = dashboard .role() - .and_then(|s| if s.is_empty() { None } else { Some(s) }) + .filter(|&s| !s.is_empty()) .map(|r| r.to_owned()); let description = dashboard .description() - .and_then(|s| if s.is_empty() { None } else { Some(s) }) + .filter(|&s| !s.is_empty()) .map(|d| d.to_owned()); let version = dashboard.version; let created_at_depricated = dashboard.created_at_deprecated(); @@ -407,9 +407,7 @@ async fn list_models( }; // Apply the optional title substring filter. - let title_pat = params - .title_pat - .and_then(|p| if p.is_empty() { None } else { Some(p) }); + let title_pat = params.title_pat.filter(|p| !p.is_empty()); let query = if let Some(title_pat) = title_pat { let pattern = format!("%{}%", title_pat.to_lowercase()); query.filter(Expr::expr(Func::lower(Expr::col(dashboards::Column::Title))).like(pattern)) diff --git a/src/ingester/src/stream.rs b/src/ingester/src/stream.rs index 3eb797ea27..6c5489b096 100644 --- a/src/ingester/src/stream.rs +++ b/src/ingester/src/stream.rs @@ -78,7 +78,7 @@ impl Stream { ) -> Result<(usize, Vec<(PathBuf, PersistStat)>)> { let mut schema_size = 0; let mut paths = Vec::new(); - for (_, partition) in self.partitions.iter() { + for partition in self.partitions.values() { let (part_schema_size, partitions) = partition .persist(id, idx, org_id, stream_type, stream_name) .await?; diff --git a/src/ingester/src/writer.rs b/src/ingester/src/writer.rs index 3e4b82ef7a..6b1c04997c 100644 --- a/src/ingester/src/writer.rs +++ b/src/ingester/src/writer.rs @@ -302,8 +302,8 @@ impl Writer { log::info!( "[INGESTER:MEM:{idx}] create file: {}/{}/{}/{}.wal", wal_dir.display(), - &key.org_id, - &key.stream_type, + key.org_id, + key.stream_type, wal_id ); @@ -573,8 +573,8 @@ impl Writer { log::info!( "[INGESTER:MEM] create file: {}/{}/{}/{}.wal", wal_dir.display(), - &self.key.org_id, - &self.key.stream_type, + self.key.org_id, + self.key.stream_type, wal_id ); let (new_wal, _header_size) = WalWriter::new( diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index 7161428744..7f50bcc980 100644 --- a/src/job/files/parquet.rs +++ b/src/job/files/parquet.rs @@ -457,7 +457,7 @@ async fn move_files( let wal_dir = wal_dir.clone(); // sort by created time let mut files_with_size = files.to_owned(); - files_with_size.sort_by(|a, b| a.meta.min_ts.cmp(&b.meta.min_ts)); + files_with_size.sort_by_key(|k| k.meta.min_ts); // check the total size let total_original_size: i64 = files_with_size .iter() @@ -672,7 +672,7 @@ async fn merge_files( new_file_size += file.meta.original_size; new_compressed_file_size += file.meta.compressed_size; new_file_list.push(file.clone()); - log::info!("[INGESTER:JOB:{thread_id}] merge small file: {}", &file.key); + log::info!("[INGESTER:JOB:{thread_id}] merge small file: {}", file.key); } // no files need to merge if new_file_list.is_empty() { diff --git a/src/job/metrics.rs b/src/job/metrics.rs index c721664b85..945f7738d7 100644 --- a/src/job/metrics.rs +++ b/src/job/metrics.rs @@ -72,7 +72,7 @@ async fn load_ingest_wal_used_bytes() -> Result<(), anyhow::Error> { Ok(path) => path, Err(_) => return Ok(()), }; - let pattern = format!("{}files/", &cfg.common.data_wal_dir); + let pattern = format!("{}files/", cfg.common.data_wal_dir); let files = scan_files(pattern, "parquet", None) .await .unwrap_or_default(); diff --git a/src/job/mmdb_downloader.rs b/src/job/mmdb_downloader.rs index 64913dff32..342e14cd0c 100644 --- a/src/job/mmdb_downloader.rs +++ b/src/job/mmdb_downloader.rs @@ -65,15 +65,15 @@ async fn run_download_files() { let fname = format!( "{}{}", - &cfg.common.mmdb_data_dir, + cfg.common.mmdb_data_dir, get_o2_config().common.mmdb_enterprise_file_name ); update_maxmind_table(&fname).await; } - let city_fname = format!("{}{}", &cfg.common.mmdb_data_dir, MMDB_CITY_FILE_NAME); - let asn_fname = format!("{}{}", &cfg.common.mmdb_data_dir, MMDB_ASN_FILE_NAME); + let city_fname = format!("{}{}", cfg.common.mmdb_data_dir, MMDB_CITY_FILE_NAME); + let asn_fname = format!("{}{}", cfg.common.mmdb_data_dir, MMDB_ASN_FILE_NAME); let download_city_files = is_digest_different(&city_fname, &cfg.common.mmdb_geolite_citydb_sha256_url) @@ -131,12 +131,12 @@ async fn run_download_files() { /// Update the maxmind client async fn update_maxmind_client() { let cfg = config::get_config(); - let city_fname = format!("{}{}", &cfg.common.mmdb_data_dir, MMDB_CITY_FILE_NAME); + let city_fname = format!("{}{}", cfg.common.mmdb_data_dir, MMDB_CITY_FILE_NAME); #[cfg(feature = "enterprise")] let city_fname = if get_o2_config().common.enable_enterprise_mmdb { format!( "{}{}", - &cfg.common.mmdb_data_dir, + cfg.common.mmdb_data_dir, get_o2_config().common.mmdb_enterprise_file_name ) } else { diff --git a/src/lib.rs b/src/lib.rs index c166edcef2..ddf90c38ff 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -13,7 +13,6 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -#![feature(lazy_get)] #![feature(variant_count)] #![recursion_limit = "256"] diff --git a/src/report_server/src/report.rs b/src/report_server/src/report.rs index 74c7caea7e..e0352f8514 100644 --- a/src/report_server/src/report.rs +++ b/src/report_server/src/report.rs @@ -293,7 +293,7 @@ pub async fn generate_report( models::ReportTimerangeType::Absolute => { let url = format!( "{web_url}/dashboards/view?org_identifier={org_id}&dashboard={dashboard_id}&folder={folder_id}&tab={tab_id}&refresh=Off&{search_type_params}&from={}&to={}&timezone={timezone}&var-Dynamic+filters=%255B%255D&print=true{dashb_vars}", - &timerange.from, &timerange.to + timerange.from, timerange.to ); log::debug!("dashb_url for dashboard {folder_id}/{dashboard_id}: {url}"); @@ -437,7 +437,7 @@ pub async fn send_email( Ok(_) => { log::info!( "email sent successfully for the report {}", - &email_details.name + email_details.name ); Ok(()) } diff --git a/src/router/grpc/ingest/logs.rs b/src/router/grpc/ingest/logs.rs index 766166febe..6fd83ad9f3 100644 --- a/src/router/grpc/ingest/logs.rs +++ b/src/router/grpc/ingest/logs.rs @@ -41,7 +41,7 @@ impl LogsService for LogsServer { if !metadata.contains_key(&cfg.grpc.org_header_key) { return Err(Status::invalid_argument(format!( "Please specify organization id with header key '{}' ", - &cfg.grpc.org_header_key + cfg.grpc.org_header_key ))); } diff --git a/src/router/grpc/ingest/metrics.rs b/src/router/grpc/ingest/metrics.rs index f6f72e2fec..dd919168b5 100644 --- a/src/router/grpc/ingest/metrics.rs +++ b/src/router/grpc/ingest/metrics.rs @@ -41,7 +41,7 @@ impl MetricsService for MetricsServer { if !metadata.contains_key(&cfg.grpc.org_header_key) { return Err(Status::invalid_argument(format!( "Please specify organization id with header key '{}' ", - &cfg.grpc.org_header_key + cfg.grpc.org_header_key ))); } diff --git a/src/router/grpc/ingest/traces.rs b/src/router/grpc/ingest/traces.rs index 55e0b65167..c0553b98cb 100644 --- a/src/router/grpc/ingest/traces.rs +++ b/src/router/grpc/ingest/traces.rs @@ -41,7 +41,7 @@ impl TraceService for TraceServer { if !metadata.contains_key(&cfg.grpc.org_header_key) { return Err(Status::invalid_argument(format!( "Please specify organization id with header key '{}' ", - &cfg.grpc.org_header_key + cfg.grpc.org_header_key ))); } diff --git a/src/router/ratelimit/resource_extractor.rs b/src/router/ratelimit/resource_extractor.rs index f7e34a7897..48593a3054 100644 --- a/src/router/ratelimit/resource_extractor.rs +++ b/src/router/ratelimit/resource_extractor.rs @@ -82,7 +82,7 @@ fn rule_extractor( } else { crate::service::users::get_user_roles(user_email.as_str(), Some(&org_id)).await }; - log::debug!("found user_roles: {:?}", &user_roles); + log::debug!("found user_roles: {:?}", user_roles); let openapi_path = openapi_path.unwrap_or(path); // find the group of the openapi_path match find_group_by_openapi(&openapi_path).await { @@ -106,15 +106,14 @@ fn rule_extractor( ) .await; log::debug!( - "org_id{}, user_roles:{:?}, api_group_name:{}, api_group_operation:{}, user_email:{}, default_resource:{}, found default_rule: {:?}, custom_rules: {:?}", - &org_id, + "org_id{org_id}, user_roles:{:?}, api_group_name:{}, api_group_operation:{}, user_email:{}, default_resource:{}, found default_rule: {:?}, custom_rules: {:?}", user_roles, api_group_name, api_group_operation.as_str(), user_email, default_resource, - &default_rule, - &custom_rules + default_rule, + custom_rules ); // Select the final rule based on priority select_final_rule_resource(default_rule, custom_rules, user_email.as_str()) diff --git a/src/service/alerts/alert.rs b/src/service/alerts/alert.rs index d6034f80ff..d8fc52ba5a 100644 --- a/src/service/alerts/alert.rs +++ b/src/service/alerts/alert.rs @@ -441,13 +441,12 @@ async fn prepare_alert( } } } - QueryType::PromQL => { - if alert.query_condition.promql.is_none() + QueryType::PromQL + if (alert.query_condition.promql.is_none() || alert.query_condition.promql.as_ref().unwrap().is_empty() - || alert.query_condition.promql_condition.is_none() - { - return Err(AlertError::PromqlMissingQuery); - } + || alert.query_condition.promql_condition.is_none()) => + { + return Err(AlertError::PromqlMissingQuery); } _ => {} } @@ -869,9 +868,8 @@ pub async fn trigger_by_id( { Ok(Some(outcome)) => { log::info!( - "Manual trigger for alert {}/{} correlated to incident {} (service: {})", - org_id, - &alert.name, + "Manual trigger for alert {org_id}/{} correlated to incident {} (service: {})", + alert.name, outcome.incident_id(), outcome.service_name(), ); @@ -879,9 +877,8 @@ pub async fn trigger_by_id( } Ok(None) => { log::debug!( - "No incident correlation for manually triggered alert {}/{}", - org_id, - &alert.name + "No incident correlation for manually triggered alert {org_id}/{}", + alert.name ); false } @@ -949,9 +946,8 @@ pub async fn trigger_by_name( { Ok(Some(outcome)) => { log::info!( - "Manual trigger for alert {}/{} correlated to incident {} (service: {})", - org_id, - &alert.name, + "Manual trigger for alert {org_id}/{} correlated to incident {} (service: {})", + alert.name, outcome.incident_id(), outcome.service_name(), ); @@ -959,9 +955,8 @@ pub async fn trigger_by_name( } Ok(None) => { log::debug!( - "No incident correlation for manually triggered alert {}/{}", - org_id, - &alert.name + "No incident correlation for manually triggered alert {org_id}/{}", + alert.name ); false } @@ -1599,18 +1594,10 @@ async fn process_dest_template( }; let mut alert_query = String::new(); - let function_content = if alert.query_condition.vrl_function.is_none() { - "".to_owned() + let function_content = if let Some(v) = &alert.query_condition.vrl_function { + format!("&functionContent={}", v.replace('+', "%2B")) } else { - format!( - "&functionContent={}", - alert - .query_condition - .vrl_function - .as_ref() - .unwrap() - .replace('+', "%2B") - ) + "".to_owned() }; let alert_url = if alert.query_condition.query_type == QueryType::PromQL { if let Some(promql) = &alert.query_condition.promql { diff --git a/src/service/alerts/deduplication.rs b/src/service/alerts/deduplication.rs index 9b390f8cef..b6438ab872 100644 --- a/src/service/alerts/deduplication.rs +++ b/src/service/alerts/deduplication.rs @@ -217,55 +217,50 @@ async fn apply_deduplication_impl( // Check if this fingerprint exists and is within time window let should_send = match get_dedup_state(db, &fingerprint).await? { - Some(existing_state) => { - if is_within_window(&existing_state, time_window_minutes) { - // Within window - update occurrence count but don't send - if let Err(e) = save_dedup_state( - db, - DedupStateParams { - org_id: org_id.as_str(), - fingerprint: &fingerprint, - alert_id: &alert_id, - first_seen_at: existing_state.first_seen_at, - last_seen_at: now, - occurrence_count: existing_state.occurrence_count + 1, - }, - ) - .await - { - log::warn!( - "Failed to update dedup state for fingerprint {}: {}", - fingerprint, - e - ); - } - - // Record suppression metric - let dedup_type = if existing_state.alert_id == alert_id { - "same_alert" - } else { - "cross_alert" - }; - config::metrics::ALERT_DEDUP_SUPPRESSED_TOTAL - .with_label_values(&[org_id.as_str(), &alert.name, dedup_type]) - .inc(); - - log::debug!( - "[dedup] Suppressed alert '{}' for org: {}, fingerprint: {}, occurrence: {}, type: {}", - alert.name, - org_id, + Some(existing_state) if is_within_window(&existing_state, time_window_minutes) => { + // Within window - update occurrence count but don't send + if let Err(e) = save_dedup_state( + db, + DedupStateParams { + org_id: org_id.as_str(), + fingerprint: &fingerprint, + alert_id: &alert_id, + first_seen_at: existing_state.first_seen_at, + last_seen_at: now, + occurrence_count: existing_state.occurrence_count + 1, + }, + ) + .await + { + log::warn!( + "Failed to update dedup state for fingerprint {}: {}", fingerprint, - existing_state.occurrence_count + 1, - dedup_type + e ); - - false - } else { - // Outside window - treat as new alert - true } + + // Record suppression metric + let dedup_type = if existing_state.alert_id == alert_id { + "same_alert" + } else { + "cross_alert" + }; + config::metrics::ALERT_DEDUP_SUPPRESSED_TOTAL + .with_label_values(&[org_id.as_str(), &alert.name, dedup_type]) + .inc(); + + log::debug!( + "[dedup] Suppressed alert '{}' for org: {}, fingerprint: {}, occurrence: {}, type: {}", + alert.name, + org_id, + fingerprint, + existing_state.occurrence_count + 1, + dedup_type + ); + + false } - None => { + _ => { // New fingerprint - should send true } diff --git a/src/service/alerts/derived_streams.rs b/src/service/alerts/derived_streams.rs index 8b253c720b..9f6fd1ea71 100644 --- a/src/service/alerts/derived_streams.rs +++ b/src/service/alerts/derived_streams.rs @@ -86,18 +86,17 @@ pub async fn save( )); } } - QueryType::PromQL => { - if derived_stream + QueryType::PromQL + if (derived_stream .query_condition .promql .as_ref() .is_some_and(|promql| promql.is_empty()) - || derived_stream.query_condition.promql_condition.is_none() - { - return Err(anyhow::anyhow!( - "Scheduled pipeline with PromQL mode should have a query and condition" - )); - } + || derived_stream.query_condition.promql_condition.is_none()) => + { + return Err(anyhow::anyhow!( + "Scheduled pipeline with PromQL mode should have a query and condition" + )); } _ => {} }; diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index fe0964953f..5925d33151 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -390,6 +390,18 @@ impl QueryConditionExt for QueryCondition { .await // SearchService::search_multi(&trace_id, org_id, stream_type, None, &req).await } else { + let encode_query_fn = if let Some(v) = &self.vrl_function { + match base64::decode_url(v) { + Ok(v) => Some(v), + Err(e) => { + return Err(anyhow::anyhow!( + "Error decoding alert vrl query function: {e}" + )); + } + } + } else { + None + }; // fire the query let req = config::meta::search::Request { query: config::meta::search::Query { @@ -403,19 +415,7 @@ impl QueryConditionExt for QueryCondition { track_total_hits: false, action_id: None, uses_zo_fn: false, - query_fn: if self.vrl_function.is_some() { - match base64::decode_url(self.vrl_function.as_ref().unwrap()) { - Ok(query_fn) => Some(query_fn), - Err(e) => { - return Err(anyhow::anyhow!( - "Error decoding alert vrl query function: {e}" /* TODO: update - * error msg */ - )); - } - } - } else { - None - }, + query_fn: encode_query_fn, skip_wal: false, sampling_config: None, sampling_ratio: None, @@ -570,7 +570,7 @@ impl ConditionListExt for ConditionList { let data_type = match schema.field_with_name(&cond.column) { Ok(field) => field.data_type(), Err(_) => { - return Err(anyhow::anyhow!("Column {} not found", &cond.column)); + return Err(anyhow::anyhow!("Column {} not found", cond.column)); } }; cond_sql_list.push(build_expr(cond, "", data_type)?); @@ -591,7 +591,7 @@ impl ConditionListExt for ConditionList { let data_type = match schema.field_with_name(&node.column) { Ok(field) => field.data_type(), Err(_) => { - return Err(anyhow::anyhow!("Column {} not found", &node.column)); + return Err(anyhow::anyhow!("Column {} not found", node.column)); } }; build_expr(node, "", data_type) @@ -1007,7 +1007,7 @@ async fn condition_item_to_sql( let data_type = match schema.field_with_name(&condition.column) { Ok(field) => field.data_type(), Err(_) => { - return Err(anyhow::anyhow!("Column {} not found", &condition.column)); + return Err(anyhow::anyhow!("Column {} not found", condition.column)); } }; @@ -1062,7 +1062,7 @@ pub async fn build_sql( Err(_) => { return Err(anyhow::anyhow!( "Aggregation column {} not found on stream {stream_name}", - &agg.having.column, + agg.having.column, )); } }; diff --git a/src/service/alerts/scheduler/handlers.rs b/src/service/alerts/scheduler/handlers.rs index 26c5aefe8d..f03c84b7ff 100644 --- a/src/service/alerts/scheduler/handlers.rs +++ b/src/service/alerts/scheduler/handlers.rs @@ -352,7 +352,7 @@ async fn handle_alert_triggers( let (_, max_retries) = get_scheduler_max_retries(); log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] Inside handle_alert_triggers: processing trigger: {}", - &trigger.module_key + trigger.module_key ); let now = Utc::now().timestamp_micros(); let triggered_at = trigger.start_time.unwrap_or_default(); @@ -571,8 +571,8 @@ async fn handle_alert_triggers( if is_realtime && is_silenced { log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] Realtime alert need wakeup, {}/{}", - &trigger.org, - &trigger.module_key + trigger.org, + trigger.module_key ); // wakeup the trigger db::scheduler::update_trigger(new_trigger, true, &query_trace_id).await?; @@ -604,8 +604,8 @@ async fn handle_alert_triggers( // next_run_at to the next expected trigger time log::info!( "[SCHEDULER trace_id {scheduler_trace_id}] This alert trigger: {}/{} has passed maximum retries, skipping to next run", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); new_trigger.next_run_at = @@ -685,7 +685,7 @@ async fn handle_alert_triggers( } log::info!( "[SCHEDULER trace_id {scheduler_trace_id}] alert {} skipped due to delay: {}", - &trigger.module_key, + trigger.module_key, delay ); (now - final_end_time, true) @@ -744,7 +744,7 @@ async fn handle_alert_triggers( let err_string = err.to_string(); log::error!( "[SCHEDULER trace_id {scheduler_trace_id}] alert {} evaluation failed: {}", - &new_trigger.module_key, + new_trigger.module_key, err_string ); if err_string.starts_with("Partial") { @@ -761,8 +761,8 @@ async fn handle_alert_triggers( if let Err(e) = set_without_updating_trigger(&trigger.org, alert_curr).await { log::error!( "[SCHEDULER trace_id {scheduler_trace_id}] Failed to update alert: {}/{} after trigger: {e}", - &trigger.org, - &trigger.module_key + trigger.org, + trigger.module_key ); } } @@ -804,14 +804,14 @@ async fn handle_alert_triggers( trigger_data_stream.query_took = trigger_results.query_took; log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] result of alert {} evaluation matched condition: {}", - &new_trigger.module_key, + new_trigger.module_key, trigger_results.data.is_some(), ); if trigger_results.data.is_some() { log::info!( "[SCHEDULER trace_id {scheduler_trace_id}] Alert conditions satisfied, org: {}, module_key: {}", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); } if let Some(tolerance) = alert.trigger_condition.tolerance_in_secs @@ -903,10 +903,9 @@ async fn handle_alert_triggers( }; log::debug!( - "[SCHEDULER trace_id {scheduler_trace_id}] Adding alert to batch, org: {}, alert: {}, fingerprint: {}, rows: {}", - &new_trigger.org, - &alert.name, - fingerprint, + "[SCHEDULER trace_id {scheduler_trace_id}] Adding alert to batch, org: {}, alert: {}, fingerprint: {fingerprint}, rows: {}", + new_trigger.org, + alert.name, data.len() ); @@ -922,8 +921,7 @@ async fn handle_alert_triggers( if batch_ready { log::info!( - "[SCHEDULER trace_id {scheduler_trace_id}] Batch {} reached max size, sending immediately", - fingerprint + "[SCHEDULER trace_id {scheduler_trace_id}] Batch {fingerprint} reached max size, sending immediately", ); if let Some(batch) = crate::service::alerts::grouping::get_ready_batch(&fingerprint) @@ -978,8 +976,8 @@ async fn handle_alert_triggers( if deduplicated_data.is_empty() && deduplicated { log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] All alert results deduplicated for org: {}, module_key: {}", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); // Mark as suppressed for history tracking @@ -1002,10 +1000,9 @@ async fn handle_alert_triggers( } Err(e) => { log::error!( - "[SCHEDULER trace_id {scheduler_trace_id}] Error applying deduplication for org: {}, module_key: {}: {}", - &new_trigger.org, - &new_trigger.module_key, - e + "[SCHEDULER trace_id {scheduler_trace_id}] Error applying deduplication for org: {}, module_key: {}: {e}", + new_trigger.org, + new_trigger.module_key, ); // On error, continue with original data to avoid missing alerts data @@ -1065,8 +1062,8 @@ async fn handle_alert_triggers( Ok(Some(outcome)) => { log::info!( "[SCHEDULER trace_id {scheduler_trace_id}] Alert {}/{} correlated to incident {} (service: {})", - &new_trigger.org, - &alert.name, + new_trigger.org, + alert.name, outcome.incident_id(), outcome.service_name(), ); @@ -1077,8 +1074,8 @@ async fn handle_alert_triggers( Ok(None) => { log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] No incident correlation for alert {}/{}", - &new_trigger.org, - &alert.name, + new_trigger.org, + alert.name, ); false } @@ -1150,15 +1147,15 @@ async fn handle_alert_triggers( if !err_msg.is_empty() { log::error!( "[SCHEDULER trace_id {scheduler_trace_id}] Some notifications for alert {}/{} could not be sent: {err_msg}", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); trigger_data_stream.error = Some(err_msg); } else { log::info!( "[SCHEDULER trace_id {scheduler_trace_id}] Alert notification sent, org: {}, module_key: {}", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); } trigger_data_stream.success_response = Some(success_msg); @@ -1177,16 +1174,16 @@ async fn handle_alert_triggers( Err(e) => { log::error!( "[SCHEDULER trace_id {scheduler_trace_id}] Error sending alert notification: org: {}, module_key: {}", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); if trigger.retries + 1 >= max_retries { // It has been tried the maximum time, just update the // next_run_at to the next expected trigger time log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] This alert trigger: {}/{} has reached maximum retries", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); // Alert could not be sent for multiple times, in the next run // if the same start time used for alert evaluation, the extended @@ -1227,8 +1224,8 @@ async fn handle_alert_triggers( } else { log::info!( "[SCHEDULER trace_id {scheduler_trace_id}] Alert conditions not satisfied, org: {}, module_key: {}", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); // Condition did not match, store the last used end_time in the triggers // In the next run, the alert will be checked from the last end_time @@ -1246,7 +1243,7 @@ async fn handle_alert_triggers( log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] publish_triggers_usage for alert: {}", - &trigger_data_stream.key + trigger_data_stream.key ); // publish the triggers as stream publish_triggers_usage(trigger_data_stream); @@ -1348,8 +1345,8 @@ async fn handle_report_triggers( let (_, max_retries) = get_scheduler_max_retries(); log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] Inside handle_report_trigger,org: {}, module_key: {}", - &trigger.org, - &trigger.module_key + trigger.org, + trigger.module_key ); let org_id = &trigger.org; // For report, trigger.module_key is the report name @@ -1576,7 +1573,7 @@ async fn handle_report_triggers( // next_run_at to the next expected trigger time log::info!( "[SCHEDULER trace_id {scheduler_trace_id}] This report trigger: {org_id}/{report_name} has passed maximum retries, skipping to next run", - org_id = &new_trigger.org, + org_id = new_trigger.org, report_name = report_name ); db::scheduler::update_trigger(new_trigger, true, &query_trace_id).await?; @@ -1661,7 +1658,7 @@ async fn handle_report_triggers( } log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] publish_triggers_usage for report: {}", - &trigger_data_stream.key + trigger_data_stream.key ); publish_triggers_usage(trigger_data_stream); @@ -2297,8 +2294,8 @@ async fn handle_derived_stream_triggers( } else { log::info!( "[SCHEDULER trace_id {scheduler_trace_id}] DerivedStream condition does not match any data for the period, org: {}, module_key: {}", - &new_trigger.org, - &new_trigger.module_key + new_trigger.org, + new_trigger.module_key ); trigger_data_stream.status = TriggerDataStatus::ConditionNotSatisfied; trigger_data_stream.query_took = trigger_results.query_took; @@ -2366,8 +2363,8 @@ async fn handle_derived_stream_triggers( // Report a final pipeline error log::warn!( "[SCHEDULER trace_id {scheduler_trace_id}] Pipeline({}/{})]: DerivedStream trigger has reached maximum retries.", - &pipeline.org, - &pipeline.name + pipeline.org, + pipeline.name ); let err_msg = format!( "[SCHEDULER trace_id {scheduler_trace_id}] DerivedStream has reached max retries of {max_retries}. Pipeline will be retried after the next scheduled run. Please fix reported errors in pipeline." @@ -2390,8 +2387,8 @@ async fn handle_derived_stream_triggers( if let Err(e) = db::scheduler::update_trigger(new_trigger, true, &query_trace_id).await { log::warn!( "[SCHEDULER trace_id {scheduler_trace_id}] Pipeline({}/{})]: DerivedStream's new trigger failed to be updated, caused by {}", - &pipeline.org, - &pipeline.name, + pipeline.org, + pipeline.name, e ); } @@ -2437,7 +2434,7 @@ async fn handle_backfill_triggers( let scheduler_trace_id = format!("{trace_id}/{query_trace_id}"); log::debug!( "[SCHEDULER trace_id {scheduler_trace_id}] Processing backfill trigger: {}", - &job_id + job_id ); let now = Utc::now().timestamp_micros(); @@ -2450,7 +2447,7 @@ async fn handle_backfill_triggers( Err(e) => { log::error!( "[SCHEDULER trace_id {trace_id}] [job_id: {}] Failed to fetch backfill job config: {e}", - &job_id + job_id ); // Delete the trigger if config is not found let _ = db::scheduler::delete( @@ -2492,7 +2489,7 @@ async fn handle_backfill_triggers( if !config.enabled { log::debug!( "[SCHEDULER trace_id {trace_id}] [job_id: {}] Backfill job is disabled, marking as completed", - &job_id + job_id ); // Mark trigger as Completed when disabled let _ = db::scheduler::update_trigger( @@ -2513,7 +2510,7 @@ async fn handle_backfill_triggers( Err(e) => { log::error!( "[SCHEDULER trace_id {trace_id}] [job_id: {}] Failed to parse backfill trigger data: {e}", - &job_id + job_id ); let new_retries = trigger.retries + 1; if new_retries >= max_retries { @@ -2571,7 +2568,7 @@ async fn handle_backfill_triggers( None => { log::error!( "[SCHEDULER trace_id {trace_id}] [job_id: {}] Missing backfill job data in trigger", - &job_id + job_id ); let new_retries = trigger.retries + 1; if new_retries >= max_retries { @@ -2630,7 +2627,7 @@ async fn handle_backfill_triggers( Err(e) => { log::error!( "[SCHEDULER trace_id {trace_id}] [job_id: {}] Failed to fetch pipeline {}: {e}", - &job_id, + job_id, config.pipeline_id ); if trigger.retries + 1 >= max_retries { @@ -2706,7 +2703,7 @@ async fn handle_backfill_triggers( _ => { log::error!( "[SCHEDULER trace_id {trace_id}] [job_id: {}] Pipeline {} is not scheduled", - &job_id, + job_id, config.pipeline_id ); // Delete the trigger as this is a configuration error @@ -2746,7 +2743,7 @@ async fn handle_backfill_triggers( Err(e) => { log::error!( "[SCHEDULER trace_id {trace_id}] [job_id: {}] Failed to get destination streams: {e}", - &job_id + job_id ); let _ = db::scheduler::delete( &trigger.org, @@ -2790,7 +2787,7 @@ async fn handle_backfill_triggers( // Initiate deletion for all destination streams log::info!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Starting deletion for {} destination stream(s), time range {}-{}", - &job_id, + job_id, destination_streams.len(), config.start_time, config.end_time @@ -2804,7 +2801,7 @@ async fn handle_backfill_triggers( for (idx, stream) in destination_streams.iter().enumerate() { log::debug!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Initiating deletion for stream {}/{} ({}/{})", - &job_id, + job_id, stream.stream_type, stream.stream_name, idx + 1, @@ -2823,7 +2820,7 @@ async fn handle_backfill_triggers( deletion_job_ids.push(deletion_job_id.clone()); log::debug!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Deletion job {} created for stream {}/{}", - &job_id, + job_id, deletion_job_id, stream.stream_type, stream.stream_name @@ -2836,7 +2833,7 @@ async fn handle_backfill_triggers( ); log::error!( "[BACKFILL trace_id {trace_id}] [job_id: {}] {}", - &job_id, + job_id, error_msg ); failed = true; @@ -2892,7 +2889,7 @@ async fn handle_backfill_triggers( log::info!( "[BACKFILL trace_id {trace_id}] [job_id: {}] {} deletion job(s) initiated, will check status in {}s", - &job_id, + job_id, deletion_job_ids.len(), delay ); @@ -2912,7 +2909,7 @@ async fn handle_backfill_triggers( Ok(status) => { log::debug!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Deletion job {} status: {}", - &job_id, + job_id, deletion_job_id, status ); @@ -2921,7 +2918,7 @@ async fn handle_backfill_triggers( Err(e) => { log::warn!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Failed to check deletion job {} status: {}", - &job_id, + job_id, deletion_job_id, e ); @@ -2933,7 +2930,7 @@ async fn handle_backfill_triggers( if all_completed { log::info!( "[BACKFILL trace_id {trace_id}] [job_id: {}] All {} deletion job(s) completed, starting backfill", - &job_id, + job_id, backfill_job.deletion_job_ids.len() ); backfill_job.deletion_status = DeletionStatus::Completed; @@ -2958,7 +2955,7 @@ async fn handle_backfill_triggers( log::debug!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Deletion in progress ({}/{} completed), checking again in {}s", - &job_id, + job_id, completed_count, backfill_job.deletion_job_ids.len(), delay @@ -2988,7 +2985,7 @@ async fn handle_backfill_triggers( log::debug!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Processing chunk: {}-{}", - &job_id, + job_id, query_start_time, query_end_time ); @@ -3006,7 +3003,7 @@ async fn handle_backfill_triggers( Err(e) => { log::error!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Failed to evaluate pipeline: {e}", - &job_id + job_id ); let error_msg = e.to_string(); @@ -3021,7 +3018,7 @@ async fn handle_backfill_triggers( // Max retries reached, report error and reset retries for next scheduled run log::warn!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Backfill job for pipeline {} has reached maximum retries.", - &job_id, + job_id, config.pipeline_id ); @@ -3096,7 +3093,7 @@ async fn handle_backfill_triggers( Err(e) => { log::error!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Failed to create executable pipeline: {e}", - &job_id + job_id ); let error_msg = e.to_string(); @@ -3111,7 +3108,7 @@ async fn handle_backfill_triggers( // Max retries reached, report error and reset retries for next scheduled run log::warn!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Backfill job for pipeline {} has reached maximum retries on pipeline creation.", - &job_id, + job_id, config.pipeline_id ); @@ -3196,7 +3193,7 @@ async fn handle_backfill_triggers( Err(e) => { log::error!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Failed to process batch: {e}", - &job_id + job_id ); let error_msg = e.to_string(); @@ -3214,7 +3211,7 @@ async fn handle_backfill_triggers( // Max retries reached, report error and reset retries for next scheduled run log::warn!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Backfill job for pipeline {} has reached maximum retries on batch processing.", - &job_id, + job_id, config.pipeline_id ); @@ -3336,14 +3333,14 @@ async fn handle_backfill_triggers( Ok(resp) if resp.status_code == 200 => { log::info!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Backfill data ingested to destination {org_id}/{stream_name}/{stream_type}, records: {records_len}", - &job_id + job_id ); } error => { let err = error.map_or_else(|e| e.to_string(), |resp| resp.message); log::error!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Failed to ingest backfill data to destination {org_id}/{stream_name}/{stream_type}: {err}", - &job_id + job_id ); ingestion_error = Some(format!( "Failed to ingest to {org_id}/{stream_name}/{stream_type}: {err}" @@ -3369,7 +3366,7 @@ async fn handle_backfill_triggers( // Max retries reached, report error and reset retries for next scheduled run log::warn!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Backfill job for pipeline {} has reached maximum retries on ingestion.", - &job_id, + job_id, config.pipeline_id ); @@ -3448,7 +3445,7 @@ async fn handle_backfill_triggers( log::info!( "[BACKFILL trace_id {scheduler_trace_id}] [job_id: {}] Backfill job completed for pipeline {}", - &job_id, + job_id, config.pipeline_id ); @@ -3547,7 +3544,7 @@ async fn handle_backfill_triggers( * 100.0) as u8; log::debug!( "[BACKFILL trace_id {trace_id}] [job_id: {}] Progress: {}%, next chunk in {}s", - &job_id, + job_id, progress, delay ); diff --git a/src/service/compact/bloom_build.rs b/src/service/compact/bloom_build.rs index bbebe6fa0f..836464d718 100644 --- a/src/service/compact/bloom_build.rs +++ b/src/service/compact/bloom_build.rs @@ -139,7 +139,7 @@ pub(crate) async fn build_for_stream( // chunk's max cardinality). let fpp = cfg.common.bloom_filter_fpp; let max_files_per_bf = cfg.common.bloom_filter_max_files_per_bf.max(1); - files.sort_by(|a, b| b.meta.records.cmp(&a.meta.records)); + files.sort_by_key(|k| std::cmp::Reverse(k.meta.records)); let base_ver = now_micros(); let chunk_total = files.len().div_ceil(max_files_per_bf); diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs index 6a30a2a3ab..d21bf6e928 100644 --- a/src/service/compact/merge.rs +++ b/src/service/compact/merge.rs @@ -450,10 +450,10 @@ pub async fn merge_by_stream( let job_strategy = MergeStrategy::from(&cfg.compact.strategy); match job_strategy { MergeStrategy::FileSize => { - files_with_size.sort_by(|a, b| a.meta.original_size.cmp(&b.meta.original_size)); + files_with_size.sort_by_key(|k| k.meta.original_size); } MergeStrategy::FileTime => { - files_with_size.sort_by(|a, b| a.meta.min_ts.cmp(&b.meta.min_ts)); + files_with_size.sort_by_key(|k| k.meta.min_ts); } MergeStrategy::TimeRange => { files_with_size = sort_by_time_range(files_with_size); @@ -807,7 +807,7 @@ pub async fn merge_files( fi += 1; log::info!( "[COMPACTOR:WORKER:{thread_id}:{fi}] merge small file: {}", - &file.key + file.key ); let buf = file_data::get(&file.account, &file.key, None).await?; let file_format = FileFormat::from_extension(&file.key) @@ -817,7 +817,7 @@ pub async fn merge_files( Err(e) => { log::error!( "[COMPACTOR:WORKER:{thread_id}:{fi}] read schema error for file: {}, err: {e}", - &file.key + file.key ); return Err(e); } @@ -953,7 +953,7 @@ pub async fn merge_files( new_files.push(FileKey::new(0, account, new_file_key, new_file_meta, false)); } MergeParquetResult::Multiple { bufs, file_metas } => { - for (buf, file_meta) in bufs.into_iter().zip(file_metas.into_iter()) { + for (buf, file_meta) in bufs.into_iter().zip(file_metas) { let mut new_file_meta = file_meta; new_file_meta.compressed_size = buf.len() as i64; if new_file_meta.compressed_size == 0 { diff --git a/src/service/compact/mod.rs b/src/service/compact/mod.rs index 89ea7a4ec3..d6cbd6c19a 100644 --- a/src/service/compact/mod.rs +++ b/src/service/compact/mod.rs @@ -149,9 +149,9 @@ pub async fn run_generate_job(job_type: CompactionJobType) -> Result<(), anyhow: ) { log::warn!( "[COMPACTOR] the stream [{}/{}/{}] is deleting, just skip", - &org_id, + org_id, stream_type, - &stream_name, + stream_name, ); continue; } @@ -244,10 +244,7 @@ pub async fn run_generate_downsampling_job() -> Result<(), anyhow::Error> { None, ) { log::warn!( - "[DOWNSAMPLING] the stream [{}/{}/{}] is deleting, just skip", - &org_id, - stream_type, - &stream_name, + "[DOWNSAMPLING] the stream [{org_id}/{stream_type}/{stream_name}] is deleting, just skip", ); continue; } diff --git a/src/service/dashboards/mod.rs b/src/service/dashboards/mod.rs index a1995d6ae6..23c869a7f1 100644 --- a/src/service/dashboards/mod.rs +++ b/src/service/dashboards/mod.rs @@ -616,7 +616,7 @@ async fn put( let title = dashboard .title() .map(|t| t.trim().to_string()) - .and_then(|t| if t.is_empty() { None } else { Some(t) }) + .filter(|t| !t.is_empty()) .ok_or_else(|| DashboardError::PutMissingTitle)?; dashboard.set_title(title); diff --git a/src/service/dashboards/reports.rs b/src/service/dashboards/reports.rs index 61a2e0f7bb..966f7487e0 100644 --- a/src/service/dashboards/reports.rs +++ b/src/service/dashboards/reports.rs @@ -284,7 +284,7 @@ pub async fn list( || permitted .as_ref() .unwrap() - .contains(&format!("report:{}", &report.report_name)) + .contains(&format!("report:{}", report.report_name)) || permitted .as_ref() .unwrap() @@ -581,7 +581,7 @@ impl SendReport for Report { let url = url::Url::parse(&format!( "{}/api/{}/reports/{}/send", - &cfg.common.report_server_url, &self.org_id, &self.name + cfg.common.report_server_url, self.org_id, self.name )) .unwrap(); match Client::builder() @@ -602,7 +602,7 @@ impl SendReport for Report { resp.bytes().await, )); } - log::info!("report sent successfully for the report {}", &self.name); + log::info!("report sent successfully for the report {}", self.name); } Err(e) => { return Err(SendReportError::ReportServerClientError(e)); @@ -682,7 +682,7 @@ async fn send_email( // Send the email match SMTP_CLIENT.as_ref().unwrap().send(email).await { Ok(_) => { - log::info!("email sent successfully for the report {}", &report.name); + log::info!("email sent successfully for the report {}", report.name); Ok(()) } Err(e) => Err(SendReportError::SendEmailError(e)), @@ -768,7 +768,7 @@ async fn generate_report( }); let web_url = format!("{}{}/web", cfg.common.web_url, cfg.common.base_uri); - log::debug!("Navigating to web url: {}", &web_url); + log::debug!("Navigating to web url: {}", web_url); let page = browser .new_page(&format!("{web_url}/login?login_as_internal_user=true")) .await?; @@ -863,7 +863,7 @@ async fn generate_report( ReportTimerangeType::Absolute => { let url = format!( "{web_url}/dashboards/view?org_identifier={org_id}&dashboard={dashboard_id}&folder={folder_id}&tab={tab_id}&refresh=Off&{search_type_params}&from={}&to={}&timezone={timezone}&var-Dynamic+filters=%255B%255D&print=true{dashb_vars}", - &timerange.from, &timerange.to + timerange.from, timerange.to ); (url.clone(), url) } diff --git a/src/service/db/alerts/destinations.rs b/src/service/db/alerts/destinations.rs index 92cb5ee8b1..d6c06fca96 100644 --- a/src/service/db/alerts/destinations.rs +++ b/src/service/db/alerts/destinations.rs @@ -91,7 +91,7 @@ pub async fn set(destination: Destination) -> Result Result<(), anyhow::Error> { item_key = updated_item_key; // Get or parse the trigger value - let item_value: db::scheduler::Trigger = - if ev.value.is_none() || ev.value.as_ref().unwrap().is_empty() { - match db::scheduler::get( - &org_id, - config::meta::triggers::TriggerModule::Alert, - &alert_id, - ) - .await - { - Ok(val) => val, - Err(e) => { - log::error!("Error getting value: {e}"); - continue; - } + let item_value: db::scheduler::Trigger = if let Some(val) = &ev.value + && !val.is_empty() + { + match json::from_slice(val) { + Ok(val) => val, + Err(e) => { + log::error!("Error parsing trigger value: {e}"); + continue; } - } else { - match json::from_slice(&ev.value.unwrap()) { - Ok(val) => val, - Err(e) => { - log::error!("Error parsing trigger value: {e}"); - continue; - } + } + } else { + match db::scheduler::get( + &org_id, + config::meta::triggers::TriggerModule::Alert, + &alert_id, + ) + .await + { + Ok(val) => val, + Err(e) => { + log::error!("Error getting value: {e}"); + continue; } - }; + } + }; REALTIME_ALERT_TRIGGERS .write() diff --git a/src/service/db/file_list/broadcast.rs b/src/service/db/file_list/broadcast.rs index 054afa5d97..c8f7636f7b 100644 --- a/src/service/db/file_list/broadcast.rs +++ b/src/service/db/file_list/broadcast.rs @@ -127,7 +127,7 @@ pub async fn send(items: &[FileKey]) -> Result<(), anyhow::Error> { node_items.iter().for_each(|item| { log::info!( "[broadcast] send event to node[{}]: file: {}, deleted: {}", - &node_addr, + node_addr, item.key, item.deleted, ); @@ -147,7 +147,7 @@ pub async fn send(items: &[FileKey]) -> Result<(), anyhow::Error> { if let Err(e) = send_to_node(node, &mut rx).await { log::error!( "[broadcast] send event to node[{}] channel failed, channel closed: {}", - &node_addr, + node_addr, e ); } @@ -163,7 +163,7 @@ pub async fn send(items: &[FileKey]) -> Result<(), anyhow::Error> { } log::error!( "[broadcast] send event to node[{}] channel failed, {}, retrying...", - &node_addr, + node_addr, e ); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; @@ -175,7 +175,7 @@ pub async fn send(items: &[FileKey]) -> Result<(), anyhow::Error> { if !ok { log::error!( "[broadcast] send event to node[{}] channel failed, dropping events", - &node_addr + node_addr ); } } @@ -195,7 +195,7 @@ async fn send_to_node( EVENTS.write().await.remove(&node.uuid); log::error!( "[broadcast] node[{}] leaved cluster, dropping events", - &node.grpc_addr, + node.grpc_addr, ); return Ok(()); } @@ -218,7 +218,7 @@ async fn send_to_node( Err(e) => { log::error!( "[broadcast] connect to node[{}] failed: {}", - &node.grpc_addr, + node.grpc_addr, e ); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; @@ -243,7 +243,7 @@ async fn send_to_node( let items = match rx.recv().await { Some(v) => v, None => { - log::info!("[broadcast] node[{}] channel closed", &node.grpc_addr); + log::info!("[broadcast] node[{}] channel closed", node.grpc_addr); EVENTS.write().await.remove(&node.uuid); return Ok(()); } @@ -262,7 +262,7 @@ async fn send_to_node( if retry_ttl >= 1800 { log::error!( "[broadcast] to node[{}] timeout, dropping event, already retried for 30 minutes", - &node.grpc_addr + node.grpc_addr ); break; } @@ -275,13 +275,13 @@ async fn send_to_node( EVENTS.write().await.remove(&node.uuid); log::error!( "[broadcast] node[{}] leaved cluster, dropping events", - &node.grpc_addr, + node.grpc_addr, ); return Ok(()); } log::error!( "[broadcast] send event to node[{}] failed: {}, retrying...", - &node.grpc_addr, + node.grpc_addr, e ); tokio::time::sleep(tokio::time::Duration::from_secs(wait_ttl)).await; diff --git a/src/service/db/ofga.rs b/src/service/db/ofga.rs index 536aa20e65..33643bcecf 100644 --- a/src/service/db/ofga.rs +++ b/src/service/db/ofga.rs @@ -129,7 +129,7 @@ pub async fn watch() -> Result<(), anyhow::Error> { continue; } }; - log::info!("[WATCH] Got store id {}", &item_value.store_id); + log::info!("[WATCH] Got store id {}", item_value.store_id); OFGA_STORE_ID.insert("store_id".to_owned(), item_value.store_id); } db::Event::Delete(_) => { @@ -146,7 +146,7 @@ pub async fn cache() -> Result<(), anyhow::Error> { let ret = db::list(key).await?; for (_, item_value) in ret { let json_val: OFGAModel = json::from_slice(&item_value).unwrap(); - log::info!("Caching store id {}", &json_val.store_id); + log::info!("Caching store id {}", json_val.store_id); OFGA_STORE_ID.insert("store_id".to_owned(), json_val.store_id); } log::info!("/ofga/model Cached "); diff --git a/src/service/db/pipeline.rs b/src/service/db/pipeline.rs index 35569666b7..a0fc3571c9 100644 --- a/src/service/db/pipeline.rs +++ b/src/service/db/pipeline.rs @@ -320,7 +320,7 @@ async fn update_cache(event: PipelineTableEvent<'_>) { .super_cluster .enabled { - let key = format!("{PIPELINES_WATCH_PREFIX}{}", &pipeline.id); + let key = format!("{PIPELINES_WATCH_PREFIX}{}", pipeline.id); match config::utils::json::to_vec(pipeline) { Err(e) => { log::error!( @@ -436,7 +436,7 @@ pub async fn watch() -> Result<(), anyhow::Error> { stream_exec_pl.insert(stream_params.clone(), exec_pl); log::info!( "[Pipeline::watch]: realtime pipeline {} added to cache.", - &pipeline.id + pipeline.id ); } }; diff --git a/src/service/db/schema.rs b/src/service/db/schema.rs index b5fafb3aff..72d3373795 100644 --- a/src/service/db/schema.rs +++ b/src/service/db/schema.rs @@ -425,8 +425,10 @@ pub async fn watch() -> Result<(), anyhow::Error> { } else { (ev.key.to_string(), ev.start_dt.unwrap_or_default()) }; - if ev_start_dt == 0 && ev.start_dt.is_some() { - ev_start_dt = ev.start_dt.unwrap(); + if ev_start_dt == 0 + && let Some(start_dt) = ev.start_dt + { + ev_start_dt = start_dt; } let item_key = ev_key.strip_prefix(key).unwrap(); @@ -650,7 +652,7 @@ pub async fn cache() -> Result<(), anyhow::Error> { if schema_versions.is_empty() { continue; } - schema_versions.sort_by(|a, b| a.0.cmp(&b.0)); + schema_versions.sort_by_key(|k| k.0); let latest_schema: Vec = json::from_slice(&schema_versions.last().unwrap().1) .map_err(|e| { anyhow::anyhow!("Error parsing schema, key: {}, error: {}", item_key, e) diff --git a/src/service/enrichment_table/geoip.rs b/src/service/enrichment_table/geoip.rs index 8f513c7790..ed1da51388 100644 --- a/src/service/enrichment_table/geoip.rs +++ b/src/service/enrichment_table/geoip.rs @@ -100,7 +100,7 @@ impl Default for GeoipConfig { GeoipConfig { path: format!( "{}{}", - &get_config().common.mmdb_data_dir, + get_config().common.mmdb_data_dir, MMDB_CITY_FILE_NAME ), locale: default_locale(), @@ -111,7 +111,7 @@ impl Default for GeoipConfig { impl GeoipConfig { pub fn new(name: &str) -> self::GeoipConfig { GeoipConfig { - path: format!("{}{}", &get_config().common.mmdb_data_dir, name), + path: format!("{}{}", get_config().common.mmdb_data_dir, name), locale: default_locale(), } } diff --git a/src/service/functions.rs b/src/service/functions.rs index 6a8a76d9e2..705977a7d7 100644 --- a/src/service/functions.rs +++ b/src/service/functions.rs @@ -432,7 +432,7 @@ pub async fn list_functions( || permitted .as_ref() .unwrap() - .contains(&format!("function:{}", &function.name)) + .contains(&format!("function:{}", function.name)) || permitted .as_ref() .unwrap() diff --git a/src/service/ingestion/mod.rs b/src/service/ingestion/mod.rs index c688021587..29b6b2f33d 100644 --- a/src/service/ingestion/mod.rs +++ b/src/service/ingestion/mod.rs @@ -317,10 +317,10 @@ pub async fn evaluate_trigger(triggers: TriggerAlertData) { if alert.trigger_condition.silence > 0 { log::debug!( "Realtime alert {}/{}/{}/{} triggered successfully, hence applying silence period", - &alert.org_id, - &alert.stream_type, - &alert.stream_name, - &alert.name + alert.org_id, + alert.stream_type, + alert.stream_name, + alert.name ); let next_run_at = Utc::now().timestamp_micros() diff --git a/src/service/metrics/otlp.rs b/src/service/metrics/otlp.rs index 136fa760b2..db78c09709 100644 --- a/src/service/metrics/otlp.rs +++ b/src/service/metrics/otlp.rs @@ -528,7 +528,7 @@ pub async fn handle_otlp_request( let need_trigger = !stream_trigger_map.contains_key(&local_metric_name); if need_trigger && !stream_alerts_map.is_empty() { // Start check for alert trigger - let key = format!("{}/{}/{}", &org_id, StreamType::Metrics, local_metric_name); + let key = format!("{}/{}/{}", org_id, StreamType::Metrics, local_metric_name); if let Some(alerts) = stream_alerts_map.get(&key) { let mut trigger_alerts: TriggerAlertData = Vec::new(); let alert_end_time = now_micros(); diff --git a/src/service/metrics/prom.rs b/src/service/metrics/prom.rs index ff61540118..c3f535ee52 100644 --- a/src/service/metrics/prom.rs +++ b/src/service/metrics/prom.rs @@ -586,7 +586,7 @@ pub async fn remote_write( let need_trigger = !stream_trigger_map.contains_key(&stream_name); if need_trigger && !stream_alerts_map.is_empty() { // Start check for alert trigger - let key = format!("{}/{}/{}", &org_id, StreamType::Metrics, stream_name); + let key = format!("{}/{}/{}", org_id, StreamType::Metrics, stream_name); if let Some(alerts) = stream_alerts_map.get(&key) { let mut trigger_alerts: TriggerAlertData = Vec::new(); let alert_end_time = now_micros(); @@ -1162,7 +1162,7 @@ async fn prom_ha_handler( "Electing {} new leader for {} as last received data from {} at {} ", replica_label, cluster_name, - &leader.name, + leader.name, Utc.timestamp_nanos(last_received * 1000) ); leader.name = replica_label.to_owned(); diff --git a/src/service/pipeline/mod.rs b/src/service/pipeline/mod.rs index 59b44e106b..1f7dff916c 100644 --- a/src/service/pipeline/mod.rs +++ b/src/service/pipeline/mod.rs @@ -223,7 +223,7 @@ pub async fn list_pipelines( || permitted .as_ref() .unwrap() - .contains(&format!("pipeline:{}", &pipeline.id)) + .contains(&format!("pipeline:{}", pipeline.id)) || permitted .as_ref() .unwrap() diff --git a/src/service/promql/engine.rs b/src/service/promql/engine.rs index 519668a7c3..f7049e6bbf 100644 --- a/src/service/promql/engine.rs +++ b/src/service/promql/engine.rs @@ -553,15 +553,11 @@ impl Engine { let start = std::time::Instant::now(); let mut metric_values = metrics.into_values().collect::>(); metric_values.par_iter_mut().for_each(|metric| { - metric - .samples - .sort_unstable_by(|a, b| a.timestamp.cmp(&b.timestamp)); - if self.ctx.query_ctx.query_exemplars && metric.exemplars.is_some() { - metric - .exemplars - .as_mut() - .unwrap() - .sort_by(|a, b| a.timestamp.cmp(&b.timestamp)); + metric.samples.sort_unstable_by_key(|k| k.timestamp); + if self.ctx.query_ctx.query_exemplars + && let Some(exemplars) = &mut metric.exemplars + { + exemplars.sort_by_key(|k| k.timestamp); } }); let values = if metric_values.is_empty() { diff --git a/src/service/promql/search/mod.rs b/src/service/promql/search/mod.rs index cdaed64368..558c9f7254 100644 --- a/src/service/promql/search/mod.rs +++ b/src/service/promql/search/mod.rs @@ -403,7 +403,7 @@ async fn search_in_cluster( log::info!( "[trace_id {trace_id}] promql->search->partition: node: {}, need_wal: {}, time_range: [{},{})", - &node.grpc_addr, + node.grpc_addr, req_need_wal, req_query.start, req_query.end, @@ -424,7 +424,7 @@ async fn search_in_cluster( Err(err) => { log::error!( "[trace_id {trace_id}] promql->search->grpc: node: {}, search err: {err:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), ); let err = ErrorCodes::from_json(err.message()) .unwrap_or(ErrorCodes::ServerInternalError(err.to_string())); @@ -435,7 +435,7 @@ async fn search_in_cluster( log::info!( "[trace_id {trace_id}] promql->search->grpc: result node: {}, need_wal: {req_need_wal}, files: {}, scan_size: {} mb, took: {} ms", - &node.get_grpc_addr(), + node.get_grpc_addr(), scan_stats.files, scan_stats.original_size, response.took, @@ -585,7 +585,7 @@ async fn merge_matrix_query(series: &[cluster_rpc::Series], org_id: &str) -> Res value: v, }) .collect::>(); - samples.sort_by(|a, b| a.timestamp.cmp(&b.timestamp)); + samples.sort_by_key(|k| k.timestamp); RangeValue::new(merged_metrics.get(&sig).unwrap().to_owned(), samples) }) .collect::>(); @@ -676,7 +676,7 @@ async fn merge_exemplars_query(series: &[cluster_rpc::Series], org_id: &str) -> .into_iter() .map(|(_ts, v)| Arc::new(v.into())) .collect::>(); - exemplars.sort_by(|a, b| a.timestamp.cmp(&b.timestamp)); + exemplars.sort_by_key(|k| k.timestamp); RangeValue::new_with_exemplars(merged_metrics.get(&sig).unwrap().to_owned(), exemplars) }) .collect::>(); diff --git a/src/service/runtime_metrics.rs b/src/service/runtime_metrics.rs index d1e7fa3c14..1d719189de 100644 --- a/src/service/runtime_metrics.rs +++ b/src/service/runtime_metrics.rs @@ -35,7 +35,7 @@ static PREV_COUNTERS: std::sync::LazyLock>> = std::sync::LazyLock::new(|| RwLock::new(HashMap::new())); pub fn register_runtime(name: String, handle: Handle) { - log::info!("Registered runtime '{}' for metrics collection", &name); + log::info!("Registered runtime '{}' for metrics collection", name); let mut handles = RUNTIME_HANDLES.lock().unwrap(); handles.push((name, handle)); } diff --git a/src/service/search/cache/cacher.rs b/src/service/search/cache/cacher.rs index e32dc74291..edd24869e6 100644 --- a/src/service/search/cache/cacher.rs +++ b/src/service/search/cache/cacher.rs @@ -963,7 +963,7 @@ fn calculate_deltas_multi( // sort the results by response end time descending let mut results = results.to_vec(); - results.sort_by(|a, b| b.response_end_time.cmp(&a.response_end_time)); + results.sort_by_key(|k| std::cmp::Reverse(k.response_end_time)); for meta in results { cache_duration += meta.response_end_time - meta.response_start_time; current_start_time = meta.response_start_time; diff --git a/src/service/search/cache/multi.rs b/src/service/search/cache/multi.rs index 23c9d414e2..1a6416c448 100644 --- a/src/service/search/cache/multi.rs +++ b/src/service/search/cache/multi.rs @@ -46,12 +46,10 @@ pub async fn get_cached_results( .result_cache_selection_strategy .as_str(), ); - loop { - let Some(idx) = cache_metas.iter().position_max_by_key(|result| { - select_cache_meta(result, &cache_req, &selection_strategy) - }) else { - break; - }; + while let Some(idx) = cache_metas + .iter() + .position_max_by_key(|result| select_cache_meta(result, &cache_req, &selection_strategy)) + { let best_cache_meta = cache_metas.remove(idx); if let Some(result) = super::cacher::get_cached_results( trace_id, diff --git a/src/service/search/cluster/cacher.rs b/src/service/search/cluster/cacher.rs index 487f8af578..6bc543fe86 100644 --- a/src/service/search/cluster/cacher.rs +++ b/src/service/search/cluster/cacher.rs @@ -75,7 +75,7 @@ pub async fn delete_cached_results(path: String, delete_ts: i64) -> bool { log::info!( "[trace_id {trace_id}] delete_cached_results->grpc: request node: {}", - &node_addr + node_addr ); let token: MetadataValue<_> = get_internal_grpc_token() @@ -84,7 +84,7 @@ pub async fn delete_cached_results(path: String, delete_ts: i64) -> bool { let channel = get_cached_channel(&node_addr).await.map_err(|err| { log::error!( "[trace_id {trace_id}] delete_cached_results->grpc: node: {}, connect err: {:?}", - &node.grpc_addr, + node.grpc_addr, err ); server_internal_error("connect search node error") @@ -108,7 +108,7 @@ pub async fn delete_cached_results(path: String, delete_ts: i64) -> bool { Err(err) => { log::error!( "[trace_id {trace_id}] delete_cached_results->grpc: node: {}, get_cached_results err: {:?}", - &node.grpc_addr, + node.grpc_addr, err ); let err = ErrorCodes::from_json(err.message())?; @@ -143,14 +143,14 @@ pub async fn delete_cached_results(path: String, delete_ts: i64) -> bool { true => { log::debug!( "[trace_id {trace_id}] delete_cached_results->grpc: node: {}, delete success", - &node.grpc_addr + node.grpc_addr ); } false => { delete_response = false; log::error!( "[trace_id {trace_id}] delete_cached_results->grpc: node delete error: node: {}", - &node.grpc_addr + node.grpc_addr ); } }, diff --git a/src/service/search/datafusion/plan/projections.rs b/src/service/search/datafusion/plan/projections.rs index ae5977053e..d44d0a3c0b 100644 --- a/src/service/search/datafusion/plan/projections.rs +++ b/src/service/search/datafusion/plan/projections.rs @@ -510,10 +510,8 @@ impl<'n> TreeNodeVisitor<'n> for ResultSchemaExtractor { self.timestamp_alias = Some(col.name.clone()) } } - Expr::ScalarFunction(_) => { - if is_ts_hist_udf(expr, &self.timestamp_alias) { - self.ts_hist_alias = Some(get_col_name(expr)); - } + Expr::ScalarFunction(_) if is_ts_hist_udf(expr, &self.timestamp_alias) => { + self.ts_hist_alias = Some(get_col_name(expr)); } _ => {} } diff --git a/src/service/search/datafusion/udf/transform_udf.rs b/src/service/search/datafusion/udf/transform_udf.rs index 05536a100e..5fa8bbd0f6 100644 --- a/src/service/search/datafusion/udf/transform_udf.rs +++ b/src/service/search/datafusion/udf/transform_udf.rs @@ -93,7 +93,7 @@ fn get_udf_vrl( col.value(i).replace("\"", "\\\"") )); } - obj_str.push_str(&format!(" \n {}", &local_func)); + obj_str.push_str(&format!(" \n {}", local_func)); match compile_vrl_function(&obj_str, &local_org_id) { Ok(res) => { let registry = res.config.get_custom::().unwrap(); diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index 0903c6bae8..d6c6ae3781 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -898,7 +898,7 @@ async fn search_tantivy_index( let need_all_term_fields = condition .need_all_term_fields() .into_iter() - .chain(get_simple_distinct_field(&idx_optimize_rule).into_iter()) + .chain(get_simple_distinct_field(&idx_optimize_rule)) .filter_map(|filed| tantivy_schema.get_field(&filed).ok()) .collect::>(); diff --git a/src/service/search/grpc/wal.rs b/src/service/search/grpc/wal.rs index 5480e05779..59522a1308 100644 --- a/src/service/search/grpc/wal.rs +++ b/src/service/search/grpc/wal.rs @@ -139,7 +139,7 @@ pub async fn search_parquet( if file.meta.min_ts > max_ts || file.meta.max_ts < min_ts { log::debug!( "[trace_id {trace_id}] skip wal parquet file: {} time_range: [{},{})", - &file.key, + file.key, file.meta.min_ts, file.meta.max_ts ); diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index 2e4c0175a4..f8f6cc224b 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.rs @@ -719,7 +719,7 @@ pub async fn query_status() -> Result { Err(err) => { log::error!( "[trace_id {trace_id}] search->grpc: node: {}, search err: {err:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), ); let err = ErrorCodes::from_json(err.message())?; return Err(Error::ErrorCode(err)); @@ -851,7 +851,7 @@ pub async fn cancel_query( Err(err) => { log::error!( "[trace_id {trace_id}] grpc_cancel_query: node: {}, search err: {err:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), ); let err = ErrorCodes::from_json(err.message())?; return Err(Error::ErrorCode(err)); diff --git a/src/service/search/sql/visitor/column.rs b/src/service/search/sql/visitor/column.rs index 88d2e726c4..d7c4eb256a 100644 --- a/src/service/search/sql/visitor/column.rs +++ b/src/service/search/sql/visitor/column.rs @@ -89,12 +89,11 @@ impl VisitorMut for ColumnVisitor<'_> { } } } - Expr::Function(f) => { + Expr::Function(f) if AGGREGATE_UDF_LIST - .contains(&trim_quotes(&f.name.to_string().to_lowercase()).as_str()) - { - self.has_agg_function = true; - } + .contains(&trim_quotes(&f.name.to_string().to_lowercase()).as_str()) => + { + self.has_agg_function = true; } _ => {} } diff --git a/src/service/search/streaming/cache.rs b/src/service/search/streaming/cache.rs index e78524a356..7225fed316 100644 --- a/src/service/search/streaming/cache.rs +++ b/src/service/search/streaming/cache.rs @@ -168,12 +168,12 @@ pub async fn handle_cache_responses_and_deltas( // sort both deltas and cache by order_by match cache_order_by { OrderBy::Desc => { - deltas.sort_by(|a, b| b.delta_start_time.cmp(&a.delta_start_time)); - cached_resp.sort_by(|a, b| b.response_start_time.cmp(&a.response_start_time)); + deltas.sort_by_key(|k| std::cmp::Reverse(k.delta_start_time)); + cached_resp.sort_by_key(|k| std::cmp::Reverse(k.response_start_time)); } OrderBy::Asc => { - deltas.sort_by(|a, b| a.delta_start_time.cmp(&b.delta_start_time)); - cached_resp.sort_by(|a, b| a.response_start_time.cmp(&b.response_start_time)); + deltas.sort_by_key(|k| k.delta_start_time); + cached_resp.sort_by_key(|k| k.response_start_time); } } diff --git a/src/service/search/streaming/execution.rs b/src/service/search/streaming/execution.rs index a66161c7c2..5ef8cb9e60 100644 --- a/src/service/search/streaming/execution.rs +++ b/src/service/search/streaming/execution.rs @@ -433,11 +433,10 @@ pub async fn do_partitioned_search( } // Remove the streaming_aggs cache - if is_streaming_aggs && partition_resp.streaming_id.is_some() { + if is_streaming_aggs && let Some(_streaming_id) = &partition_resp.streaming_id { #[cfg(feature = "enterprise")] { - let streaming_id = partition_resp.streaming_id.as_ref().unwrap(); - streaming_aggs_exec::remove_cache(streaming_id) + streaming_aggs_exec::remove_cache(_streaming_id) } } @@ -574,7 +573,7 @@ pub async fn process_delta( "[HTTP2_STREAM] Found {} partitions for trace_id: {}, partitions: {:?}", partitions.len(), trace_id, - &partitions + partitions ); // for dashboards & histograms, expect for ui @@ -790,9 +789,9 @@ pub async fn process_delta( } // Remove the streaming_aggs cache - if is_streaming_aggs && partition_resp.streaming_id.is_some() { + if is_streaming_aggs && let Some(_streaming_id) = partition_resp.streaming_id { #[cfg(feature = "enterprise")] - streaming_aggs_exec::remove_cache(&partition_resp.streaming_id.unwrap()) + streaming_aggs_exec::remove_cache(&_streaming_id) } Ok(()) diff --git a/src/service/search/streaming/sorting.rs b/src/service/search/streaming/sorting.rs index 3e30616229..88fd783262 100644 --- a/src/service/search/streaming/sorting.rs +++ b/src/service/search/streaming/sorting.rs @@ -47,7 +47,7 @@ fn determine_sort_strategy( if let Some(order_by) = search_res.order_by { log::info!( "[trace_id: {}] Using user-specified ORDER BY: {:?}", - &search_res.trace_id, + search_res.trace_id, order_by ); return SortStrategy::SqlOrderBy; @@ -63,7 +63,7 @@ fn determine_sort_strategy( if let Some((col, is_string)) = determine_sort_column(&search_res.hits[0]) { log::info!( "[trace_id: {}] Auto-sorting by column: {}, type: {}", - &search_res.trace_id, + search_res.trace_id, col, if is_string { "string" } else { "numeric" } ); @@ -119,7 +119,7 @@ fn find_fallback_column(search_res: &Response, fallback_col: Option) -> .map(|k| { log::info!( "[trace_id: {}] Using fallback ORDER BY: {}", - &search_res.trace_id, + search_res.trace_id, k ); k.to_string() diff --git a/src/service/search_jobs.rs b/src/service/search_jobs.rs index c5be71b243..709e89881b 100644 --- a/src/service/search_jobs.rs +++ b/src/service/search_jobs.rs @@ -257,9 +257,8 @@ async fn filter_partition_job( let mut needed_partitions_jobs = Vec::new(); for partition_job in partition_jobs.iter() { // if the result_path is not none, means the partition job is done - if partition_job.result_path.is_some() { - let path = partition_job.result_path.as_ref().unwrap(); - let buf = storage::get_bytes("", path).await?; + if let Some(result_path) = partition_job.result_path.as_ref() { + let buf = storage::get_bytes("", result_path).await?; let res: Response = json::from_str(String::from_utf8(buf.to_vec())?.as_str())?; need -= res.total as i64; } else { @@ -450,9 +449,8 @@ pub async fn get_result( Ok(res) => res.into_inner(), Err(err) => { log::error!( - "search->grpc: node: {}, search err: {:?}", - &node.get_grpc_addr(), - err + "search->grpc: node: {}, search err: {err:?}", + node.get_grpc_addr(), ); let err = ErrorCodes::from_json(err.message())?; return Err(Error::ErrorCode(err)); @@ -498,9 +496,8 @@ pub async fn delete_result(paths: Vec) -> Result<(), anyhow::Error> { Ok(res) => res.into_inner(), Err(err) => { log::error!( - "search->grpc: node: {}, search err: {:?}", - &node.get_grpc_addr(), - err + "search->grpc: node: {}, search err: {err}", + node.get_grpc_addr(), ); let err = ErrorCodes::from_json(err.message())?; return Err(Error::ErrorCode(err)); diff --git a/src/service/self_reporting/ingestion.rs b/src/service/self_reporting/ingestion.rs index fab05a0955..45e98bb5d6 100644 --- a/src/service/self_reporting/ingestion.rs +++ b/src/service/self_reporting/ingestion.rs @@ -128,7 +128,7 @@ pub(super) async fn ingest_usages(mut curr_usages: Vec) { let creds = if cfg.common.usage_reporting_creds.starts_with("Basic") { cfg.common.usage_reporting_creds.to_string() } else { - format!("Basic {}", &cfg.common.usage_reporting_creds) + format!("Basic {}", cfg.common.usage_reporting_creds) }; match reqwest::Client::new() .post(url) diff --git a/src/service/self_reporting/search.rs b/src/service/self_reporting/search.rs index ddd59f6f25..c302f6798a 100644 --- a/src/service/self_reporting/search.rs +++ b/src/service/self_reporting/search.rs @@ -112,7 +112,7 @@ pub async fn get_license_usage_data_from_node(node_id: String) -> Result { log::error!( "[trace_id: license-usage-grpc-request] error getting license usage info node {} : {err:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), ); let err = ErrorCodes::from_json(err.message())?; Err(anyhow::anyhow!("error getting license usage info : {err}")) diff --git a/src/service/sourcemaps/mod.rs b/src/service/sourcemaps/mod.rs index 120c80b37b..8d5efc8e91 100644 --- a/src/service/sourcemaps/mod.rs +++ b/src/service/sourcemaps/mod.rs @@ -321,7 +321,7 @@ async fn get_sourcemap_file_data( Err(err) => { log::error!( "[trace_id: {trace_id}] error getting sourcemap file from cluster {cluster} node {} for org_id {org} path {file_path} : {err:?}", - &node.get_grpc_addr(), + node.get_grpc_addr(), ); let err = ErrorCodes::from_json(err.message())?; Err(anyhow::anyhow!( diff --git a/src/service/stream.rs b/src/service/stream.rs index 12a9e21f45..6013332b2f 100644 --- a/src/service/stream.rs +++ b/src/service/stream.rs @@ -638,9 +638,6 @@ pub async fn update_stream_settings( stream_type: StreamType, mut new_settings: UpdateStreamSettings, ) -> Result { - let Ok(mut schema) = infra::schema::get(org_id, stream_name, stream_type).await else { - return Ok(MetaHttpResponse::not_found("stream not found")); - }; let Some(mut settings) = infra::schema::get_settings(org_id, stream_name, stream_type).await else { return Ok(MetaHttpResponse::not_found("stream not found")); @@ -659,20 +656,19 @@ pub async fn update_stream_settings( fields.push(arrow_schema::Field::new(name, data_type, true)); } let new_schema = Schema::new(fields); - schema = - match infra::schema::merge(org_id, stream_name, stream_type, &new_schema, None).await { - Ok(Some((s, _))) => s, - Ok(None) => { - return Err(Error::other( - "error in update stream settings: update schema is empty".to_string(), - )); - } - Err(e) => { - return Err(Error::other(format!( - "error in update stream settings: {e}" - ))); - } - }; + match infra::schema::merge(org_id, stream_name, stream_type, &new_schema, None).await { + Ok(Some(_)) => {} + Ok(None) => { + return Err(Error::other( + "error in update stream settings: update schema is empty".to_string(), + )); + } + Err(e) => { + return Err(Error::other(format!( + "error in update stream settings: {e}" + ))); + } + }; } if let Some(v) = new_settings.storage_type { diff --git a/src/service/tls/mod.rs b/src/service/tls/mod.rs index 6fc29a6139..cde03dcbc1 100644 --- a/src/service/tls/mod.rs +++ b/src/service/tls/mod.rs @@ -28,7 +28,7 @@ pub fn http_tls_config() -> Result { &mut BufReader::new(std::fs::File::open(&cfg.http.tls_cert_path).map_err(|e| { anyhow::anyhow!( "Failed to open TLS certificate file {}: {}", - &cfg.http.tls_cert_path, + cfg.http.tls_cert_path, e ) })?); @@ -36,7 +36,7 @@ pub fn http_tls_config() -> Result { &mut BufReader::new(std::fs::File::open(&cfg.http.tls_key_path).map_err(|e| { anyhow::anyhow!( "Failed to open TLS key file {}: {}", - &cfg.http.tls_key_path, + cfg.http.tls_key_path, e ) })?); @@ -81,7 +81,7 @@ pub fn client_tls_config() -> Result { std::fs::File::open(&cfg.http.tls_cert_path).map_err(|e| { anyhow::anyhow!( "Failed to open TLS certificate file {}: {}", - &cfg.http.tls_cert_path, + cfg.http.tls_cert_path, e ) })?, diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index 9a15e48b99..ad4da61556 100644 --- a/src/service/traces/mod.rs +++ b/src/service/traces/mod.rs @@ -260,7 +260,7 @@ pub async fn handle_otlp_request( log::warn!( "[TRACES:OTLP] Failed to ensure gen_ai schema fields for {}/{}: {e}", org_id, - &traces_stream_name + traces_stream_name ); } } @@ -757,7 +757,7 @@ pub async fn ingest_json( if timestamp < min_ts { log::error!( "[TRACES:JSON] skipping span with timestamp older than allowed retention period, trace_id: {}", - &trace_id + trace_id ); partial_success.rejected_spans += 1; continue; @@ -765,7 +765,7 @@ pub async fn ingest_json( if timestamp > max_ts { log::error!( "[TRACES:JSON] skipping span with timestamp newer than allowed retention period, trace_id: {}", - &trace_id + trace_id ); partial_success.rejected_spans += 1; continue; @@ -781,7 +781,7 @@ pub async fn ingest_json( _ => { log::error!( "[TRACES:JSON] stream did not receive a valid json object, trace_id: {}", - &trace_id + trace_id ); return Ok(( http::StatusCode::INTERNAL_SERVER_ERROR, diff --git a/src/service/traces/otel/extractors/input_output.rs b/src/service/traces/otel/extractors/input_output.rs index 688c22fd17..b72a2b8ee6 100644 --- a/src/service/traces/otel/extractors/input_output.rs +++ b/src/service/traces/otel/extractors/input_output.rs @@ -358,10 +358,9 @@ impl InputOutputExtractor { ) -> Option<(Option, Option)> { let events: Vec = if let Some(s) = events_val.as_str() { serde_json::from_str(s).ok()? - } else if let Some(arr) = events_val.as_array() { - arr.clone() } else { - return None; + let arr = events_val.as_array()?; + arr.clone() }; let choice_event = events diff --git a/src/service/traces/service_graph/api.rs b/src/service/traces/service_graph/api.rs index a2899fda96..c5b9c4cdd4 100644 --- a/src/service/traces/service_graph/api.rs +++ b/src/service/traces/service_graph/api.rs @@ -149,9 +149,8 @@ pub async fn get_current_topology( ); log::debug!( - "[ServiceGraph] Processing {} edge records for org '{}'", + "[ServiceGraph] Processing {} edge records for org '{org_id}'", edges.len(), - &org_id ); let (nodes, edges) = o2_enterprise::enterprise::service_graph::build_topology(edges, baselines); diff --git a/src/service/users.rs b/src/service/users.rs index 132332cd81..4feb06ccba 100644 --- a/src/service/users.rs +++ b/src/service/users.rs @@ -80,7 +80,7 @@ pub async fn post_user( } let cfg = get_config(); usr_req.email = usr_req.email.to_lowercase(); - if usr_req.role.custom_role.is_some() { + if let Some(_custom_roles) = &usr_req.role.custom_role { #[cfg(not(feature = "enterprise"))] return Ok(MetaHttpResponse::bad_request("Custom roles not allowed")); #[cfg(feature = "enterprise")] @@ -89,7 +89,7 @@ pub async fn post_user( } else { match o2_openfga::authorizer::roles::get_all_roles(org_id, None).await { Ok(res) => { - for custom_role in usr_req.role.custom_role.as_ref().unwrap() { + for custom_role in _custom_roles { if !res.contains(custom_role) { return Ok(MetaHttpResponse::bad_request("Custom role not found")); } @@ -200,9 +200,8 @@ pub async fn post_user( if usr_req.role.base_role.eq(&UserRole::ServiceAccount) { get_service_account_creation_tuple(&org_id, &usr_req.email, &mut tuples); } - if usr_req.role.custom_role.is_some() { - let custom_role = usr_req.role.custom_role.unwrap(); - custom_role.iter().for_each(|crole| { + if let Some(custom_roles) = usr_req.role.custom_role { + custom_roles.iter().for_each(|crole| { tuples.push(get_user_crole_tuple(&org_id, crole, &usr_req.email)); }); } @@ -356,17 +355,15 @@ pub async fn update_user( } new_user = local_user.clone(); if update_mode.is_self_update() - && user.old_password.is_some() - && user.new_password.is_some() + && let Some(old_pass) = &user.old_password + && let Some(new_pass) = &user.new_password { - if local_user.password.eq(&get_hash( - &user.clone().old_password.unwrap(), - &local_user.salt, - )) { - let new_pass = user.new_password.unwrap(); - - new_user.password = get_hash(&new_pass, &local_user.salt); - new_user.password_ext = Some(get_hash(&new_pass, password_ext_salt)); + if local_user + .password + .eq(&get_hash(old_pass, &local_user.salt)) + { + new_user.password = get_hash(new_pass, &local_user.salt); + new_user.password_ext = Some(get_hash(new_pass, password_ext_salt)); log::info!("Password self updated for user: {email}"); is_updated = true; } else { @@ -380,11 +377,9 @@ pub async fn update_user( message = "Please provide existing password"; } else if !update_mode.is_self_update() && allow_password_update - && user.new_password.is_some() && !local_user.is_external + && let Some(new_pass) = user.new_password { - let new_pass = user.new_password.unwrap(); - new_user.password = get_hash(&new_pass, &local_user.salt); new_user.password_ext = Some(get_hash(&new_pass, password_ext_salt)); log::info!("Password by root updated for user: {email}"); @@ -393,15 +388,19 @@ pub async fn update_user( } else if user.new_password.is_some() { message = "You are not authorised to change the password"; } - if user.first_name.is_some() && !local_user.is_external { - new_user.first_name = user.first_name.unwrap(); + if let Some(first_name) = user.first_name + && !local_user.is_external + { + new_user.first_name = first_name; is_updated = true; } - if user.last_name.is_some() && !local_user.is_external { - new_user.last_name = user.last_name.unwrap(); + if let Some(last_name) = user.last_name + && !local_user.is_external + { + new_user.last_name = last_name; is_updated = true; } - if user.role.is_some() + if let Some(role) = user.role && !local_user.is_external && (!update_mode.is_self_update() || (local_user.role.eq(&UserRole::Admin) @@ -412,7 +411,7 @@ pub async fn update_user( // if the User Role is Root, we do not change the Role // Admins Role can still be mutable. { - let new_org_role = UserOrgRole::from(&user.role.unwrap()); + let new_org_role = UserOrgRole::from(&role); old_role = Some(new_user.role); new_user.role = new_org_role.base_role; new_role = Some(new_user.role.clone()); @@ -433,8 +432,10 @@ pub async fn update_user( // Token replacement is a privileged operation — only allow if the // initiator is updating their own token OR has password-update rights // (i.e. Root or Admin updating a non-root user). - if user.token.is_some() && (update_mode.is_self_update() || allow_password_update) { - new_user.token = user.token.unwrap(); + if let Some(token) = user.token + && (update_mode.is_self_update() || allow_password_update) + { + new_user.token = token; is_org_updated = true; } @@ -686,9 +687,8 @@ pub async fn add_user_to_org( if get_openfga_config().enabled { let mut tuples = vec![]; get_add_user_to_org_tuples(org_id, email, &base_role.to_string(), &mut tuples); - if role.custom_role.is_some() { - let custom_role = role.custom_role.unwrap(); - custom_role.iter().for_each(|crole| { + if let Some(custom_roles) = role.custom_role { + custom_roles.iter().for_each(|crole| { tuples.push(get_user_crole_tuple(org_id, crole, email)); }); } @@ -917,7 +917,7 @@ pub async fn list_users( } } - user_list.sort_by(|a, b| b.created_at.cmp(&a.created_at)); + user_list.sort_by_key(|k| std::cmp::Reverse(k.created_at)); Ok(MetaHttpResponse::json(UserList { data: user_list })) } @@ -1010,7 +1010,7 @@ pub async fn remove_user_from_org( user_role.to_string() }; if get_openfga_config().enabled { - log::debug!("delete user single org, role: {}", &user_fga_role); + log::debug!("delete user single org, role: {user_fga_role}"); delete_user_from_org(org_id, email_id, &user_fga_role).await; if user_role.eq(&UserRole::ServiceAccount) { delete_service_account_from_org(org_id, email_id).await; diff --git a/src/tantivy_utils/src/puffin_directory/caching_directory.rs b/src/tantivy_utils/src/puffin_directory/caching_directory.rs index 56c3e68539..0cd83b1800 100644 --- a/src/tantivy_utils/src/puffin_directory/caching_directory.rs +++ b/src/tantivy_utils/src/puffin_directory/caching_directory.rs @@ -72,7 +72,7 @@ impl fmt::Debug for CachingFileHandle { write!( f, "CachingFileHandle(path={:?}, underlying={:?})", - &self.path, + self.path, self.underlying_filehandle.as_ref() ) } diff --git a/src/tantivy_utils/src/puffin_directory/reader.rs b/src/tantivy_utils/src/puffin_directory/reader.rs index efbc4c7ea4..81861b49f4 100644 --- a/src/tantivy_utils/src/puffin_directory/reader.rs +++ b/src/tantivy_utils/src/puffin_directory/reader.rs @@ -158,7 +158,7 @@ impl Directory for PuffinDirReader { Some(ext) => ext, None => return Ok(false), }; - let dir_path = format!("{}.{}", &EMPTY_PUFFIN_SEG_ID.as_str(), ext); + let dir_path = format!("{}.{}", EMPTY_PUFFIN_SEG_ID.as_str(), ext); EMPTY_PUFFIN_DIRECTORY.exists(&PathBuf::from(dir_path)) } else { Ok(true)