diff --git a/Cargo.lock b/Cargo.lock index b8172af3c..4f977a855 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -385,6 +385,15 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "asap-gorilla" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "thiserror 1.0.69", +] + [[package]] name = "asap-precompute-rs" version = "0.1.0" @@ -418,7 +427,7 @@ dependencies = [ "pretty_assertions", "promql-parser", "promql_utilities", - "reqwest", + "reqwest 0.11.27", "serde", "serde_json", "serde_yaml", @@ -527,6 +536,22 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" +[[package]] +name = "attohttpc" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16e2cdb6d5ed835199484bb92bb8b3edd526effe995c61732580439c1a67e2e9" +dependencies = [ + "base64 0.22.1", + "http 1.4.0", + "log", + "rustls 0.23.40", + "serde", + "serde_json", + "url", + "webpki-roots 1.0.7", +] + [[package]] name = "atty" version = "0.2.14" @@ -544,6 +569,54 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "aws-creds" +version = "0.39.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca3b85155d265df828f84e53886ed9e427aed979dd8a39f5b8b2162c77e142d7" +dependencies = [ + "attohttpc", + "home", + "log", + "quick-xml", + "rust-ini", + "serde", + "thiserror 2.0.18", + "time", + "url", +] + +[[package]] +name = "aws-lc-rs" +version = "1.16.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec6fb3fe69024a75fa7e1bfb48aa6cf59706a101658ea01bfd33b2b248a038f" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.40.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f50037ee5e1e41e7b8f9d161680a725bd1626cb6f8c7e901f91f942850852fe7" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", +] + +[[package]] +name = "aws-region" +version = "0.28.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "838b36c8dc927b6db1b6c6b8f5d05865f2213550b9e83bf92fa99ed6525472c0" +dependencies = [ + "thiserror 2.0.18", +] + [[package]] name = "axum" version = "0.7.9" @@ -762,6 +835,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + [[package]] name = "cfgrammar" version = "0.13.10" @@ -883,6 +962,15 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + [[package]] name = "colorchoice" version = "1.0.5" @@ -1534,6 +1622,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" dependencies = [ "powerfmt", + "serde_core", ] [[package]] @@ -1564,6 +1653,21 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "dlv-list" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "442039f5147480ba31067cb00ada1adae6892028e40e45fc5de7b7df6dcc1b5f" +dependencies = [ + "const-random", +] + +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "either" version = "1.15.0" @@ -1692,6 +1796,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futures" version = "0.3.32" @@ -1806,8 +1916,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -1817,9 +1929,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 5.3.0", "wasip2", + "wasm-bindgen", ] [[package]] @@ -1913,6 +2027,8 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ + "allocator-api2", + "equivalent", "foldhash", ] @@ -1967,6 +2083,24 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest", +] + +[[package]] +name = "home" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "http" version = "0.2.12" @@ -2095,9 +2229,25 @@ dependencies = [ "futures-util", "http 0.2.12", "hyper 0.14.32", - "rustls", + "rustls 0.21.12", "tokio", - "tokio-rustls", + "tokio-rustls 0.24.1", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" +dependencies = [ + "http 1.4.0", + "hyper 1.9.0", + "hyper-util", + "rustls 0.23.40", + "tokio", + "tokio-rustls 0.26.4", + "tower-service", + "webpki-roots 1.0.7", ] [[package]] @@ -2119,13 +2269,16 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ + "base64 0.22.1", "bytes", "futures-channel", "futures-util", "http 1.4.0", "http-body 1.0.1", "hyper 1.9.0", + "ipnet", "libc", + "percent-encoding", "pin-project-lite", "socket2 0.6.3", "tokio", @@ -2154,7 +2307,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core", + "windows-core 0.62.2", ] [[package]] @@ -2631,6 +2784,21 @@ dependencies = [ "vob", ] +[[package]] +name = "lru" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" +dependencies = [ + "hashbrown 0.15.5", +] + +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "lz4_flex" version = "0.11.6" @@ -2666,6 +2834,17 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" +[[package]] +name = "maybe-async" +version = "0.2.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5cf92c10c7e361d6b99666ec1c6f9805b0bea2c3bd8c78dc6fe98ac5bd78db11" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "md-5" version = "0.10.6" @@ -2676,6 +2855,12 @@ dependencies = [ "digest", ] +[[package]] +name = "md5" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae960838283323069879657ca3de837e9f7bbb4c7bf6ea7f1b290d5e9476d2e0" + [[package]] name = "memchr" version = "2.8.0" @@ -2741,6 +2926,15 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" +[[package]] +name = "ntapi" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3b335231dfd352ffb0f8017f3b6027a4917f7df785ea2143d8af2adc66980ae" +dependencies = [ + "winapi", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -2871,6 +3065,25 @@ dependencies = [ "libc", ] +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags 2.11.1", +] + +[[package]] +name = "objc2-io-kit" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33fafba39597d6dc1fb709123dfa8289d39406734be322956a69f0931c73bb15" +dependencies = [ + "libc", + "objc2-core-foundation", +] + [[package]] name = "object" version = "0.37.3" @@ -2928,6 +3141,16 @@ dependencies = [ "num-traits", ] +[[package]] +name = "ordered-multimap" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49203cdcae0030493bad186b28da2fa25645fa276a51b6fec8010d281e02ef79" +dependencies = [ + "dlv-list", + "hashbrown 0.14.5", +] + [[package]] name = "packedvec" version = "1.2.5" @@ -2958,7 +3181,7 @@ dependencies = [ "libc", "redox_syscall 0.5.18", "smallvec", - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -3420,6 +3643,7 @@ dependencies = [ "anyhow", "arc-swap", "arrow", + "asap-gorilla", "asap-precompute-rs", "asap_otel_proto", "asap_planner", @@ -3442,6 +3666,7 @@ dependencies = [ "futures", "hex", "lazy_static", + "lru", "memmap2", "moka", "prometheus", @@ -3450,9 +3675,10 @@ dependencies = [ "prost", "rdkafka", "regex", - "reqwest", + "reqwest 0.11.27", "rmp-serde", "rusqlite", + "rust-s3", "serde", "serde_json", "serde_yaml", @@ -3474,6 +3700,71 @@ dependencies = [ "zstd", ] +[[package]] +name = "quick-xml" +version = "0.38.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c" +dependencies = [ + "memchr", + "serde", +] + +[[package]] +name = "quinn" +version = "0.11.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e20a958963c291dc322d98411f541009df2ced7b5a4f2bd52337638cfccf20" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls 0.23.40", + "socket2 0.6.3", + "thiserror 2.0.18", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" +dependencies = [ + "bytes", + "getrandom 0.3.4", + "lru-slab", + "rand 0.9.4", + "ring", + "rustc-hash", + "rustls 0.23.40", + "rustls-pki-types", + "slab", + "thiserror 2.0.18", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2 0.6.3", + "tracing", + "windows-sys 0.52.0", +] + [[package]] name = "quote" version = "1.0.45" @@ -3686,7 +3977,7 @@ dependencies = [ "http 0.2.12", "http-body 0.4.6", "hyper 0.14.32", - "hyper-rustls", + "hyper-rustls 0.24.2", "ipnet", "js-sys", "log", @@ -3694,7 +3985,7 @@ dependencies = [ "once_cell", "percent-encoding", "pin-project-lite", - "rustls", + "rustls 0.21.12", "rustls-pemfile", "serde", "serde_json", @@ -3702,16 +3993,57 @@ dependencies = [ "sync_wrapper 0.1.2", "system-configuration", "tokio", - "tokio-rustls", + "tokio-rustls 0.24.1", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots", + "webpki-roots 0.25.4", "winreg", ] +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-core", + "futures-util", + "http 1.4.0", + "http-body 1.0.1", + "http-body-util", + "hyper 1.9.0", + "hyper-rustls 0.27.9", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls 0.23.40", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper 1.0.2", + "tokio", + "tokio-rustls 0.26.4", + "tokio-util", + "tower 0.5.3", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "wasm-streams", + "web-sys", + "webpki-roots 1.0.7", +] + [[package]] name = "ring" version = "0.17.14" @@ -3759,6 +4091,56 @@ dependencies = [ "smallvec", ] +[[package]] +name = "rust-ini" +version = "0.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "796e8d2b6696392a43bea58116b667fb4c29727dc5abd27d6acf338bb4f688c7" +dependencies = [ + "cfg-if", + "ordered-multimap", +] + +[[package]] +name = "rust-s3" +version = "0.37.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aeedb13abdaa7e48d391de05b0569b37fa0a7a64a668dff6ffb2141ad0c2527e" +dependencies = [ + "async-trait", + "aws-creds", + "aws-region", + "base64 0.22.1", + "bytes", + "cfg-if", + "futures-util", + "hex", + "hmac", + "http 1.4.0", + "log", + "maybe-async", + "md5", + "percent-encoding", + "quick-xml", + "reqwest 0.12.28", + "serde", + "serde_derive", + "serde_json", + "sha2", + "sysinfo", + "thiserror 2.0.18", + "time", + "tokio", + "tokio-stream", + "url", +] + +[[package]] +name = "rustc-hash" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" + [[package]] name = "rustc_version" version = "0.4.1" @@ -3789,10 +4171,26 @@ checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e" dependencies = [ "log", "ring", - "rustls-webpki", + "rustls-webpki 0.101.7", "sct", ] +[[package]] +name = "rustls" +version = "0.23.40" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" +dependencies = [ + "aws-lc-rs", + "log", + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki 0.103.13", + "subtle", + "zeroize", +] + [[package]] name = "rustls-pemfile" version = "1.0.4" @@ -3802,6 +4200,16 @@ dependencies = [ "base64 0.21.7", ] +[[package]] +name = "rustls-pki-types" +version = "1.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" +dependencies = [ + "web-time", + "zeroize", +] + [[package]] name = "rustls-webpki" version = "0.101.7" @@ -3812,6 +4220,18 @@ dependencies = [ "untrusted", ] +[[package]] +name = "rustls-webpki" +version = "0.103.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +dependencies = [ + "aws-lc-rs", + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.22" @@ -4237,6 +4657,9 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] [[package]] name = "synstructure" @@ -4249,6 +4672,20 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "sysinfo" +version = "0.37.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16607d5caffd1c07ce073528f9ed972d88db15dd44023fa57142963be3feb11f" +dependencies = [ + "libc", + "memchr", + "ntapi", + "objc2-core-foundation", + "objc2-io-kit", + "windows", +] + [[package]] name = "system-configuration" version = "0.5.1" @@ -4420,6 +4857,21 @@ dependencies = [ "serde_json", ] +[[package]] +name = "tinyvec" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e61e67053d25a4e82c844e8424039d9745781b3fc4f32b8d55ed50f5f667ef3" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.52.2" @@ -4454,7 +4906,17 @@ version = "0.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081" dependencies = [ - "rustls", + "rustls 0.21.12", + "tokio", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls 0.23.40", "tokio", ] @@ -4603,6 +5065,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a28f0d049ccfaa566e14e9663d304d8577427b368cb4710a20528690287a738b" +dependencies = [ + "bitflags 2.11.1", + "bytes", + "futures-util", + "http 1.4.0", + "http-body 1.0.1", + "pin-project-lite", + "tower 0.5.3", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -4969,6 +5449,19 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasmparser" version = "0.244.0" @@ -4991,12 +5484,31 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-roots" version = "0.25.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" +[[package]] +name = "webpki-roots" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52f5ee44c96cf55f1b349600768e3ece3a8f26010c05265ab73f945bb1a2eb9d" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "winapi" version = "0.3.9" @@ -5028,6 +5540,41 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windows" +version = "0.61.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893" +dependencies = [ + "windows-collections", + "windows-core 0.61.2", + "windows-future", + "windows-link 0.1.3", + "windows-numerics", +] + +[[package]] +name = "windows-collections" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8" +dependencies = [ + "windows-core 0.61.2", +] + +[[package]] +name = "windows-core" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link 0.1.3", + "windows-result 0.3.4", + "windows-strings 0.4.2", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -5036,9 +5583,20 @@ checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" dependencies = [ "windows-implement", "windows-interface", - "windows-link", - "windows-result", - "windows-strings", + "windows-link 0.2.1", + "windows-result 0.4.1", + "windows-strings 0.5.1", +] + +[[package]] +name = "windows-future" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" +dependencies = [ + "windows-core 0.61.2", + "windows-link 0.1.3", + "windows-threading", ] [[package]] @@ -5063,19 +5621,53 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "windows-link" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-numerics" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1" +dependencies = [ + "windows-core 0.61.2", + "windows-link 0.1.3", +] + +[[package]] +name = "windows-result" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows-result" version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" dependencies = [ - "windows-link", + "windows-link 0.2.1", +] + +[[package]] +name = "windows-strings" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" +dependencies = [ + "windows-link 0.1.3", ] [[package]] @@ -5084,7 +5676,7 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -5111,7 +5703,7 @@ version = "0.61.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" dependencies = [ - "windows-link", + "windows-link 0.2.1", ] [[package]] @@ -5145,6 +5737,15 @@ dependencies = [ "windows_x86_64_msvc 0.52.6", ] +[[package]] +name = "windows-threading" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6" +dependencies = [ + "windows-link 0.1.3", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.48.5" @@ -5448,6 +6049,12 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zeroize" +version = "1.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" + [[package]] name = "zerotrie" version = "0.2.4" diff --git a/asap-query-engine/Cargo.toml b/asap-query-engine/Cargo.toml index cfe6e59f8..08d53de78 100644 --- a/asap-query-engine/Cargo.toml +++ b/asap-query-engine/Cargo.toml @@ -86,6 +86,17 @@ asap-precompute-rs = { path = "../../ASAPCollector/asap-precompute-rs" } moka = { version = "0.12", features = ["sync"] } memmap2 = "0.9" crc32fast = "1.4" +# Phase 3 — Gorilla-S3 cold engine. `asap-gorilla` (path-dep, mirrors +# the `asap-precompute-rs` sibling-checkout pattern) provides the +# `GORILLA1` block decoder + per-hour `index.json` catalog. `rust-s3` +# (`s3` package on crates.io) is a single-crate S3-compatible client +# with first-class MinIO support; we hold it behind +# `default-features = false` + `tokio-rustls-tls` so it shares the +# rustls backend already pulled in by `reqwest`. `lru` powers the +# Phase 3 `ChunkCache` keyed on chunk object key. +asap-gorilla = { path = "../../ASAPCollector/asap-gorilla" } +s3 = { version = "0.37", package = "rust-s3", default-features = false, features = ["tokio-rustls-tls"] } +lru = "0.12" [[bin]] name = "precompute_engine" diff --git a/asap-query-engine/src/drivers/query/fallback/cold_store/gorilla_s3.rs b/asap-query-engine/src/drivers/query/fallback/cold_store/gorilla_s3.rs new file mode 100644 index 000000000..023686620 --- /dev/null +++ b/asap-query-engine/src/drivers/query/fallback/cold_store/gorilla_s3.rs @@ -0,0 +1,1000 @@ +//! Gorilla-on-S3 [`ColdStore`] adapter — Phase 3 of the +//! Gorilla-S3-cold-engine. +//! +//! Lists per-hour `index.json` catalogs out of an S3-compatible +//! bucket, prunes them by time range, then fetches + decodes the +//! selected `GORILLA1` chunks via the freshly-merged +//! [`asap_gorilla`] crate (`ASAPCollector` PR #281). +//! +//! Sits alongside [`super::LocalFsColdStore`] — both impls satisfy +//! the same [`super::ColdStore`] trait, so the existing +//! `s3_adapter::ColdFallback` query path can swap between them +//! without code change. The Phase 3 trait extension +//! ([`super::ColdStore::list_chunks`] / [`super::ColdStore::read_chunk`]) +//! lets the upcoming Phase 4 `GorillaQueryEngine` pull chunks one +//! at a time without materialising every sample. +//! +//! # Object key layout +//! +//! `GorillaS3ColdStore` is **agnostic** about the on-S3 chunk-key +//! shape. Two layouts are known to coexist (see PR #281): +//! +//! * design.md canonical: +//! `//YYYY/MM/DD/HH/part-NNNNNN.gor` +//! * Telegraf-side `gorilla_s3` output: +//! `/block---.gorilla` (random suffix) +//! +//! The per-hour `index.json` is the source of truth for what keys +//! exist; we treat [`asap_gorilla::IndexEntry::key`] as opaque and +//! do not try to parse it. The `prefix_template` config field +//! controls only where the **index** files live, not the chunks. +//! +//! # S3 client +//! +//! Backed by the `rust-s3` crate (`s3 = "0.37"`) — single-crate +//! dep, MinIO-friendly out of the box (no AWS-specific signing +//! quirks, supports custom endpoint URLs + path-style addressing). +//! Hidden behind the [`ObjectStore`] trait below so tests use an +//! in-memory mock and do not need a live MinIO. + +use std::num::NonZeroUsize; +use std::sync::Arc; + +#[cfg(test)] +use std::collections::HashMap; + +use async_trait::async_trait; +use chrono::{DateTime, Datelike, Timelike, Utc}; +use lru::LruCache; +use thiserror::Error; +use tokio::sync::Mutex; +use tracing::debug; + +use asap_gorilla::{GorillaDecoder, IndexFile}; + +use super::{ChunkRef, ColdStore, ColdStoreError, RawSample}; + +// ───────────────────────────────────────────────────────────────────── +// Public config +// ───────────────────────────────────────────────────────────────────── + +/// Tunable configuration for [`GorillaS3ColdStore`]. +/// +/// Use [`GorillaS3Config::from_env`] to pull values from environment +/// variables in deployment, or build manually for tests. +#[derive(Debug, Clone)] +pub struct GorillaS3Config { + /// `None` for AWS S3 (the SDK uses the standard regional + /// endpoint), `Some("http://minio:9000")` for MinIO / a custom + /// S3-compatible endpoint. + pub endpoint: Option, + /// Bucket name to list / read from. + pub bucket: String, + /// Tenant identifier prepended to every index-file prefix. + /// Empty string is allowed for single-tenant deployments. + pub tenant: String, + /// Prefix template for per-hour `index.json` files. Supports + /// the placeholders `{tenant}`, `{metric}`, `{year}`, `{month}`, + /// `{day}`, `{hour}` (zero-padded). Default: + /// `"{tenant}/{metric}/{year}/{month}/{day}/{hour}/"`. + pub prefix_template: String, + /// AWS-region the bucket lives in (e.g. `"us-east-1"`). For + /// MinIO any non-empty placeholder works. + pub region: String, + /// Optional static credential override. Both fields must be set + /// together; if either is `None` the underlying SDK falls back + /// to its environment / IMDS resolution. + pub access_key_id: Option, + /// See [`Self::access_key_id`]. + pub secret_access_key: Option, + /// LRU capacity (in number of decoded chunks). Default `256`. + pub cache_capacity: usize, + /// `false` switches the SDK to plain HTTP — required for local + /// MinIO / docker-compose smoke tests. Default `true`. + pub use_ssl: bool, +} + +impl Default for GorillaS3Config { + fn default() -> Self { + Self { + endpoint: None, + bucket: String::new(), + tenant: String::new(), + prefix_template: "{tenant}/{metric}/{year}/{month}/{day}/{hour}/".to_string(), + region: "us-east-1".to_string(), + access_key_id: None, + secret_access_key: None, + cache_capacity: 256, + use_ssl: true, + } + } +} + +/// Errors raised by [`GorillaS3Config::from_env`]. +#[derive(Debug, Error)] +pub enum GorillaS3ConfigError { + /// A required environment variable was missing. + #[error("missing required env var: {0}")] + MissingEnv(&'static str), + /// `ASAP_GORILLA_S3_CACHE_CAPACITY` could not be parsed as a + /// positive `usize`. + #[error("invalid env var {var}: {value} ({source})")] + InvalidEnv { + /// Variable name. + var: &'static str, + /// Raw value the user supplied. + value: String, + /// Underlying parse error. + source: std::num::ParseIntError, + }, +} + +impl GorillaS3Config { + /// Read a config from process environment variables. Required: + /// + /// * `ASAP_GORILLA_S3_BUCKET` + /// * `ASAP_GORILLA_S3_REGION` + /// + /// Optional (with defaults shown above): + /// + /// * `ASAP_GORILLA_S3_ENDPOINT` + /// * `ASAP_GORILLA_S3_TENANT` + /// * `ASAP_GORILLA_S3_PREFIX_TEMPLATE` + /// * `ASAP_GORILLA_S3_ACCESS_KEY_ID` / `..._SECRET_ACCESS_KEY` + /// * `ASAP_GORILLA_S3_CACHE_CAPACITY` + /// * `ASAP_GORILLA_S3_USE_SSL` (`"true"` / `"false"`, + /// case-insensitive) + pub fn from_env() -> Result { + let bucket = std::env::var("ASAP_GORILLA_S3_BUCKET") + .map_err(|_| GorillaS3ConfigError::MissingEnv("ASAP_GORILLA_S3_BUCKET"))?; + let region = std::env::var("ASAP_GORILLA_S3_REGION") + .map_err(|_| GorillaS3ConfigError::MissingEnv("ASAP_GORILLA_S3_REGION"))?; + let endpoint = std::env::var("ASAP_GORILLA_S3_ENDPOINT").ok(); + let tenant = std::env::var("ASAP_GORILLA_S3_TENANT").unwrap_or_default(); + let prefix_template = std::env::var("ASAP_GORILLA_S3_PREFIX_TEMPLATE") + .unwrap_or_else(|_| "{tenant}/{metric}/{year}/{month}/{day}/{hour}/".to_string()); + let access_key_id = std::env::var("ASAP_GORILLA_S3_ACCESS_KEY_ID").ok(); + let secret_access_key = std::env::var("ASAP_GORILLA_S3_SECRET_ACCESS_KEY").ok(); + let cache_capacity = match std::env::var("ASAP_GORILLA_S3_CACHE_CAPACITY") { + Ok(s) => s + .parse::() + .map_err(|e| GorillaS3ConfigError::InvalidEnv { + var: "ASAP_GORILLA_S3_CACHE_CAPACITY", + value: s, + source: e, + })?, + Err(_) => 256, + }; + let use_ssl = std::env::var("ASAP_GORILLA_S3_USE_SSL") + .map(|s| !matches!(s.trim().to_ascii_lowercase().as_str(), "false" | "0" | "no")) + .unwrap_or(true); + Ok(Self { + endpoint, + bucket, + tenant, + prefix_template, + region, + access_key_id, + secret_access_key, + cache_capacity, + use_ssl, + }) + } +} + +// ───────────────────────────────────────────────────────────────────── +// ObjectStore — internal trait so tests don't need real S3 +// ───────────────────────────────────────────────────────────────────── + +/// Minimal async object-fetch interface. +/// +/// Sized + `Send + Sync` so [`GorillaS3ColdStore`] can hold one +/// behind an `Arc` regardless of how it's backed. +/// Production callers use [`S3ObjectStore`] (rust-s3); tests use the +/// in-memory mock at the bottom of this file. +#[async_trait] +pub trait ObjectStore: Send + Sync { + /// Fetch the full object body for `key`. + /// + /// Returns [`ColdStoreError::Backend`] for transport errors and + /// [`ColdStoreError::Backend`] (with a `not found` substring) + /// for missing keys; callers distinguish via + /// [`ObjectStore::object_missing`] if they need to. + async fn get_object(&self, key: &str) -> Result, ColdStoreError>; + + /// True iff `err` was raised because the requested key did not + /// exist (vs. a transport / permission failure). Used by the + /// list path to treat a missing `index.json` as "no chunks for + /// this hour" rather than a hard error. + fn object_missing(&self, err: &ColdStoreError) -> bool { + matches!(err, ColdStoreError::Backend(msg) if msg.contains("not found")) + } +} + +// ───────────────────────────────────────────────────────────────────── +// rust-s3 backed production ObjectStore +// ───────────────────────────────────────────────────────────────────── + +mod rust_s3_backend { + use super::*; + use s3::creds::Credentials; + use s3::region::Region as S3Region; + use s3::Bucket; + + /// `rust-s3`-backed [`ObjectStore`]. Default production choice. + pub struct S3ObjectStore { + bucket: Box, + } + + impl S3ObjectStore { + /// Build from a [`GorillaS3Config`]. Sets + /// `path_style = true` whenever a custom endpoint is + /// configured (MinIO mandates path-style addressing). + pub fn new(cfg: &GorillaS3Config) -> Result { + let region = match &cfg.endpoint { + Some(ep) => { + let endpoint = if ep.starts_with("http://") || ep.starts_with("https://") { + ep.clone() + } else if cfg.use_ssl { + format!("https://{}", ep) + } else { + format!("http://{}", ep) + }; + S3Region::Custom { + region: cfg.region.clone(), + endpoint, + } + } + None => cfg + .region + .parse::() + .map_err(|e| ColdStoreError::Backend(format!("region parse: {e}")))?, + }; + let creds = match (&cfg.access_key_id, &cfg.secret_access_key) { + (Some(ak), Some(sk)) => { + Credentials::new(Some(ak), Some(sk), None, None, None).map_err(|e| { + ColdStoreError::Backend(format!("credentials: {e}")) + })? + } + _ => Credentials::default().map_err(|e| { + ColdStoreError::Backend(format!("default credentials: {e}")) + })?, + }; + let bucket = Bucket::new(&cfg.bucket, region, creds) + .map_err(|e| ColdStoreError::Backend(format!("bucket: {e}")))?; + // MinIO + most S3-compatibles require path-style addressing + // when a custom endpoint is in play. AWS S3 supports both, + // so leaving it on for the AWS path is safe but slightly + // less efficient — only flip when an endpoint is set. + let bucket = if cfg.endpoint.is_some() { + bucket.with_path_style() + } else { + bucket + }; + Ok(Self { bucket }) + } + } + + #[async_trait] + impl ObjectStore for S3ObjectStore { + async fn get_object(&self, key: &str) -> Result, ColdStoreError> { + let resp = self + .bucket + .get_object(key) + .await + .map_err(|e| ColdStoreError::Backend(format!("s3 get {key}: {e}")))?; + if resp.status_code() == 404 { + return Err(ColdStoreError::Backend(format!( + "s3 get {key}: not found" + ))); + } + if !(200..300).contains(&resp.status_code()) { + return Err(ColdStoreError::Backend(format!( + "s3 get {key}: status {}", + resp.status_code() + ))); + } + Ok(resp.to_vec()) + } + } +} + +pub use rust_s3_backend::S3ObjectStore; + +// ───────────────────────────────────────────────────────────────────── +// GorillaS3ColdStore +// ───────────────────────────────────────────────────────────────────── + +/// LRU cache keyed by chunk object key. Stored values are +/// pre-decoded `RawSample` lists so repeated reads of the same +/// chunk skip the Gorilla decode pass entirely. +type ChunkCache = Mutex>>>; + +/// `ColdStore` adapter that reads `GORILLA1`-format chunks out of +/// an S3-compatible bucket. See module docs for layout + S3 client +/// notes. +pub struct GorillaS3ColdStore { + object_store: Arc, + config: GorillaS3Config, + cache: ChunkCache, +} + +impl GorillaS3ColdStore { + /// Build with an explicit object-store backend. The production + /// constructor [`Self::with_default_backend`] wires up + /// `S3ObjectStore` from `cfg`; tests inject the in-memory mock. + pub fn new(object_store: Arc, config: GorillaS3Config) -> Self { + let cap = NonZeroUsize::new(config.cache_capacity.max(1)) + .unwrap_or(NonZeroUsize::new(1).unwrap()); + Self { + object_store, + config, + cache: Mutex::new(LruCache::new(cap)), + } + } + + /// Build from a [`GorillaS3Config`] using the default + /// `rust-s3`-backed [`ObjectStore`]. + pub fn with_default_backend(config: GorillaS3Config) -> Result { + let backend = Arc::new(S3ObjectStore::new(&config)?); + Ok(Self::new(backend, config)) + } + + /// Borrow the active config — useful for diagnostics. + pub fn config(&self) -> &GorillaS3Config { + &self.config + } + + /// Render the configured `prefix_template` for one + /// `(metric, hour)` bucket and append `index.json`. + fn index_key(&self, metric: &str, ts_ms: i64) -> String { + let dt: DateTime = DateTime::::from_timestamp_millis(ts_ms) + .unwrap_or_else(|| DateTime::::from_timestamp(0, 0).unwrap()); + let prefix = self + .config + .prefix_template + .replace("{tenant}", &self.config.tenant) + .replace("{metric}", metric) + .replace("{year}", &format!("{:04}", dt.year())) + .replace("{month}", &format!("{:02}", dt.month())) + .replace("{day}", &format!("{:02}", dt.day())) + .replace("{hour}", &format!("{:02}", dt.hour())); + let mut key = prefix; + if !key.ends_with('/') { + key.push('/'); + } + key.push_str("index.json"); + key + } + + /// Iterate the wall-clock-hour starts (in ms) covered by + /// `[start_ms, end_ms)`. Always emits at least one bucket. + fn hour_starts(start_ms: i64, end_ms: i64) -> Vec { + const HOUR_MS: i64 = 3_600_000; + if end_ms <= start_ms { + let h = (start_ms / HOUR_MS) * HOUR_MS; + return vec![h]; + } + let first = (start_ms / HOUR_MS) * HOUR_MS; + let last = ((end_ms - 1) / HOUR_MS) * HOUR_MS; + let mut out = Vec::new(); + let mut cur = first; + while cur <= last { + out.push(cur); + cur += HOUR_MS; + } + out + } + + /// Fetch + parse one hour's `index.json`. Missing index = empty + /// catalog (the producer may not have flushed yet); transport + /// failure surfaces as `ColdStoreError::Backend`. + async fn fetch_index(&self, metric: &str, hour_ms: i64) -> Result { + let key = self.index_key(metric, hour_ms); + match self.object_store.get_object(&key).await { + Ok(bytes) => IndexFile::read(bytes.as_slice()).map_err(|e| { + ColdStoreError::Malformed(format!("index.json at {key}: {e}")) + }), + Err(e) if self.object_store.object_missing(&e) => { + debug!(key = %key, "gorilla-s3: index.json missing for hour bucket; skipping"); + Ok(IndexFile::new(0)) + } + Err(e) => Err(e), + } + } +} + +#[async_trait] +impl ColdStore for GorillaS3ColdStore { + async fn scan( + &self, + metric: &str, + start_ms: i64, + end_ms: i64, + ) -> Result, ColdStoreError> { + let chunks = self.list_chunks(metric, start_ms, end_ms).await?; + let mut out = Vec::new(); + for chunk in chunks { + let samples = self.read_chunk(&chunk).await?; + for s in samples { + if s.ts_ms >= start_ms && s.ts_ms < end_ms { + out.push(s); + } + } + } + Ok(out) + } + + async fn list_chunks( + &self, + metric: &str, + start_ms: i64, + end_ms: i64, + ) -> Result, ColdStoreError> { + // Convert the request window to the nanosecond unit the + // index file uses (`IndexEntry.time_range` is `(ns, ns)`, + // mirroring the Go encoder's `time.Time.UnixNano()` source). + let start_ns = (start_ms as i128).saturating_mul(1_000_000) as u64; + // `end_ms` is exclusive on the ms side; the index iter + // overlap test is inclusive so subtract 1 ns to keep the + // semantics aligned. If `end_ms == start_ms` we still want + // to scan the bucket containing `start_ms`. + let end_ns = if end_ms <= start_ms { + start_ns + } else { + ((end_ms as i128).saturating_mul(1_000_000) - 1).max(0) as u64 + }; + + let mut out = Vec::new(); + for hour_ms in Self::hour_starts(start_ms, end_ms) { + let idx = self.fetch_index(metric, hour_ms).await?; + for entry in idx.prune_by_time((start_ns, end_ns)) { + let (entry_start_ms, entry_end_ms) = ( + (entry.time_range.0 / 1_000_000) as i64, + (entry.time_range.1 / 1_000_000) as i64, + ); + out.push(ChunkRef { + key: entry.key.clone(), + metric: metric.to_string(), + time_range_ms: (entry_start_ms, entry_end_ms), + label_hash: entry.label_hash, + sample_count: entry.sample_count, + size_bytes: entry.size_bytes, + }); + } + } + Ok(out) + } + + async fn read_chunk(&self, chunk: &ChunkRef) -> Result, ColdStoreError> { + // Cache hit fast path. + { + let mut guard = self.cache.lock().await; + if let Some(cached) = guard.get(&chunk.key).cloned() { + return Ok((*cached).clone()); + } + } + + let bytes = self.object_store.get_object(&chunk.key).await?; + let samples = decode_block(&bytes) + .map_err(|e| ColdStoreError::Malformed(format!("decode {}: {e}", chunk.key)))?; + + let arc = Arc::new(samples.clone()); + { + let mut guard = self.cache.lock().await; + guard.put(chunk.key.clone(), arc); + } + Ok(samples) + } +} + +// ───────────────────────────────────────────────────────────────────── +// Decoding helper — converts a GORILLA1 block to RawSample units +// ───────────────────────────────────────────────────────────────────── + +/// Decode a single in-memory `GORILLA1` block into [`RawSample`]s. +/// +/// Walks every series in the block; multi-series blocks are +/// flattened into one `Vec`. Timestamps are converted from the +/// on-wire nanoseconds (Go `time.Time.UnixNano()` source) to the +/// [`RawSample::ts_ms`] millisecond unit. +fn decode_block(bytes: &[u8]) -> Result, asap_gorilla::DecodeError> { + let mut decoder = GorillaDecoder::from_reader(bytes)?; + let mut out: Vec = Vec::new(); + while let Some(header) = decoder.header().cloned() { + let labels: std::collections::BTreeMap = + header.labels.iter().cloned().collect(); + for sample in decoder.samples() { + let (ts_ns, value) = sample?; + out.push(RawSample { + ts_ms: (ts_ns / 1_000_000) as i64, + labels: labels.clone(), + value, + }); + } + if !decoder.next_series()? { + break; + } + } + Ok(out) +} + +// ───────────────────────────────────────────────────────────────────── +// In-memory ObjectStore mock — pub(crate) so tests in sibling files +// can exercise the same fixture without a live MinIO. +// ───────────────────────────────────────────────────────────────────── + +/// In-memory [`ObjectStore`] used by `gorilla_s3` tests. +/// +/// Holds a `HashMap>` plus a per-key fetch counter so +/// cache-hit assertions are first-class. Optionally fails every +/// `get_object` call for the network-error test. +#[cfg(test)] +#[derive(Default)] +pub(crate) struct InMemoryObjectStore { + inner: Mutex, +} + +#[cfg(test)] +#[derive(Default)] +struct InMemoryState { + objects: HashMap>, + fetch_counts: HashMap, + fail_all: Option, +} + +#[cfg(test)] +impl InMemoryObjectStore { + pub(crate) fn new() -> Self { + Self::default() + } + + pub(crate) async fn put(&self, key: impl Into, body: Vec) { + let mut g = self.inner.lock().await; + g.objects.insert(key.into(), body); + } + + pub(crate) async fn get_count(&self, key: &str) -> usize { + let g = self.inner.lock().await; + g.fetch_counts.get(key).copied().unwrap_or(0) + } + + /// Make every subsequent `get_object` fail with a backend error + /// containing `msg`. Used by the network-error test. + pub(crate) async fn fail_all(&self, msg: impl Into) { + let mut g = self.inner.lock().await; + g.fail_all = Some(msg.into()); + } +} + +#[cfg(test)] +#[async_trait] +impl ObjectStore for InMemoryObjectStore { + async fn get_object(&self, key: &str) -> Result, ColdStoreError> { + let mut g = self.inner.lock().await; + if let Some(msg) = g.fail_all.clone() { + return Err(ColdStoreError::Backend(msg)); + } + *g.fetch_counts.entry(key.to_string()).or_insert(0) += 1; + match g.objects.get(key) { + Some(b) => Ok(b.clone()), + None => Err(ColdStoreError::Backend(format!("get {key}: not found"))), + } + } +} + +// Static `Send` assertion — `GorillaS3ColdStore` must be storable +// behind an `Arc` in the existing s3_adapter chain. +const _: fn() = || { + fn _assert_send() {} + _assert_send::(); +}; + +// ───────────────────────────────────────────────────────────────────── +// Tests +// ───────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + + use asap_gorilla::{GorillaEncoder, IndexEntry, IndexFile}; + use chrono::TimeZone; + + /// Build a minimal index.json fixture. + fn make_index(entries: Vec) -> Vec { + let mut idx = IndexFile::new(0); + idx.entries = entries; + let mut buf = Vec::new(); + idx.write(&mut buf).unwrap(); + buf + } + + /// Encode a single-series Gorilla block from `(ts_ms, value)` pairs. + fn make_block(metric: &str, labels: &[(&str, &str)], samples: &[(i64, f64)]) -> Vec { + let mut enc = GorillaEncoder::new( + metric.to_string(), + labels + .iter() + .map(|(k, v)| ((*k).to_string(), (*v).to_string())) + .collect(), + ); + for (ts_ms, v) in samples { + // ts_ms → ts_ns + enc.append((*ts_ms as u64) * 1_000_000, *v); + } + enc.finalize().unwrap() + } + + fn ms(year: i32, month: u32, day: u32, hour: u32, min: u32, sec: u32) -> i64 { + Utc.with_ymd_and_hms(year, month, day, hour, min, sec) + .unwrap() + .timestamp_millis() + } + + fn cfg() -> GorillaS3Config { + GorillaS3Config { + endpoint: Some("http://mock".to_string()), + bucket: "test-bucket".to_string(), + tenant: "tenant1".to_string(), + prefix_template: "{tenant}/{metric}/{year}/{month}/{day}/{hour}/".to_string(), + region: "us-east-1".to_string(), + access_key_id: None, + secret_access_key: None, + cache_capacity: 4, + use_ssl: false, + } + } + + /// Layout: hour bucket H, three chunks A/B/C in time order, the + /// requested window only overlaps B → list returns B alone. + #[tokio::test] + async fn list_chunks_via_indexfile_prunes_by_time() { + let store = InMemoryObjectStore::new(); + + let h0 = ms(2026, 5, 6, 12, 0, 0); + let metric = "node_cpu_seconds_total"; + + let key_a = "tenant1/node_cpu_seconds_total/2026/05/06/12/part-A.gor".to_string(); + let key_b = "tenant1/node_cpu_seconds_total/2026/05/06/12/part-B.gor".to_string(); + let key_c = "tenant1/node_cpu_seconds_total/2026/05/06/12/part-C.gor".to_string(); + + let entries = vec![ + IndexEntry { + key: key_a.clone(), + time_range: ((h0) as u64 * 1_000_000, (h0 + 999) as u64 * 1_000_000), + sample_count: 10, + label_hash: 0xAAAA, + size_bytes: 100, + }, + IndexEntry { + key: key_b.clone(), + time_range: ((h0 + 5_000) as u64 * 1_000_000, (h0 + 6_000) as u64 * 1_000_000), + sample_count: 11, + label_hash: 0xBBBB, + size_bytes: 110, + }, + IndexEntry { + key: key_c.clone(), + time_range: ((h0 + 10_000) as u64 * 1_000_000, (h0 + 11_000) as u64 * 1_000_000), + sample_count: 12, + label_hash: 0xCCCC, + size_bytes: 120, + }, + ]; + + store + .put( + "tenant1/node_cpu_seconds_total/2026/05/06/12/index.json", + make_index(entries), + ) + .await; + + let cs = GorillaS3ColdStore::new(Arc::new(store), cfg()); + let chunks = cs + .list_chunks(metric, h0 + 5_500, h0 + 5_800) + .await + .unwrap(); + + assert_eq!(chunks.len(), 1); + assert_eq!(chunks[0].key, key_b); + assert_eq!(chunks[0].sample_count, 11); + assert_eq!(chunks[0].label_hash, 0xBBBB); + assert_eq!(chunks[0].metric, metric); + } + + #[tokio::test] + async fn read_chunk_decodes_via_asap_gorilla() { + let store = InMemoryObjectStore::new(); + + let h0 = ms(2026, 5, 6, 12, 0, 0); + let metric = "node_cpu_seconds_total"; + let labels = &[("instance", "i-1"), ("mode", "user")]; + + let block = make_block( + metric, + labels, + &[(h0 + 1_000, 0.5), (h0 + 2_000, 0.7), (h0 + 3_000, 0.7)], + ); + let chunk_key = "tenant1/node_cpu_seconds_total/2026/05/06/12/part-001.gor".to_string(); + + store.put(chunk_key.clone(), block.clone()).await; + store + .put( + "tenant1/node_cpu_seconds_total/2026/05/06/12/index.json", + make_index(vec![IndexEntry { + key: chunk_key.clone(), + time_range: ( + (h0 + 1_000) as u64 * 1_000_000, + (h0 + 3_000) as u64 * 1_000_000, + ), + sample_count: 3, + label_hash: 0x1234, + size_bytes: block.len() as u32, + }]), + ) + .await; + + let cs = GorillaS3ColdStore::new(Arc::new(store), cfg()); + let chunks = cs.list_chunks(metric, h0, h0 + 60_000).await.unwrap(); + assert_eq!(chunks.len(), 1); + + let samples = cs.read_chunk(&chunks[0]).await.unwrap(); + assert_eq!(samples.len(), 3); + assert_eq!(samples[0].ts_ms, h0 + 1_000); + assert_eq!(samples[0].value, 0.5); + assert_eq!(samples[1].ts_ms, h0 + 2_000); + assert_eq!(samples[1].value, 0.7); + assert_eq!(samples[2].ts_ms, h0 + 3_000); + assert_eq!(samples[2].value, 0.7); + assert_eq!(samples[0].labels.get("instance").map(String::as_str), Some("i-1")); + assert_eq!(samples[0].labels.get("mode").map(String::as_str), Some("user")); + } + + #[tokio::test] + async fn cache_hit_skips_s3_fetch() { + let store = Arc::new(InMemoryObjectStore::new()); + + let h0 = ms(2026, 5, 6, 12, 0, 0); + let metric = "m"; + let block = make_block(metric, &[], &[(h0 + 1_000, 1.0), (h0 + 2_000, 2.0)]); + let chunk_key = "tenant1/m/2026/05/06/12/part-X.gor".to_string(); + store.put(chunk_key.clone(), block.clone()).await; + store + .put( + "tenant1/m/2026/05/06/12/index.json", + make_index(vec![IndexEntry { + key: chunk_key.clone(), + time_range: ( + (h0 + 1_000) as u64 * 1_000_000, + (h0 + 2_000) as u64 * 1_000_000, + ), + sample_count: 2, + label_hash: 0, + size_bytes: block.len() as u32, + }]), + ) + .await; + + let cs = GorillaS3ColdStore::new(store.clone(), cfg()); + let chunks = cs.list_chunks(metric, h0, h0 + 60_000).await.unwrap(); + assert_eq!(chunks.len(), 1); + + let _ = cs.read_chunk(&chunks[0]).await.unwrap(); + let count_after_first = store.get_count(&chunk_key).await; + let _ = cs.read_chunk(&chunks[0]).await.unwrap(); + let count_after_second = store.get_count(&chunk_key).await; + + assert_eq!(count_after_first, 1); + assert_eq!( + count_after_second, 1, + "second read_chunk must hit cache and skip S3 GET" + ); + } + + #[tokio::test] + async fn lru_eviction_under_pressure() { + // cache_capacity=2, fill with three chunks then re-read the + // first → that triggers an S3 GET because the LRU evicted it. + let store = Arc::new(InMemoryObjectStore::new()); + let h0 = ms(2026, 5, 6, 12, 0, 0); + + let mut chunk_refs: Vec = Vec::new(); + let mut entries: Vec = Vec::new(); + for i in 0..3i64 { + let block = make_block( + "m", + &[("i", &i.to_string())], + &[(h0 + i * 1_000, i as f64), (h0 + i * 1_000 + 100, i as f64 + 0.5)], + ); + let key = format!("tenant1/m/2026/05/06/12/part-{i}.gor"); + store.put(key.clone(), block.clone()).await; + entries.push(IndexEntry { + key: key.clone(), + time_range: ( + ((h0 + i * 1_000) as u64) * 1_000_000, + ((h0 + i * 1_000 + 100) as u64) * 1_000_000, + ), + sample_count: 2, + label_hash: i as u64, + size_bytes: block.len() as u32, + }); + chunk_refs.push(ChunkRef { + key, + metric: "m".to_string(), + time_range_ms: (h0 + i * 1_000, h0 + i * 1_000 + 100), + label_hash: i as u64, + sample_count: 2, + size_bytes: block.len() as u32, + }); + } + store + .put("tenant1/m/2026/05/06/12/index.json", make_index(entries)) + .await; + + let mut config = cfg(); + config.cache_capacity = 2; + let cs = GorillaS3ColdStore::new(store.clone(), config); + + cs.read_chunk(&chunk_refs[0]).await.unwrap(); + cs.read_chunk(&chunk_refs[1]).await.unwrap(); + cs.read_chunk(&chunk_refs[2]).await.unwrap(); // evicts chunk_refs[0] + + let before = store.get_count(&chunk_refs[0].key).await; + cs.read_chunk(&chunk_refs[0]).await.unwrap(); + let after = store.get_count(&chunk_refs[0].key).await; + assert_eq!( + after, + before + 1, + "evicted chunk must trigger a fresh S3 GET" + ); + } + + #[tokio::test] + async fn index_json_corrupted_returns_error() { + let store = InMemoryObjectStore::new(); + let h0 = ms(2026, 5, 6, 12, 0, 0); + store + .put( + "tenant1/m/2026/05/06/12/index.json", + b"this is not json {{{".to_vec(), + ) + .await; + + let cs = GorillaS3ColdStore::new(Arc::new(store), cfg()); + let res = cs.list_chunks("m", h0, h0 + 60_000).await; + match res { + Err(ColdStoreError::Malformed(msg)) => { + assert!(msg.contains("index.json"), "msg should name the key: {msg}") + } + other => panic!("expected Malformed, got {other:?}"), + } + } + + #[tokio::test] + async fn s3_unavailable_returns_error() { + let store = Arc::new(InMemoryObjectStore::new()); + store.fail_all("simulated network outage").await; + let cs = GorillaS3ColdStore::new(store, cfg()); + let h0 = ms(2026, 5, 6, 12, 0, 0); + let res = cs.list_chunks("m", h0, h0 + 60_000).await; + match res { + Err(ColdStoreError::Backend(msg)) => assert!(msg.contains("simulated network outage")), + other => panic!("expected Backend, got {other:?}"), + } + } + + #[tokio::test] + async fn missing_index_is_empty_not_error() { + let store = InMemoryObjectStore::new(); + let cs = GorillaS3ColdStore::new(Arc::new(store), cfg()); + let h0 = ms(2026, 5, 6, 12, 0, 0); + let chunks = cs.list_chunks("never_written", h0, h0 + 60_000).await.unwrap(); + assert!(chunks.is_empty()); + let samples = cs.scan("never_written", h0, h0 + 60_000).await.unwrap(); + assert!(samples.is_empty()); + } + + #[tokio::test] + async fn scan_filters_to_requested_range() { + // Chunk has samples at h0+1_000 and h0+10_000; request only + // [h0+5_000, h0+9_000) — chunk overlaps the request, but the + // matching sample is *outside* the inner filter, so scan + // returns 0 samples (read_chunk would still load + cache the + // chunk). + let store = Arc::new(InMemoryObjectStore::new()); + let h0 = ms(2026, 5, 6, 12, 0, 0); + let block = make_block( + "m", + &[], + &[(h0 + 1_000, 1.0), (h0 + 10_000, 2.0)], + ); + let key = "tenant1/m/2026/05/06/12/part-Z.gor".to_string(); + store.put(key.clone(), block.clone()).await; + store + .put( + "tenant1/m/2026/05/06/12/index.json", + make_index(vec![IndexEntry { + key: key.clone(), + time_range: ( + (h0 + 1_000) as u64 * 1_000_000, + (h0 + 10_000) as u64 * 1_000_000, + ), + sample_count: 2, + label_hash: 0, + size_bytes: block.len() as u32, + }]), + ) + .await; + + let cs = GorillaS3ColdStore::new(store, cfg()); + let samples = cs.scan("m", h0 + 5_000, h0 + 9_000).await.unwrap(); + assert!(samples.is_empty(), "no sample inside [5_000, 9_000) ms"); + + let samples = cs.scan("m", h0, h0 + 60_000).await.unwrap(); + assert_eq!(samples.len(), 2); + } + + #[tokio::test] + async fn list_chunks_spans_two_hour_buckets() { + let store = InMemoryObjectStore::new(); + let h12 = ms(2026, 5, 6, 12, 0, 0); + let h13 = ms(2026, 5, 6, 13, 0, 0); + let key12 = "tenant1/m/2026/05/06/12/part-1.gor".to_string(); + let key13 = "tenant1/m/2026/05/06/13/part-1.gor".to_string(); + store + .put( + "tenant1/m/2026/05/06/12/index.json", + make_index(vec![IndexEntry { + key: key12.clone(), + time_range: ( + (h12 + 3_500_000) as u64 * 1_000_000, + (h12 + 3_590_000) as u64 * 1_000_000, + ), + sample_count: 1, + label_hash: 0, + size_bytes: 50, + }]), + ) + .await; + store + .put( + "tenant1/m/2026/05/06/13/index.json", + make_index(vec![IndexEntry { + key: key13.clone(), + time_range: ( + (h13 + 1_000) as u64 * 1_000_000, + (h13 + 30_000) as u64 * 1_000_000, + ), + sample_count: 1, + label_hash: 0, + size_bytes: 50, + }]), + ) + .await; + let cs = GorillaS3ColdStore::new(Arc::new(store), cfg()); + let chunks = cs + .list_chunks("m", h12 + 3_500_000, h13 + 30_000) + .await + .unwrap(); + assert_eq!(chunks.len(), 2); + assert_eq!(chunks[0].key, key12); + assert_eq!(chunks[1].key, key13); + } + + #[test] + fn from_env_requires_bucket() { + // Don't pollute global env in a unit test; just exercise the + // missing-var path. + let prev_bucket = std::env::var("ASAP_GORILLA_S3_BUCKET").ok(); + std::env::remove_var("ASAP_GORILLA_S3_BUCKET"); + let res = GorillaS3Config::from_env(); + if let Some(v) = prev_bucket { + std::env::set_var("ASAP_GORILLA_S3_BUCKET", v); + } + match res { + Err(GorillaS3ConfigError::MissingEnv("ASAP_GORILLA_S3_BUCKET")) => {} + other => panic!("expected MissingEnv(BUCKET), got {other:?}"), + } + } +} diff --git a/asap-query-engine/src/drivers/query/fallback/cold_store/mod.rs b/asap-query-engine/src/drivers/query/fallback/cold_store/mod.rs index c9a8a6791..77bd8e6c4 100644 --- a/asap-query-engine/src/drivers/query/fallback/cold_store/mod.rs +++ b/asap-query-engine/src/drivers/query/fallback/cold_store/mod.rs @@ -34,9 +34,11 @@ use std::collections::BTreeMap; use thiserror::Error; pub mod format; +pub mod gorilla_s3; pub mod local_fs; pub use format::{part_path_prefix, RawSample}; +pub use gorilla_s3::{GorillaS3ColdStore, GorillaS3Config, GorillaS3ConfigError}; pub use local_fs::LocalFsColdStore; /// Error surface for cold-store scans. @@ -46,6 +48,51 @@ pub enum ColdStoreError { Io(#[from] std::io::Error), #[error("malformed raw record: {0}")] Malformed(String), + /// Backend-storage error (e.g. an S3 GET failed) that is not + /// itself a `std::io::Error`. Phase 3 introduced this variant for + /// the Gorilla-S3 cold store; the local-FS path keeps using + /// [`ColdStoreError::Io`]. + #[error("backend error: {0}")] + Backend(String), + /// A trait method that this `ColdStore` impl does not support. + /// Returned by the default `list_chunks` / `read_chunk` impls on + /// JSONL-only stores; Gorilla-S3 / future chunk-native stores + /// override. + #[error("unsupported cold-store operation: {0}")] + Unsupported(&'static str), +} + +/// Descriptor for a single immutable cold-store chunk. +/// +/// Returned by [`ColdStore::list_chunks`] for chunk-native backends +/// (Phase 3+ Gorilla-S3). Carries enough metadata for callers to +/// prune by time / label without reading the chunk body. +#[derive(Debug, Clone, PartialEq)] +pub struct ChunkRef { + /// Opaque object key (e.g. an S3 key). The Telegraf-side + /// `gorilla_s3` output uses + /// `/block---.gorilla`; the + /// design.md-style layout is `//YYYY/MM/DD/HH/ + /// part-NNNNNN.gor`. Either is fine — the index file is the + /// source of truth for what keys exist. + pub key: String, + /// Metric name the chunk was fetched against. Recovered from + /// the caller's `list_chunks` request rather than the on-wire + /// chunk metadata, since not all backends require chunks to be + /// metric-pure. + pub metric: String, + /// `(start_unix_ms, end_unix_ms)` covered by the chunk — + /// converted from the on-wire nanosecond range so it can be + /// directly compared with [`ColdStore::scan`]'s + /// `[start_ms, end_ms)` window. + pub time_range_ms: (i64, i64), + /// 64-bit canonical-label-set hash — for prune-by-label-equality + /// without fetching the chunk. + pub label_hash: u64, + /// Number of samples in the chunk. + pub sample_count: u32, + /// On-wire size of the chunk object in bytes. + pub size_bytes: u32, } /// Read-only view over a cold raw-sample store. @@ -62,6 +109,17 @@ pub enum ColdStoreError { /// Label matching is **not** pushed down here — callers filter /// samples client-side. This keeps the trait small and makes the /// local-FS / S3 impls trivially swappable. +/// +/// # Phase-3 trait extension +/// +/// The `list_chunks` / `read_chunk` pair is additive (default impls +/// return [`ColdStoreError::Unsupported`]) so the existing JSONL +/// `LocalFsColdStore` keeps compiling unchanged. Chunk-native +/// backends (Gorilla-S3) override both so the upcoming +/// `GorillaQueryEngine` can iterate chunks one at a time without +/// materialising every sample up front. See +/// [`docs/design-gorilla-s3-cold-engine.md` §7.2](#) for the +/// rationale. #[async_trait] pub trait ColdStore: Send + Sync { /// Return all samples for `metric` whose timestamp lies in @@ -72,6 +130,33 @@ pub trait ColdStore: Send + Sync { start_ms: i64, end_ms: i64, ) -> Result, ColdStoreError>; + + /// List chunk descriptors covering `[start_ms, end_ms)` without + /// decoding any bodies. Default impl returns + /// [`ColdStoreError::Unsupported`] — only chunk-native backends + /// (e.g. [`GorillaS3ColdStore`]) override. + async fn list_chunks( + &self, + _metric: &str, + _start_ms: i64, + _end_ms: i64, + ) -> Result, ColdStoreError> { + Err(ColdStoreError::Unsupported("list_chunks")) + } + + /// Decode a single chunk into an owned `Vec`. + /// + /// Returning `Vec` rather than a streaming iterator keeps the + /// trait object-safe and matches the existing `scan` contract; + /// chunks are bounded-size in practice (Phase 1 emits one series + /// per ~1 hour). The decoded samples can also be cached cheaply + /// by the impl. Default returns [`ColdStoreError::Unsupported`]. + async fn read_chunk( + &self, + _chunk: &ChunkRef, + ) -> Result, ColdStoreError> { + Err(ColdStoreError::Unsupported("read_chunk")) + } } /// Convenience alias: a label set as stored in a [`RawSample`].