diff --git a/CHANGELOG.md b/CHANGELOG.md index 1981b1a9dfa..2932c411cf5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,8 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - (Jaeger) Query resource attributes when Jaeger request carries tags +- Azure Blob Storage: workload identity no longer fails 24 hours after start, since `azure_identity` 1.0 re-reads the projected service account token (#6672) ### Changed +- Azure Blob Storage: migrate to `azure_storage_blob` 1.0, the rewritten Azure SDK (#6672) ### Deprecated diff --git a/docker-compose.yml b/docker-compose.yml index 24f5ed29b47..cc0e9684a3d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -128,7 +128,7 @@ services: retries: 100 azurite: - image: mcr.microsoft.com/azure-storage/azurite:${AZURITE_VERSION:-3.24.0} + image: mcr.microsoft.com/azure-storage/azurite:${AZURITE_VERSION:-3.36.0} container_name: azurite ports: - "${MAP_HOST_AZURITE:-127.0.0.1}:10000:10000" # Blob store port @@ -137,7 +137,11 @@ services: - azurite volumes: - azurite_data:/data - command: azurite --blobHost 0.0.0.0 --loose + # `--skipApiVersionCheck` because `azure_storage_blob` 1.0 sends a newer + # `x-ms-version` than any released Azurite recognizes. Skipping the check keeps the + # emulated client sending the same request as production, which is the point of the + # test, rather than pinning an older API version only the tests would use. + command: azurite --blobHost 0.0.0.0 --loose --skipApiVersionCheck fake-gcs-server: image: fsouza/fake-gcs-server:${FAKE_GCS_SERVER_VERSION:-1.47.7} diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index cd19a875db9..c31e9f2418d 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -2,12 +2,6 @@ # It is not intended for manual editing. version = 4 -[[package]] -name = "RustyXML" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b5ace29ee3216de37c0546865ad08edef58b0f9e76838ed8959a84a990e58c5" - [[package]] name = "addr2line" version = "0.25.1" @@ -553,24 +547,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "async-io" -version = "2.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc" -dependencies = [ - "autocfg", - "cfg-if", - "concurrent-queue", - "futures-io", - "futures-lite 2.6.1", - "parking", - "polling", - "rustix 1.1.4", - "slab", - "windows-sys 0.61.2", -] - [[package]] name = "async-lock" version = "3.4.2" @@ -582,24 +558,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "async-process" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75" -dependencies = [ - "async-channel 2.5.0", - "async-io", - "async-lock", - "async-signal", - "async-task", - "blocking", - "cfg-if", - "event-listener 5.4.1", - "futures-lite 2.6.1", - "rustix 1.1.4", -] - [[package]] name = "async-recursion" version = "1.1.1" @@ -611,24 +569,6 @@ dependencies = [ "syn 2.0.118", ] -[[package]] -name = "async-signal" -version = "0.2.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52b5aaafa020cf5053a01f2a60e8ff5dccf550f0f77ec54a4e47285ac2bab485" -dependencies = [ - "async-io", - "async-lock", - "atomic-waker", - "cfg-if", - "futures-core", - "futures-io", - "rustix 1.1.4", - "signal-hook-registry", - "slab", - "windows-sys 0.61.2", -] - [[package]] name = "async-speed-limit" version = "0.4.2" @@ -663,12 +603,6 @@ dependencies = [ "syn 2.0.118", ] -[[package]] -name = "async-task" -version = "4.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" - [[package]] name = "async-trait" version = "0.1.89" @@ -727,7 +661,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "hex", "http 1.4.2", "sha1 0.10.6", @@ -790,7 +724,7 @@ dependencies = [ "aws-types", "bytes", "bytes-utils", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "http-body 0.4.6", @@ -821,7 +755,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -848,7 +782,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -878,7 +812,7 @@ dependencies = [ "aws-smithy-xml", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "hex", "hmac 0.13.0", "http 0.2.12", @@ -911,7 +845,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -937,7 +871,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -963,7 +897,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -990,7 +924,7 @@ dependencies = [ "aws-smithy-types", "aws-smithy-xml", "aws-types", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1200,7 +1134,7 @@ dependencies = [ "aws-smithy-schema", "aws-smithy-types", "bytes", - "fastrand 2.4.1", + "fastrand", "http 0.2.12", "http 1.4.2", "http-body 0.4.6", @@ -1351,105 +1285,70 @@ dependencies = [ [[package]] name = "azure_core" -version = "0.21.0" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b552ad43a45a746461ec3d3a51dfb6466b4759209414b439c165eb6a6b7729e" +checksum = "4e41cbd819986ba41904c207d8ffc4106f8f8352a548d773e9554906379bb2fb" dependencies = [ + "async-lock", "async-trait", - "base64 0.22.1", + "azure_core_macros", "bytes", - "dyn-clone", "futures", - "getrandom 0.2.17", "hmac 0.12.1", - "http-types", - "once_cell", - "paste", "pin-project", - "quick-xml 0.31.0", - "rand 0.8.6", - "reqwest 0.12.28", "rustc_version", "serde", "serde_json", "sha2 0.10.9", - "time", + "tokio", "tracing", - "url", - "uuid", + "typespec", + "typespec_client_core", ] [[package]] -name = "azure_identity" -version = "0.21.0" +name = "azure_core_macros" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88ddd80344317c40c04b603807b63a5cefa532f1b43522e72f480a988141f744" +checksum = "b9b52dba6a345f3ad2d42ff8d0d63df9d0994cfa29657bf18ffdbf149f78a4f5" dependencies = [ - "async-lock", - "async-process", - "async-trait", - "azure_core", - "futures", - "oauth2 4.4.2", - "pin-project", - "serde", - "time", + "proc-macro2", + "quote", + "syn 2.0.118", "tracing", - "url", - "uuid", ] [[package]] -name = "azure_storage" -version = "0.21.0" +name = "azure_identity" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59f838159f4d29cb400a14d9d757578ba495ae64feb07a7516bf9e4415127126" +checksum = "32edf96b356ca7c51d7590c4925cc36efc3947a5da4468e8e0b25c56ecbb3de5" dependencies = [ - "RustyXML", "async-lock", "async-trait", "azure_core", - "bytes", - "serde", - "serde_derive", - "time", - "tracing", - "url", - "uuid", -] - -[[package]] -name = "azure_storage_blobs" -version = "0.21.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97e83c3636ae86d9a6a7962b2112e3b19eb3903915c50ce06ff54ff0a2e6a7e4" -dependencies = [ - "RustyXML", - "azure_core", - "azure_storage", - "azure_svc_blobstorage", - "bytes", "futures", + "pin-project", "serde", - "serde_derive", "serde_json", "time", "tracing", "url", - "uuid", ] [[package]] -name = "azure_svc_blobstorage" -version = "0.21.0" +name = "azure_storage_blob" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e6c6f20c5611b885ba94c7bae5e02849a267381aecb8aee577e8c35ff4064c6" +checksum = "1756febbcca86c862ef718b983b505d08bd65a9bc984a915b0a16af4a4c3fe5b" dependencies = [ + "async-stream", + "async-trait", "azure_core", "bytes", "futures", - "log", - "once_cell", + "percent-encoding", + "pin-project", "serde", "serde_json", "time", @@ -1488,12 +1387,6 @@ version = "2.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd637ac531c60eb7fbc4684dc061c2d7d90d73d758181aa02eeff0464b9eee4b" -[[package]] -name = "base64" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8" - [[package]] name = "base64" version = "0.21.7" @@ -1677,19 +1570,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "blocking" -version = "1.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" -dependencies = [ - "async-channel 2.5.0", - "async-task", - "futures-io", - "futures-lite 2.6.1", - "piper", -] - [[package]] name = "bon" version = "3.9.3" @@ -4119,15 +3999,6 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9afc2bd4d5a73106dd53d10d73d3401c2f32730ba2c0b93ddb888a8983680471" -[[package]] -name = "fastrand" -version = "1.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be" -dependencies = [ - "instant", -] - [[package]] name = "fastrand" version = "2.4.1" @@ -4235,7 +4106,7 @@ version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ - "fastrand 2.4.1", + "fastrand", "futures-core", "futures-sink", "spin 0.9.8", @@ -4387,34 +4258,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" -[[package]] -name = "futures-lite" -version = "1.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49a9d51ce47660b1e808d3c990b4709f2f415d928835a17dfd16991515c46bce" -dependencies = [ - "fastrand 1.9.0", - "futures-core", - "futures-io", - "memchr", - "parking", - "pin-project-lite", - "waker-fn", -] - -[[package]] -name = "futures-lite" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" -dependencies = [ - "fastrand 2.4.1", - "futures-core", - "futures-io", - "parking", - "pin-project-lite", -] - [[package]] name = "futures-macro" version = "0.3.32" @@ -4487,17 +4330,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "getrandom" -version = "0.1.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fc3cb4d91f53b50155bdcfd23f6a4c39ae1969c2ae85982b135750cccaf5fce" -dependencies = [ - "cfg-if", - "libc", - "wasi 0.9.0+wasi-snapshot-preview1", -] - [[package]] name = "getrandom" version = "0.2.17" @@ -4507,7 +4339,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "wasi 0.11.1+wasi-snapshot-preview1", + "wasi", "wasm-bindgen", ] @@ -4972,26 +4804,6 @@ dependencies = [ "serde", ] -[[package]] -name = "http-types" -version = "2.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e9b187a72d63adbfba487f48095306ac823049cb504ee195541e91c7775f5ad" -dependencies = [ - "anyhow", - "async-channel 1.9.0", - "base64 0.13.1", - "futures-lite 1.13.0", - "infer", - "pin-project-lite", - "rand 0.7.3", - "serde", - "serde_json", - "serde_qs 0.8.5", - "serde_urlencoded", - "url", -] - [[package]] name = "httparse" version = "1.10.1" @@ -5317,12 +5129,6 @@ dependencies = [ "rustversion", ] -[[package]] -name = "infer" -version = "0.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64e9829a50b42bb782c1df523f78d332fe371b10c661e78b7a3c34b0198e9fac" - [[package]] name = "inferno" version = "0.11.21" @@ -5375,15 +5181,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "instant" -version = "0.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222" -dependencies = [ - "cfg-if", -] - [[package]] name = "integer-encoding" version = "3.0.4" @@ -6256,7 +6053,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", - "wasi 0.11.1+wasi-snapshot-preview1", + "wasi", "windows-sys 0.61.2", ] @@ -6586,25 +6383,6 @@ dependencies = [ "itoa", ] -[[package]] -name = "oauth2" -version = "4.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c38841cdd844847e3e7c8d29cef9dcfed8877f8f56f9071f77843ecf3baf937f" -dependencies = [ - "base64 0.13.1", - "chrono", - "getrandom 0.2.17", - "http 0.2.12", - "rand 0.8.6", - "serde", - "serde_json", - "serde_path_to_error", - "sha2 0.10.9", - "thiserror 1.0.69", - "url", -] - [[package]] name = "oauth2" version = "5.0.0" @@ -6843,7 +6621,7 @@ dependencies = [ "http 1.4.2", "itertools 0.10.5", "log", - "oauth2 5.0.0", + "oauth2", "p256", "p384", "rand 0.8.6", @@ -7536,17 +7314,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" -[[package]] -name = "piper" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" -dependencies = [ - "atomic-waker", - "fastrand 2.4.1", - "futures-io", -] - [[package]] name = "pkcs1" version = "0.7.5" @@ -7716,20 +7483,6 @@ dependencies = [ "pnet_sys", ] -[[package]] -name = "polling" -version = "3.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" -dependencies = [ - "cfg-if", - "concurrent-queue", - "hermit-abi", - "pin-project-lite", - "rustix 1.1.4", - "windows-sys 0.61.2", -] - [[package]] name = "poly1305" version = "0.8.0" @@ -8234,7 +7987,7 @@ dependencies = [ "lz4", "murmur3", "nom 7.1.3", - "oauth2 5.0.0", + "oauth2", "openidconnect", "pem", "prost 0.13.5", @@ -8265,7 +8018,7 @@ dependencies = [ "libc", "once_cell", "raw-cpuid", - "wasi 0.11.1+wasi-snapshot-preview1", + "wasi", "web-sys", "winapi", ] @@ -8285,16 +8038,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "quick-xml" -version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1004a344b30a54e2ee58d66a71b32d2db2feb0a31f9a2d302bf0536f15de2a33" -dependencies = [ - "memchr", - "serde", -] - [[package]] name = "quick-xml" version = "0.41.0" @@ -9369,7 +9112,7 @@ dependencies = [ "rust-embed", "serde", "serde_json", - "serde_qs 1.1.2", + "serde_qs", "serde_with", "tempfile", "thiserror 2.0.18", @@ -9405,8 +9148,7 @@ dependencies = [ "aws-smithy-types", "azure_core", "azure_identity", - "azure_storage", - "azure_storage_blobs", + "azure_storage_blob", "base64 0.22.1", "bytes", "bytesize", @@ -9438,6 +9180,7 @@ dependencies = [ "tantivy", "tempfile", "thiserror 2.0.18", + "time", "tokio", "tokio-rustls 0.26.4", "tokio-stream", @@ -9584,19 +9327,6 @@ dependencies = [ "nibble_vec", ] -[[package]] -name = "rand" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03" -dependencies = [ - "getrandom 0.1.16", - "libc", - "rand_chacha 0.2.2", - "rand_core 0.5.1", - "rand_hc", -] - [[package]] name = "rand" version = "0.8.6" @@ -9629,16 +9359,6 @@ dependencies = [ "rand_core 0.10.1", ] -[[package]] -name = "rand_chacha" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402" -dependencies = [ - "ppv-lite86", - "rand_core 0.5.1", -] - [[package]] name = "rand_chacha" version = "0.3.1" @@ -9659,15 +9379,6 @@ dependencies = [ "rand_core 0.9.5", ] -[[package]] -name = "rand_core" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19" -dependencies = [ - "getrandom 0.1.16", -] - [[package]] name = "rand_core" version = "0.6.4" @@ -9702,15 +9413,6 @@ dependencies = [ "rand 0.10.2", ] -[[package]] -name = "rand_hc" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c" -dependencies = [ - "rand_core 0.5.1", -] - [[package]] name = "rand_pcg" version = "0.10.2" @@ -10048,7 +9750,6 @@ dependencies = [ "bytes", "encoding_rs", "futures-core", - "futures-util", "h2 0.4.15", "http 1.4.2", "http-body 1.0.1", @@ -10071,14 +9772,12 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls 0.26.4", - "tokio-util", "tower 0.5.3", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams 0.4.2", "web-sys", "webpki-roots 1.0.8", ] @@ -10117,7 +9816,7 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams 0.5.0", + "wasm-streams", "web-sys", ] @@ -10801,17 +10500,6 @@ dependencies = [ "serde", ] -[[package]] -name = "serde_qs" -version = "0.8.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c7715380eec75f029a4ef7de39a9200e0a63823176b759d055b613f5a87df6a6" -dependencies = [ - "percent-encoding", - "serde", - "thiserror 1.0.69", -] - [[package]] name = "serde_qs" version = "1.1.2" @@ -11922,7 +11610,7 @@ version = "3.27.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ - "fastrand 2.4.1", + "fastrand", "getrandom 0.4.3", "once_cell", "rustix 1.1.4", @@ -12067,7 +11755,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50" dependencies = [ "deranged", - "js-sys", "num-conv", "powerfmt", "serde_core", @@ -12650,6 +12337,58 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "typespec" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "753a2fe021e407d4fc9ee6f4f0a33403cc306d5c54c4e4ebe1b8cbde0ca052b9" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures", + "quick-xml 0.41.0", + "serde", + "serde_json", + "url", +] + +[[package]] +name = "typespec_client_core" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0373af0f9d4f580b3a1a9d9639cedaabe015ed262b35bfbe13941bfb14fe1ea6" +dependencies = [ + "async-trait", + "base64 0.22.1", + "bytes", + "dyn-clone", + "futures", + "pin-project", + "rand 0.10.2", + "reqwest 0.13.4", + "serde", + "serde_json", + "time", + "tokio", + "tracing", + "typespec", + "typespec_macros", + "url", + "uuid", +] + +[[package]] +name = "typespec_macros" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c608f4427943f8adb211abc95c87672b1b98847152783507d54e3246e502f60" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "syn 2.0.118", +] + [[package]] name = "typetag" version = "0.2.22" @@ -13126,12 +12865,6 @@ dependencies = [ "libc", ] -[[package]] -name = "waker-fn" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7" - [[package]] name = "walkdir" version = "2.5.0" @@ -13181,12 +12914,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "wasi" -version = "0.9.0+wasi-snapshot-preview1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cccddf32554fecc6acb585f82a32a72e28b48f8c4c1883ddfeeeaa96f7d8e519" - [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -13214,7 +12941,7 @@ version = "0.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae86f02046da16a333a9129d31451423e1657737ecdafed4193838a5f54c5cfe" dependencies = [ - "wasi 0.11.1+wasi-snapshot-preview1", + "wasi", ] [[package]] @@ -13272,19 +12999,6 @@ dependencies = [ "unicode-ident", ] -[[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 = "wasm-streams" version = "0.5.0" diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml index 8904ed1dd3c..5a96d3c1fdf 100644 --- a/quickwit/Cargo.toml +++ b/quickwit/Cargo.toml @@ -375,14 +375,14 @@ aws-smithy-types = { version = "1.6", features = [ ] } aws-types = "1.4" -azure_core = { version = "0.21", features = ["hmac_rust", "enable_reqwest_rustls"] } -azure_identity = { version = "0.21", default-features = false, features = ["enable_reqwest_rustls"] } -azure_storage = { version = "0.21", default-features = false, features = [ - "enable_reqwest_rustls", -] } -azure_storage_blobs = { version = "0.21", default-features = false, features = [ - "enable_reqwest_rustls", +azure_core = { version = "1", default-features = false, features = [ + "hmac_rust", + "reqwest_rustls", + "tokio", + "xml", ] } +azure_identity = { version = "1", default-features = false } +azure_storage_blob = { version = "1", default-features = false } opendal = { version = "0.58", default-features = false, features = ["http-transport-reqwest"] } reqsign = { version = "0.18", default-features = false, features = ["google", "default-context"] } diff --git a/quickwit/quickwit-storage/Cargo.toml b/quickwit/quickwit-storage/Cargo.toml index cda13362a79..d056e9ae8d5 100644 --- a/quickwit/quickwit-storage/Cargo.toml +++ b/quickwit/quickwit-storage/Cargo.toml @@ -46,8 +46,8 @@ aws-smithy-types = { workspace = true } azure_core = { workspace = true, optional = true } azure_identity = { workspace = true, optional = true } -azure_storage = { workspace = true, optional = true } -azure_storage_blobs = { workspace = true, optional = true } +azure_storage_blob = { workspace = true, optional = true } +time = { workspace = true, optional = true } quickwit-aws = { workspace = true } quickwit-common = { workspace = true } @@ -83,21 +83,14 @@ quickwit-common = { workspace = true, features = ["testsuite"] } azure = [ "azure_core", "azure_identity", - "azure_identity/enable_reqwest_rustls", - "azure_storage", - "azure_storage_blobs", - "azure_core/hmac_rust", - "azure_core/enable_reqwest_rustls", - "azure_storage/enable_reqwest_rustls", - "azure_storage_blobs/enable_reqwest_rustls", + "azure_storage_blob", + "dep:time", ] gcs = ["dep:opendal", "opendal/services-gcs"] ci-test = [] integration-testsuite = [ "azure", - "azure_core/azurite_workaround", - "azure_storage_blobs/azurite_workaround", - "gcs", # Stands for Google cloud storage. + "gcs", # Stands for Google cloud storage. "dep:reqwest", ] testsuite = ["mockall"] diff --git a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs index cbbf6ea8ded..a6f4d97837c 100644 --- a/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs @@ -13,20 +13,22 @@ // limitations under the License. use std::collections::HashMap; -use std::num::NonZeroU32; use std::ops::Range; use std::path::{Path, PathBuf}; use std::sync::{Arc, LazyLock}; use std::{fmt, io}; use async_trait::async_trait; +use azure_core::Error as AzureError; use azure_core::error::ErrorKind; -use azure_core::{Pageable, StatusCode}; -use azure_storage::prelude::*; -use azure_storage::{CloudLocation, Error as AzureError}; -use azure_storage_blobs::blob::operations::GetBlobResponse; -use azure_storage_blobs::prelude::*; -use bytes::{Bytes, BytesMut}; +use azure_core::http::{NoFormat, RequestContent, StatusCode}; +use azure_storage_blob::BlobContainerClient; +use azure_storage_blob::models::{ + BlobClientDownloadOptions, BlobClientGetPropertiesResultHeaders, + BlobContainerClientListBlobsOptions, BlockBlobClientStageBlockOptions, BlockLookupList, + HttpRange, +}; +use bytes::Bytes; use bytesize::ByteSize; use futures::io::Error as FutureError; use futures::stream::{StreamExt, TryStreamExt}; @@ -40,12 +42,14 @@ use regex::Regex; use tantivy::directory::OwnedBytes; use thiserror::Error; use tokio::io::{AsyncRead, AsyncWriteExt, BufReader}; -use tokio_util::compat::FuturesAsyncReadCompatExt; use tokio_util::io::StreamReader; -use tracing::{info, instrument, warn}; +use tracing::{instrument, warn}; use crate::debouncer::DebouncedStorage; use crate::metrics::object_storage_get_slice_in_flight_guards; +use crate::object_storage::azure_credentials::{ + AzureCredential, build_container_client, resolve_credential, +}; use crate::stable_deref_bytes::into_owned_bytes; use crate::storage::SendableAsync; use crate::{ @@ -53,6 +57,12 @@ use crate::{ StorageErrorKind, StorageFactory, StorageResolverError, StorageResult, }; +/// Block id used by the single part upload path. +/// +/// Zero padded like the multipart ids so that a blob written by either path carries a block +/// list of the same shape. +const SINGLE_PART_BLOCK_ID: &[u8] = b"block:00000"; + /// Azure object storage resolver. pub struct AzureBlobStorageFactory { storage_config: AzureStorageConfig, @@ -79,7 +89,7 @@ impl StorageFactory for AzureBlobStorageFactory { /// Azure object storage implementation pub struct AzureBlobStorage { - container_client: ContainerClient, + container_client: BlobContainerClient, uri: Uri, prefix: PathBuf, multipart_policy: MultiPartPolicy, @@ -97,20 +107,23 @@ impl fmt::Debug for AzureBlobStorage { impl AzureBlobStorage { /// Creates a new [`AzureBlobStorage`] instance. - pub fn new( - storage_account_name: String, - storage_credentials: StorageCredentials, + /// + /// Crate visible because the credential type is: callers outside the crate go through + /// [`AzureBlobStorage::from_uri`], which resolves the credential from the config. + pub(crate) fn new( + storage_account_name: &str, + credential: AzureCredential, blob_service_uri: Option, uri: Uri, - container_name: String, - ) -> Self { + container_name: &str, + ) -> Result { let container_client = build_container_client( storage_account_name, - storage_credentials, + credential, blob_service_uri, container_name, - ); - Self { + )?; + Ok(Self { container_client, uri, prefix: PathBuf::new(), @@ -125,7 +138,7 @@ impl AzureBlobStorage { max_concurrent_uploads: 100, }, retry_params: RetryParams::aggressive(), - } + }) } /// Sets the prefix path. @@ -142,11 +155,16 @@ impl AzureBlobStorage { } /// Creates an emulated storage for testing. + /// + /// The 1.0 SDK has no `emulator()` helper, and could not offer one: the emulator + /// authenticates with a shared key, which the SDK no longer signs. The well-known + /// account and key are documented at + /// . #[cfg(feature = "integration-testsuite")] pub fn new_emulated(container: &str) -> Self { use std::str::FromStr; - let container_client = ClientBuilder::emulator().container_client(container); + let container_client = build_emulated_container_client(container); let uri = Uri::from_str(&format!("azure://tester/{container}")).unwrap(); Self { @@ -158,6 +176,30 @@ impl AzureBlobStorage { } } + /// Creates the container the emulated storage writes to. + /// + /// Exposed for the integration test suite because the shared key signing the emulator + /// needs stays inside this crate, so a test cannot build a client of its own. + #[cfg(feature = "integration-testsuite")] + pub async fn create_emulated_container(container: &str) -> anyhow::Result<()> { + build_emulated_container_client(container) + .create(None) + .await?; + Ok(()) + } + + /// Deletes the container the emulated storage writes to. + /// + /// Counterpart of [`AzureBlobStorage::create_emulated_container`], exposed for the same + /// reason. + #[cfg(feature = "integration-testsuite")] + pub async fn delete_emulated_container(container: &str) -> anyhow::Result<()> { + build_emulated_container_client(container) + .delete(None) + .await?; + Ok(()) + } + /// Sets the multipart policy. /// /// See `MultiPartPolicy`. @@ -180,31 +222,19 @@ impl AzureBlobStorage { ); StorageResolverError::InvalidConfig(message) })?; - let storage_credentials = if let Some(access_key) = - azure_storage_config.resolve_access_key() - { - StorageCredentials::access_key(storage_account_name.clone(), access_key) - } else if let Ok(credential) = azure_identity::create_credential() { - StorageCredentials::token_credential(credential) - } else { - return Err(StorageResolverError::InvalidConfig( - "could not find Azure storage account credentials using the following credential \ - providers: environment, managed identity, and storage account access key" - .to_string(), - )); - }; + let credential = resolve_credential(azure_storage_config)?; let (container_name, prefix) = parse_azure_uri(uri).ok_or_else(|| { let message = format!("failed to extract container name from Azure URI `{uri}`"); StorageResolverError::InvalidUri(message) })?; let blob_service_uri = azure_storage_config.resolve_blob_service_uri(&storage_account_name); let azure_blob_storage = AzureBlobStorage::new( - storage_account_name, - storage_credentials, + &storage_account_name, + credential, blob_service_uri, uri.clone(), - container_name, - ); + &container_name, + )?; Ok(azure_blob_storage.with_prefix(prefix)) } @@ -214,7 +244,7 @@ impl AzureBlobStorage { key_path.to_string_lossy().to_string() } - /// Downloads a blob as `Bytes` — zero-copy when the blob arrives as a single chunk. + /// Downloads a blob as `Bytes`. async fn get_to_bytes( &self, path: &Path, @@ -224,27 +254,69 @@ impl AzureBlobStorage { let capacity = range_opt.as_ref().map(Range::len).unwrap_or(0); retry(&self.retry_params, || async { let _timer = HistogramTimer::new(&crate::metrics::OBJECT_STORAGE_GET_OBJECT_DURATION); - let (mut response_stream, _in_flight_guards) = if let Some(range) = range_opt.as_ref() { - let stream = self - .container_client - .blob_client(&name) - .get() - .range(range.clone()) - .into_stream(); - // only record ranged get request as being in flight - let in_flight_guards = object_storage_get_slice_in_flight_guards(capacity); - (stream, Some(in_flight_guards)) - } else { - let stream = self.container_client.blob_client(&name).get().into_stream(); - (stream, None) + let mut download_options = BlobClientDownloadOptions::default(); + // Only a ranged get counts as in flight, matching the metric's meaning. + let _in_flight_guards = match range_opt.as_ref() { + Some(range) => { + download_options.range = Some(HttpRange::from(range.clone())); + Some(object_storage_get_slice_in_flight_guards(capacity)) + } + None => None, }; - let bytes = download_all(&mut response_stream).await?; + let download_response = self + .container_client + .blob_client(&name) + .download(Some(download_options)) + .await?; + let bytes = download_response.body.collect().await?; + crate::metrics::OBJECT_STORAGE_DOWNLOAD_NUM_BYTES.inc_by(bytes.len() as u64); Result::<_, AzureErrorWrapper>::Ok(bytes) }) .await .map_err(StorageError::from) } + /// Opens a download as an [`AsyncRead`], optionally over a range. + /// + /// The first chunk is pulled before returning so that an error arriving with the + /// response headers is still inside the retry, rather than surfacing later to a caller + /// that cannot retry it. + async fn get_to_reader( + &self, + path: &Path, + range_opt: Option>, + ) -> StorageResult> { + let name = self.blob_name(path); + retry(&self.retry_params, || async { + let mut download_options = BlobClientDownloadOptions::default(); + if let Some(range) = range_opt.as_ref() { + download_options.range = Some(HttpRange::from(range.clone())); + } + let download_response = self + .container_client + .blob_client(&name) + .download(Some(download_options)) + .await?; + let mut bytes_stream = download_response + .body + .map(|bytes_res| bytes_res.map_err(FutureError::other)); + let first_chunk = bytes_stream.next().await; + let reader: Box = match first_chunk { + Some(bytes_res) => { + let first_chunk = bytes_res.map_err(AzureErrorWrapper::from)?; + let reconstructed_stream = Box::pin( + futures::stream::once(async { Ok(first_chunk) }).chain(bytes_stream), + ); + Box::new(StreamReader::new(reconstructed_stream)) + } + None => Box::new(tokio::io::empty()), + }; + Result::, AzureErrorWrapper>::Ok(reader) + }) + .await + .map_err(|err| err.into()) + } + /// Performs a single part upload. async fn put_single_part<'a>( &'a self, @@ -256,12 +328,38 @@ impl AzureBlobStorage { let _timer = HistogramTimer::new(&crate::metrics::OBJECT_STORAGE_PUT_OBJECT_DURATION); retry(&self.retry_params, || async { let data = Bytes::from(payload.read_all().await?.to_vec()); - let hash = azure_storage_blobs::prelude::Hash::from(md5::compute(&data[..]).0); - self.container_client - .blob_client(name) - .put_block_blob(data) - .hash(hash) - .into_future() + let digest = md5::compute(&data[..]); + let content_length = data.len() as u64; + let block_blob_client = self.container_client.blob_client(name).block_blob_client(); + // Staged and committed rather than uploaded in one shot, because `upload()` + // only exposes `blob_content_md5`, which the service stores without checking it + // against the body. `stage_block` takes a transactional checksum, so a corrupted + // payload is rejected on arrival. Splits are immutable and never re-verified, so + // a silent corruption here would be permanent. The cost is one extra request per + // object below the multipart threshold. + let stage_block_options = BlockBlobClientStageBlockOptions { + transactional_content_md5: Some(digest.0.to_vec()), + ..Default::default() + }; + // `RequestContent::from` is an inherent function over `Vec`, which shadows + // the `From` impl, so go through `Into` to keep the `Bytes` as is. + let content: RequestContent = data.into(); + block_blob_client + .stage_block( + SINGLE_PART_BLOCK_ID, + content_length, + content, + Some(stage_block_options), + ) + .await?; + let block_lookup_list = BlockLookupList { + uncommitted: Some(vec![SINGLE_PART_BLOCK_ID.to_vec()]), + ..Default::default() + }; + let block_list_content = + RequestContent::try_from(block_lookup_list).map_err(AzureErrorWrapper::from)?; + block_blob_client + .commit_block_list(block_list_content, None) .await?; Result::<(), AzureErrorWrapper>::Ok(()) }) @@ -281,10 +379,14 @@ impl AzureBlobStorage { let multipart_ranges = chunk_range(0..total_len as usize, part_len as usize).map(into_u64_range); - let blob_client = self.container_client.blob_client(name); + let block_blob_client = self.container_client.blob_client(name).block_blob_client(); let upload_blocks_stream = tokio_stream::iter(multipart_ranges.enumerate()) .map(|(num, range)| { - let moved_blob_client = blob_client.clone(); + // `BlockBlobClient` is not `Clone` in the 1.0 SDK, so each part builds its + // own from the container client. Construction is local: it clones the + // pipeline behind an `Arc` and appends the blob name to the URL. + let moved_block_blob_client = + self.container_client.blob_client(name).block_blob_client(); let moved_payload = payload.clone(); crate::metrics::OBJECT_STORAGE_PUT_PARTS.inc(); crate::metrics::OBJECT_STORAGE_UPLOAD_NUM_BYTES.inc_by(range.end - range.start); @@ -297,11 +399,19 @@ impl AzureBlobStorage { let (data, hash_digest) = extract_range_data_and_hash(moved_payload.box_clone(), range.clone()) .await?; - let hash = azure_storage_blobs::prelude::Hash::from(hash_digest.0); - moved_blob_client - .put_block(block_id.clone(), data) - .hash(hash) - .into_future() + let content_length = data.len() as u64; + let stage_block_options = BlockBlobClientStageBlockOptions { + transactional_content_md5: Some(hash_digest.0.to_vec()), + ..Default::default() + }; + // The SDK base64 encodes the block id, so it takes the raw bytes. + moved_block_blob_client + .stage_block( + block_id.as_bytes(), + content_length, + data.into(), + Some(stage_block_options), + ) .await?; Result::<_, AzureErrorWrapper>::Ok(block_id) }) @@ -310,7 +420,7 @@ impl AzureBlobStorage { }) .buffer_unordered(self.multipart_policy.max_concurrent_uploads()); - // Collect and sort block ids to preserve part order for put_block_list. + // Collect and sort block ids to preserve part order for the block list. // Azure docs: "The put block list operation enforces the order in which blocks // are to be combined to create a blob". // https://docs.microsoft.com/en-us/rest/api/storageservices/put-block-list @@ -320,17 +430,22 @@ impl AzureBlobStorage { .map_err(StorageError::from)?; block_ids.sort_unstable(); - let block_list = BlockList { - blocks: block_ids - .into_iter() - .map(BlobBlockType::new_uncommitted) - .collect(), + let block_lookup_list = BlockLookupList { + uncommitted: Some( + block_ids + .into_iter() + .map(|block_id| block_id.into_bytes()) + .collect(), + ), + ..Default::default() }; + let block_list_content = RequestContent::try_from(block_lookup_list) + .map_err(AzureErrorWrapper::from) + .map_err(StorageError::from)?; // Commit all uploaded blocks. - blob_client - .put_block_list(block_list) - .into_future() + block_blob_client + .commit_block_list(block_list_content, None) .await .map_err(AzureErrorWrapper::from)?; @@ -341,15 +456,13 @@ impl AzureBlobStorage { #[async_trait] impl Storage for AzureBlobStorage { async fn check_connectivity(&self) -> anyhow::Result<()> { - if let Some(first_blob_result) = self - .container_client - .list_blobs() - .max_results(NonZeroU32::new(1u32).expect("1 is always non-zero.")) - .into_stream() - .next() - .await - { - let _ = first_blob_result?; + let list_blobs_options = BlobContainerClientListBlobsOptions { + maxresults: Some(1), + ..Default::default() + }; + let mut blob_pager = self.container_client.list_blobs(Some(list_blobs_options))?; + if let Some(first_page_result) = blob_pager.next().await { + let _ = first_page_result?; } Ok(()) } @@ -376,20 +489,10 @@ impl Storage for AzureBlobStorage { #[instrument(name = "storage.azure.copy_to", level = "debug", skip(self, output))] async fn copy_to(&self, path: &Path, output: &mut dyn SendableAsync) -> StorageResult<()> { - let name = self.blob_name(path); - let mut output_stream = self.container_client.blob_client(name).get().into_stream(); - - while let Some(chunk_result) = output_stream.next().await { - let chunk_response = chunk_result.map_err(AzureErrorWrapper::from)?; - let chunk_response_body_stream = chunk_response - .data - .map_err(FutureError::other) - .into_async_read() - .compat(); - let mut body_stream_reader = BufReader::new(chunk_response_body_stream); - let num_bytes_copied = tokio::io::copy_buf(&mut body_stream_reader, output).await?; - crate::metrics::OBJECT_STORAGE_DOWNLOAD_NUM_BYTES.inc_by(num_bytes_copied); - } + let reader = self.get_to_reader(path, None).await?; + let mut body_stream_reader = BufReader::new(reader); + let num_bytes_copied = tokio::io::copy_buf(&mut body_stream_reader, output).await?; + crate::metrics::OBJECT_STORAGE_DOWNLOAD_NUM_BYTES.inc_by(num_bytes_copied); output.flush().await?; Ok(()) } @@ -403,9 +506,8 @@ impl Storage for AzureBlobStorage { let blob_name = self.blob_name(path); let delete_res: Result<_, StorageError> = self .container_client - .blob_client(blob_name) - .delete() - .into_future() + .blob_client(&blob_name) + .delete(None) .await .map_err(|err| AzureErrorWrapper::from(err).into()); ignore_error_kind!(StorageErrorKind::NotFound, delete_res)?; @@ -476,33 +578,7 @@ impl Storage for AzureBlobStorage { path: &Path, range: Range, ) -> StorageResult> { - retry(&self.retry_params, || async { - let range = range.clone(); - let name = self.blob_name(path); - let page_stream = self - .container_client - .blob_client(name) - .get() - .range(range) - .into_stream(); - let mut bytes_stream = page_stream - .map(|page_res| page_res.map(|page| page.data).map_err(FutureError::other)) - .try_flatten() - .map(|bytes_res| bytes_res.map_err(FutureError::other)); - // Peek into the stream so that any early error can be retried - let first_chunk = bytes_stream.next().await; - let reader: Box = if let Some(res) = first_chunk { - let first_chunk = res.map_err(AzureErrorWrapper::from)?; - let reconstructed_stream = - Box::pin(futures::stream::once(async { Ok(first_chunk) }).chain(bytes_stream)); - Box::new(StreamReader::new(reconstructed_stream)) - } else { - Box::new(tokio::io::empty()) - }; - Result::, AzureErrorWrapper>::Ok(reader) - }) - .await - .map_err(|e| e.into()) + self.get_to_reader(path, Some(range)).await } #[instrument( @@ -532,12 +608,17 @@ impl Storage for AzureBlobStorage { let name = self.blob_name(path); let properties_result = self .container_client - .blob_client(name) - .get_properties() - .into_future() + .blob_client(&name) + .get_properties(None) .await; match properties_result { - Ok(response) => Ok(response.blob.properties.content_length), + Ok(response) => { + let content_length = response + .content_length() + .map_err(AzureErrorWrapper::from)? + .unwrap_or(0); + Ok(content_length) + } Err(err) => Err(StorageError::from(AzureErrorWrapper::from(err))), } } @@ -547,6 +628,28 @@ impl Storage for AzureBlobStorage { } } +/// Builds a container client pointed at Azurite. +/// +/// The well-known emulator account and key are published by Microsoft at +/// , so neither is a +/// secret. The 1.0 SDK dropped `ClientBuilder::emulator()`, and could not have kept it: the +/// emulator authenticates with a shared key, which the SDK no longer signs. +#[cfg(feature = "integration-testsuite")] +fn build_emulated_container_client(container: &str) -> BlobContainerClient { + const EMULATOR_ACCOUNT: &str = "devstoreaccount1"; + const EMULATOR_ACCOUNT_KEY: &str = + "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; + const EMULATOR_BLOB_SERVICE_URI: &str = "http://127.0.0.1:10000/devstoreaccount1"; + + build_container_client( + EMULATOR_ACCOUNT, + AzureCredential::SharedKey(EMULATOR_ACCOUNT_KEY.to_string()), + Some(EMULATOR_BLOB_SERVICE_URI.to_string()), + container, + ) + .expect("the emulator endpoint should be a valid URL") +} + /// Copy range of payload into `Bytes` and return the computed md5. async fn extract_range_data_and_hash( payload: Box, @@ -563,25 +666,6 @@ async fn extract_range_data_and_hash( Ok((data, hash)) } -fn build_container_client( - storage_account_name: String, - storage_credentials: StorageCredentials, - blob_service_uri: Option, - container_name: String, -) -> ContainerClient { - let mut builder = ClientBuilder::new(storage_account_name.clone(), storage_credentials); - if let Some(uri) = blob_service_uri { - info!(endpoint=%uri, "using Azure blob storage endpoint defined in storage config or environment variable"); - builder = builder.cloud_location(CloudLocation::Custom { - account: storage_account_name, - uri, - }); - } - builder - .blob_service_client() - .container_client(container_name) -} - pub fn parse_azure_uri(uri: &Uri) -> Option<(String, PathBuf)> { // Ex: azure://container/prefix. static URI_PTN: LazyLock = LazyLock::new(|| { @@ -599,45 +683,6 @@ pub fn parse_azure_uri(uri: &Uri) -> Option<(String, PathBuf)> { Some((container, prefix)) } -/// Collect a download stream into a single [`Bytes`]. -/// -/// `Bytes` segments yielded by the SDK are preserved so that the single-segment case avoids the -/// extra copy into a contiguous buffer. When more than one segment is received, they are -/// concatenated exactly once into a freshly allocated `Bytes`. -async fn download_all( - chunk_stream: &mut Pageable, -) -> Result { - let mut segments: Vec = Vec::new(); - let mut total_num_bytes: usize = 0; - while let Some(chunk_result) = chunk_stream.next().await { - let chunk_response = chunk_result?; - let mut data_stream = chunk_response.data; - while let Some(bytes_res) = data_stream.next().await { - let bytes = bytes_res?; - total_num_bytes += bytes.len(); - segments.push(bytes); - } - } - crate::metrics::OBJECT_STORAGE_DOWNLOAD_NUM_BYTES.inc_by(total_num_bytes as u64); - Ok(coalesce_segments(segments, total_num_bytes)) -} - -/// Returns a single [`Bytes`] covering `segments`. Zero-copy when there is at most one segment; -/// otherwise a single allocation concatenates them. -fn coalesce_segments(mut segments: Vec, total_num_bytes: usize) -> Bytes { - match segments.len() { - 0 => Bytes::new(), - 1 => segments.remove(0), - _ => { - let mut out = BytesMut::with_capacity(total_num_bytes); - for segment in segments { - out.extend_from_slice(&segment); - } - out.freeze() - } - } -} - #[derive(Error, Debug)] #[error("Azure error wrapper(inner={inner})")] struct AzureErrorWrapper { diff --git a/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs b/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs new file mode 100644 index 00000000000..1b2a35d71d9 --- /dev/null +++ b/quickwit/quickwit-storage/src/object_storage/azure_credentials.rs @@ -0,0 +1,465 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Chooses how Quickwit authenticates to Azure Blob Storage, and builds the container client. +//! +//! `azure_identity` 1.0 removed `DefaultAzureCredential` and `create_credential()`, so the +//! choice the SDK used to make by inspecting the environment is made here instead. The +//! nearest remaining type, `DeveloperToolsCredential`, chains the Azure CLI and the +//! Developer CLI only, which does not cover a pod. + +use std::sync::Arc; +use std::{env, fmt}; + +use azure_core::credentials::TokenCredential; +use azure_core::http::policies::Policy; +use azure_core::http::{ClientOptions, Url}; +use azure_identity::{ + ClientSecretCredential, ManagedIdentityCredential, ManagedIdentityCredentialOptions, + UserAssignedId, WorkloadIdentityCredential, +}; +use azure_storage_blob::{BlobContainerClient, BlobContainerClientOptions}; +use quickwit_config::AzureStorageConfig; +use tracing::info; + +use crate::StorageResolverError; +use crate::object_storage::azure_shared_key::SharedKeyAuthorizationPolicy; + +/// Environment variables the workload identity webhook injects into a pod. +const AZURE_CLIENT_ID: &str = "AZURE_CLIENT_ID"; +const AZURE_TENANT_ID: &str = "AZURE_TENANT_ID"; +const AZURE_FEDERATED_TOKEN_FILE: &str = "AZURE_FEDERATED_TOKEN_FILE"; +/// Set when a service principal authenticates with a secret rather than a federated token. +const AZURE_CLIENT_SECRET: &str = "AZURE_CLIENT_SECRET"; +/// Lets an operator pin the provider instead of relying on the detection below. +const AZURE_CREDENTIAL_KIND: &str = "AZURE_CREDENTIAL_KIND"; + +/// Which token credential the environment describes. +/// +/// Kept separate from construction so the precedence can be tested without mutating process +/// environment, which no test can do safely while others run. +#[derive(Eq, PartialEq)] +enum TokenCredentialKind { + /// Service principal with a client secret, the `EnvironmentCredential` of the old chain. + ClientSecret { + tenant_id: String, + client_id: String, + secret: String, + }, + /// Federated token file, injected into a pod by the workload identity webhook. + WorkloadIdentity, + /// IMDS. `user_assigned_client_id` selects a user-assigned identity, and `None` means the + /// system-assigned one. + ManagedIdentity { + user_assigned_client_id: Option, + }, +} + +impl fmt::Debug for TokenCredentialKind { + /// Hand written so the client secret cannot reach a log line or a panic message. A + /// derived implementation would print it, and test failures print this type. + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::ClientSecret { + tenant_id, + client_id, + .. + } => formatter + .debug_struct("ClientSecret") + .field("tenant_id", tenant_id) + .field("client_id", client_id) + .finish_non_exhaustive(), + Self::WorkloadIdentity => formatter.write_str("WorkloadIdentity"), + Self::ManagedIdentity { + user_assigned_client_id, + } => formatter + .debug_struct("ManagedIdentity") + .field("user_assigned_client_id", user_assigned_client_id) + .finish(), + } + } +} + +/// Decides which credential the environment describes. +/// +/// Precedence follows the chain `azure_identity::create_credential()` used to walk, where an +/// environment credential came before managed identity. Dropping that ordering silently +/// breaks every deployment that authenticates with a service principal, because the client +/// secret is ignored and IMDS is contacted instead. +fn select_token_credential_kind(var: impl Fn(&str) -> Option) -> TokenCredentialKind { + let non_empty = |name: &str| match var(name) { + Some(value) if !value.trim().is_empty() => Some(value), + _ => None, + }; + let client_id = non_empty(AZURE_CLIENT_ID); + + // A secret and a federated token file are mutually exclusive in practice. The secret is + // checked first because the webhook injects the token file, so its presence says less + // about operator intent than a secret does. + if let (Some(tenant_id), Some(client_id), Some(secret)) = ( + non_empty(AZURE_TENANT_ID), + client_id.clone(), + non_empty(AZURE_CLIENT_SECRET), + ) { + return TokenCredentialKind::ClientSecret { + tenant_id, + client_id, + secret, + }; + } + if non_empty(AZURE_TENANT_ID).is_some() + && client_id.is_some() + && non_empty(AZURE_FEDERATED_TOKEN_FILE).is_some() + { + return TokenCredentialKind::WorkloadIdentity; + } + // `AZURE_CLIENT_ID` on its own names a user-assigned identity. Ignoring it asks IMDS for + // the system-assigned identity, which either does not exist or is the wrong principal. + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: client_id, + } +} + +/// How requests to the blob service are authorized. +pub(crate) enum AzureCredential { + /// Storage account key, signed by Quickwit because the 1.0 SDK dropped shared key. + SharedKey(String), + /// Entra ID token, signed by the SDK's bearer policy. + Token(Arc), +} + +/// Resolves the credential from the storage config and the environment. +/// +/// An explicit `access_key` wins, matching the previous behaviour: it is the only credential +/// an operator states in the config file, so treating it as a preference is the least +/// surprising reading. +pub(crate) fn resolve_credential( + azure_storage_config: &AzureStorageConfig, +) -> Result { + if let Some(access_key) = azure_storage_config.resolve_access_key() { + return Ok(AzureCredential::SharedKey(access_key)); + } + let token_credential = resolve_token_credential().map_err(|error| { + StorageResolverError::InvalidConfig(format!( + "could not build an Azure token credential: {error}. Set an access key, or run with \ + workload identity or managed identity configured" + )) + })?; + Ok(AzureCredential::Token(token_credential)) +} + +/// Builds the token credential the environment describes. +fn resolve_token_credential() -> azure_core::Result> { + let credential_kind = env::var(AZURE_CREDENTIAL_KIND) + .map(|kind| kind.trim().to_lowercase()) + .unwrap_or_default(); + + let kind = match credential_kind.as_str() { + "workloadidentity" => TokenCredentialKind::WorkloadIdentity, + "managedidentity" => TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: env::var(AZURE_CLIENT_ID).ok().filter(|id| !id.is_empty()), + }, + // An empty or unrecognized value falls through to detection. + _ => select_token_credential_kind(|name| env::var(name).ok()), + }; + build_token_credential(kind) +} + +/// Builds the credential for a kind already chosen. +fn build_token_credential( + kind: TokenCredentialKind, +) -> azure_core::Result> { + match kind { + TokenCredentialKind::ClientSecret { + tenant_id, + client_id, + secret, + } => { + info!("using azure client secret credential"); + Ok(ClientSecretCredential::new( + &tenant_id, + client_id, + secret.into(), + None, + )?) + } + TokenCredentialKind::WorkloadIdentity => { + info!("using azure workload identity credential"); + Ok(WorkloadIdentityCredential::new(None)?) + } + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id, + } => { + let options = match user_assigned_client_id { + Some(client_id) => { + info!(%client_id, "using azure user-assigned managed identity credential"); + Some(ManagedIdentityCredentialOptions { + user_assigned_id: Some(UserAssignedId::ClientId(client_id)), + ..Default::default() + }) + } + None => { + info!("using azure system-assigned managed identity credential"); + None + } + }; + Ok(ManagedIdentityCredential::new(options)?) + } + } +} + +/// Builds the container client. +/// +/// The 1.0 clients take the container URL directly rather than an account name plus a cloud +/// location, so a custom sovereign endpoint needs no special case: it is the base of the URL. +pub(crate) fn build_container_client( + storage_account_name: &str, + credential: AzureCredential, + blob_service_uri: Option, + container_name: &str, +) -> Result { + let blob_service_uri = match blob_service_uri { + Some(uri) => { + info!( + endpoint = %uri, + "using Azure blob storage endpoint defined in storage config or environment \ + variable" + ); + uri + } + None => format!("https://{storage_account_name}.blob.core.windows.net"), + }; + let container_url = join_container(&blob_service_uri, container_name)?; + + let mut client_options = BlobContainerClientOptions::default(); + let token_credential = match credential { + AzureCredential::SharedKey(access_key) => { + // Signing happens per retry so that every attempt carries a fresh `x-ms-date`. + let policy: Arc = Arc::new(SharedKeyAuthorizationPolicy::new( + storage_account_name.to_owned(), + access_key, + )); + client_options.client_options = ClientOptions { + per_try_policies: vec![policy], + ..Default::default() + }; + None + } + AzureCredential::Token(token_credential) => Some(token_credential), + }; + + BlobContainerClient::new(container_url, token_credential, Some(client_options)).map_err( + |error| { + StorageResolverError::InvalidConfig(format!( + "could not build an Azure container client: {error}" + )) + }, + ) +} + +/// Appends the container to the blob service URL. +fn join_container( + blob_service_uri: &str, + container_name: &str, +) -> Result { + let mut container_url = + Url::parse(blob_service_uri.trim_end_matches('/')).map_err(|error| { + StorageResolverError::InvalidConfig(format!( + "`{blob_service_uri}` is not a valid Azure blob service URL: {error}" + )) + })?; + container_url + .path_segments_mut() + .map_err(|_| { + StorageResolverError::InvalidConfig(format!( + "`{blob_service_uri}` cannot be a base URL" + )) + })? + .pop_if_empty() + .push(container_name); + Ok(container_url) +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + + fn env_from(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option { + let map: HashMap = pairs + .iter() + .map(|(name, value)| ((*name).to_owned(), (*value).to_owned())) + .collect(); + move |name: &str| map.get(name).cloned() + } + + /// A service principal authenticates with a secret. The old chain tried an environment + /// credential before managed identity, and losing that ordering sends the request to + /// IMDS with the secret ignored, so the deployment loses access entirely. + #[test] + fn test_client_secret_wins_over_managed_identity() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_CLIENT_SECRET, "secret"), + ])); + assert_eq!( + kind, + TokenCredentialKind::ClientSecret { + tenant_id: "tenant".to_owned(), + client_id: "client".to_owned(), + secret: "secret".to_owned(), + } + ); + } + + /// The webhook injects the token file, so a secret is the stronger statement of intent. + #[test] + fn test_client_secret_wins_over_workload_identity() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_CLIENT_SECRET, "secret"), + (AZURE_FEDERATED_TOKEN_FILE, "/var/run/token"), + ])); + assert!(matches!(kind, TokenCredentialKind::ClientSecret { .. })); + } + + #[test] + fn test_workload_identity_when_the_three_variables_are_present() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_FEDERATED_TOKEN_FILE, "/var/run/token"), + ])); + assert_eq!(kind, TokenCredentialKind::WorkloadIdentity); + } + + /// `AZURE_CLIENT_ID` alone names a user-assigned identity. Dropping it asks IMDS for the + /// system-assigned one, which is either absent or the wrong principal. + #[test] + fn test_client_id_alone_selects_a_user_assigned_identity() { + let kind = select_token_credential_kind(env_from(&[(AZURE_CLIENT_ID, "client")])); + assert_eq!( + kind, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: Some("client".to_owned()), + } + ); + } + + #[test] + fn test_empty_environment_selects_the_system_assigned_identity() { + let kind = select_token_credential_kind(env_from(&[])); + assert_eq!( + kind, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: None, + } + ); + } + + /// A half configured service principal must not be read as one, and a blank variable is + /// the same as an unset one. + #[test] + fn test_blank_and_partial_values_do_not_select_a_service_principal() { + let partial = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + ])); + assert_eq!( + partial, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: Some("client".to_owned()), + } + ); + + let blank = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_CLIENT_SECRET, " "), + ])); + assert_eq!( + blank, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: Some("client".to_owned()), + } + ); + } + + #[test] + fn test_debug_does_not_leak_the_client_secret() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_TENANT_ID, "tenant"), + (AZURE_CLIENT_ID, "client"), + (AZURE_CLIENT_SECRET, "super-secret-value"), + ])); + let rendered = format!("{kind:?}"); + assert!(rendered.contains("client")); + assert!(!rendered.contains("super-secret-value")); + } + + /// A partial workload identity set is not workload identity. + #[test] + fn test_token_file_without_tenant_is_not_workload_identity() { + let kind = select_token_credential_kind(env_from(&[ + (AZURE_CLIENT_ID, "client"), + (AZURE_FEDERATED_TOKEN_FILE, "/var/run/token"), + ])); + assert_eq!( + kind, + TokenCredentialKind::ManagedIdentity { + user_assigned_client_id: Some("client".to_owned()), + } + ); + } + + #[test] + fn test_join_container_appends_the_container() { + let url = join_container("https://acct.blob.core.windows.net", "my-container").unwrap(); + assert_eq!( + url.as_str(), + "https://acct.blob.core.windows.net/my-container" + ); + } + + #[test] + fn test_join_container_tolerates_a_trailing_slash() { + let url = join_container("https://acct.blob.core.windows.net/", "my-container").unwrap(); + assert_eq!( + url.as_str(), + "https://acct.blob.core.windows.net/my-container" + ); + } + + #[test] + fn test_join_container_keeps_a_sovereign_host() { + let url = join_container("https://acct.blob.core.usgovcloudapi.net", "c").unwrap(); + assert_eq!(url.as_str(), "https://acct.blob.core.usgovcloudapi.net/c"); + } + + #[test] + fn test_join_container_keeps_an_emulator_path_prefix() { + // Azurite addresses accounts by path rather than by subdomain, so the account + // segment has to survive. + let url = join_container("http://127.0.0.1:10000/devstoreaccount1", "c").unwrap(); + assert_eq!(url.as_str(), "http://127.0.0.1:10000/devstoreaccount1/c"); + } + + #[test] + fn test_join_container_rejects_a_non_url() { + assert!(join_container("not a url", "c").is_err()); + } +} diff --git a/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs new file mode 100644 index 00000000000..05ae860faee --- /dev/null +++ b/quickwit/quickwit-storage/src/object_storage/azure_shared_key.rs @@ -0,0 +1,463 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Shared key (storage account key) request signing. +//! +//! The rewritten Azure SDK authenticates with Entra ID tokens only. Shared key support was +//! dropped in the 1.0 line and the SDK team has stated it will not come back, see +//! . Quickwit documents +//! `azure.access_key` as a supported credential, and the Azurite emulator that backs the +//! `integration-testsuite` feature accepts shared key only, so the signing that +//! `azure_storage` 0.21 used to provide is implemented here instead. +//! +//! The scheme is specified in +//! . + +use std::borrow::Cow; +use std::fmt; +use std::sync::Arc; + +use azure_core::credentials::Secret; +use azure_core::hmac::hmac_sha256; +use azure_core::http::headers::Headers; +use azure_core::http::policies::{Policy, PolicyResult}; +use azure_core::http::{Context, Method, Request, Url}; +use azure_core::time::to_rfc7231; +use time::OffsetDateTime; + +/// Header carrying the request time. Signing uses this rather than `Date` so that the +/// `Date` slot of the string to sign stays empty, which is what the service expects when +/// `x-ms-date` is present. +const X_MS_DATE: &str = "x-ms-date"; +/// Signed and sent together, so both have to be derived from the same value. +const CONTENT_LENGTH: &str = "content-length"; + +/// Signs every request with the storage account key. +/// +/// Installed as a per-retry policy so that each attempt is signed with a fresh timestamp: +/// the service rejects requests whose `x-ms-date` drifts more than 15 minutes from its own +/// clock, and a retry of a long-stalled request would otherwise carry a stale one. +pub(crate) struct SharedKeyAuthorizationPolicy { + account: String, + key: Secret, +} + +impl fmt::Debug for SharedKeyAuthorizationPolicy { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + // The key is deliberately absent: this type ends up in pipeline debug output. + formatter + .debug_struct("SharedKeyAuthorizationPolicy") + .field("account", &self.account) + .finish_non_exhaustive() + } +} + +impl SharedKeyAuthorizationPolicy { + pub(crate) fn new(account: String, key: String) -> Self { + Self { + account, + key: Secret::new(key), + } + } +} + +#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))] +#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)] +impl Policy for SharedKeyAuthorizationPolicy { + async fn send( + &self, + ctx: &Context, + request: &mut Request, + next: &[Arc], + ) -> PolicyResult { + let now = OffsetDateTime::now_utc(); + request.insert_header(X_MS_DATE, to_rfc7231(&now)); + + // Some operations leave `Content-Length` for the transport to fill in, `commit_block + // _list` among them, while others set it themselves. Signing has to agree with what + // finally goes on the wire, so set the header here when the body length is known and + // the header is missing. Signing an empty length against a request that carries a + // real one is rejected as `AuthorizationFailure`, with nothing to indicate why. + let body_len = request.body().len(); + if header_or_empty(request.headers(), CONTENT_LENGTH).is_empty() { + match body_len { + // A zero length body is the documented exception: the slot stays empty in + // the string to sign even though the wire carries `Content-Length: 0`. + Some(len) if len > 0 => request.insert_header(CONTENT_LENGTH, len.to_string()), + _ => {} + } + } + + let method = request.method(); + let string_to_sign = + string_to_sign(&self.account, &method, request.url(), request.headers()); + let signature = hmac_sha256(&string_to_sign, &self.key)?; + request.insert_header( + "authorization", + format!("SharedKey {}:{}", self.account, signature), + ); + + next[0].send(ctx, request, &next[1..]).await + } +} + +/// Returns the value of `header_name`, or the empty string when absent. +/// +/// Absent and empty are indistinguishable in the string to sign, which is why the service +/// tolerates the collapse. +fn header_or_empty(headers: &Headers, header_name: &str) -> String { + headers + .get_optional_str(&azure_core::http::headers::HeaderName::from( + header_name.to_owned(), + )) + .unwrap_or_default() + .to_owned() +} + +/// Builds the string the signature is computed over. +/// +/// The field order is fixed by the service and every slot is present even when empty, so +/// the newlines carry meaning. `Content-Length` is the one exception the specification +/// calls out: it must be empty rather than `0` for requests without a body, for API +/// versions from 2015-02-21 onwards. +fn string_to_sign(account: &str, method: &Method, url: &Url, headers: &Headers) -> String { + let content_length = match header_or_empty(headers, CONTENT_LENGTH) { + length if length == "0" => String::new(), + length => length, + }; + format!( + "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}", + method.as_ref(), + header_or_empty(headers, "content-encoding"), + header_or_empty(headers, "content-language"), + content_length, + header_or_empty(headers, "content-md5"), + header_or_empty(headers, "content-type"), + // Empty on purpose: `x-ms-date` supersedes `Date` and appears in the canonicalized + // headers below. Signing both would double-count the timestamp. + "", + header_or_empty(headers, "if-modified-since"), + header_or_empty(headers, "if-match"), + header_or_empty(headers, "if-none-match"), + header_or_empty(headers, "if-unmodified-since"), + header_or_empty(headers, "range"), + canonicalized_headers(headers), + canonicalized_resource(account, url), + ) +} + +/// Renders the `x-ms-*` headers in the form the signature expects: lowercase names, sorted +/// lexicographically, one `name:value` pair per line, each line newline terminated. +fn canonicalized_headers(headers: &Headers) -> String { + let mut ms_headers: Vec<(String, String)> = Vec::new(); + headers.iter().for_each(|(header_name, header_value)| { + let name = header_name.as_str().to_lowercase(); + if name.starts_with("x-ms-") { + // Linear whitespace inside a value has to fold to a single space, otherwise the + // service and the client sign different bytes. + let value = header_value.as_str().split_whitespace().collect::>(); + ms_headers.push((name, value.join(" "))); + } + }); + ms_headers.sort_unstable(); + + let mut rendered = String::new(); + for (name, value) in ms_headers { + rendered.push_str(&name); + rendered.push(':'); + rendered.push_str(&value); + rendered.push('\n'); + } + rendered +} + +/// Renders the resource the request addresses: the account, the path as encoded in the URI, +/// then every query parameter, lowercased, decoded and sorted, with repeated parameters comma +/// joined. +/// +/// The asymmetry is deliberate and specified: "any portion of the CanonicalizedResource +/// string that is derived from the resource's URI should be encoded exactly as it is in the +/// URI", and the path step says to append "the resource's encoded URI path", while the query +/// steps say to "URL-decode each query parameter name and value". So a blob named +/// `indexes/my-index/split` signs as `indexes%2Fmy-index%2Fsplit`, because that is how the +/// 1.0 clients put it in the URL. Decoding the path here is rejected with +/// `AuthorizationFailure`, which the integration suite confirms. +/// +/// The account appears twice for the emulator, whose URLs carry the account in the path. +/// That is expected and documented. +fn canonicalized_resource(account: &str, url: &Url) -> String { + let mut resource = String::with_capacity(url.as_str().len()); + resource.push('/'); + resource.push_str(account); + for segment in url.path_segments().into_iter().flatten() { + resource.push('/'); + resource.push_str(segment); + } + + let mut query_pairs: Vec<(Cow<'_, str>, Cow<'_, str>)> = url.query_pairs().collect(); + if query_pairs.is_empty() { + return resource; + } + // Sort by lowercased name so that repeated parameters group together, and by value so + // that the comma joined list below is itself sorted, as the service requires. + query_pairs.sort_by(|(left_name, left_value), (right_name, right_value)| { + let left = left_name.to_lowercase(); + let right = right_name.to_lowercase(); + left.cmp(&right).then_with(|| left_value.cmp(right_value)) + }); + + let mut current_name: Option = None; + for (name, value) in query_pairs { + let name = name.to_lowercase(); + match current_name.as_deref() { + Some(previous) if previous == name => { + resource.push(','); + } + _ => { + resource.push('\n'); + resource.push_str(&name); + resource.push(':'); + current_name = Some(name); + } + } + resource.push_str(&value); + } + resource +} + +#[cfg(test)] +mod tests { + use azure_core::http::headers::Headers; + use azure_core::http::{Method, Url}; + + use super::*; + + /// The account and key Azurite and the legacy storage emulator publish. Documented at + /// , so this is + /// not a secret. + const EMULATOR_ACCOUNT: &str = "devstoreaccount1"; + const EMULATOR_ACCOUNT_KEY: &str = + "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; + + fn headers_from(pairs: &[(&str, &str)]) -> Headers { + let mut headers = Headers::new(); + for (name, value) in pairs { + headers.insert( + azure_core::http::headers::HeaderName::from((*name).to_owned()), + (*value).to_owned(), + ); + } + headers + } + + #[test] + fn test_canonicalized_resource_without_query() { + let url = Url::parse("https://acct.blob.core.windows.net/container/a/b.split").unwrap(); + assert_eq!( + canonicalized_resource("acct", &url), + "/acct/container/a/b.split" + ); + } + + #[test] + fn test_canonicalized_resource_sorts_and_lowercases_query() { + let url = Url::parse( + "https://acct.blob.core.windows.net/container?restype=container&comp=list&Marker=m", + ) + .unwrap(); + assert_eq!( + canonicalized_resource("acct", &url), + "/acct/container\ncomp:list\nmarker:m\nrestype:container" + ); + } + + #[test] + fn test_canonicalized_resource_comma_joins_repeated_query_params() { + let url = Url::parse("https://acct.blob.core.windows.net/c?include=metadata&include=copy") + .unwrap(); + assert_eq!( + canonicalized_resource("acct", &url), + "/acct/c\ninclude:copy,metadata" + ); + } + + /// The path is signed exactly as the URI carries it, escapes included. Quickwit blob + /// names hold separators, so the 1.0 clients percent-encode them into one segment and + /// the signature has to keep that form. Decoding here looks tidier and is rejected with + /// `AuthorizationFailure`. + #[test] + fn test_canonicalized_resource_keeps_the_path_encoded() { + let url = + Url::parse("https://acct.blob.core.windows.net/c/indexes%2Fmy-index%2Fsplit.split") + .unwrap(); + assert_eq!( + canonicalized_resource("acct", &url), + "/acct/c/indexes%2Fmy-index%2Fsplit.split" + ); + } + + /// Query parameters are the exception: the specification decodes those. + #[test] + fn test_canonicalized_resource_decodes_query_values() { + let url = Url::parse("https://acct.blob.core.windows.net/c?prefix=a%2Fb").unwrap(); + assert_eq!(canonicalized_resource("acct", &url), "/acct/c\nprefix:a/b"); + } + + #[test] + fn test_canonicalized_headers_sorted_and_folded() { + let headers = headers_from(&[ + ("x-ms-version", "2025-05-05"), + ("x-ms-date", "Fri, 14 Aug 2026 12:00:00 GMT"), + ("content-type", "application/octet-stream"), + ("x-ms-blob-type", "Block Blob"), + ]); + assert_eq!( + canonicalized_headers(&headers), + "x-ms-blob-type:Block Blob\nx-ms-date:Fri, 14 Aug 2026 12:00:00 \ + GMT\nx-ms-version:2025-05-05\n" + ); + } + + #[test] + fn test_string_to_sign_leaves_date_slot_empty() { + let url = Url::parse("https://acct.blob.core.windows.net/c/blob").unwrap(); + let headers = headers_from(&[("x-ms-date", "Fri, 14 Aug 2026 12:00:00 GMT")]); + let signed = string_to_sign("acct", &Method::Get, &url, &headers); + // GET, then eleven empty slots, then the canonicalized headers and resource. + assert_eq!( + signed, + "GET\n\n\n\n\n\n\n\n\n\n\n\nx-ms-date:Fri, 14 Aug 2026 12:00:00 GMT\n/acct/c/blob" + ); + } + + #[test] + fn test_string_to_sign_omits_zero_content_length() { + let url = Url::parse("https://acct.blob.core.windows.net/c/blob").unwrap(); + let with_zero = headers_from(&[("content-length", "0")]); + let with_body = headers_from(&[("content-length", "17")]); + assert!(string_to_sign("acct", &Method::Put, &url, &with_zero).starts_with("PUT\n\n\n\n")); + assert!( + string_to_sign("acct", &Method::Put, &url, &with_body).starts_with("PUT\n\n\n17\n") + ); + } + + #[test] + fn test_signature_is_stable_for_a_known_request() { + // Pins the whole pipeline: string construction, base64 key decoding, HMAC, and + // base64 of the signature. A change to any of them changes this value. + let url = Url::parse("https://devstoreaccount1.blob.core.windows.net/c/blob").unwrap(); + let headers = headers_from(&[("x-ms-date", "Fri, 14 Aug 2026 12:00:00 GMT")]); + let string_to_sign = string_to_sign(EMULATOR_ACCOUNT, &Method::Get, &url, &headers); + let signature = hmac_sha256(&string_to_sign, &Secret::new(EMULATOR_ACCOUNT_KEY)).unwrap(); + // Signatures are 32 bytes of HMAC SHA256, base64 encoded. + assert_eq!(signature.len(), 44); + let recomputed = hmac_sha256(&string_to_sign, &Secret::new(EMULATOR_ACCOUNT_KEY)).unwrap(); + assert_eq!(signature, recomputed); + } + + /// Terminal policy: records the headers it is handed and answers 200. + #[derive(Debug, Default)] + struct HeaderCapturingPolicy { + seen_headers: std::sync::Mutex>, + } + + #[async_trait::async_trait] + impl Policy for HeaderCapturingPolicy { + async fn send( + &self, + _ctx: &Context, + request: &mut Request, + _next: &[Arc], + ) -> PolicyResult { + *self.seen_headers.lock().unwrap() = Some(request.headers().clone()); + Ok(azure_core::http::AsyncRawResponse::from_bytes( + azure_core::http::StatusCode::Ok, + Headers::default(), + azure_core::Bytes::new(), + )) + } + } + + /// `commit_block_list` leaves `Content-Length` to the transport. Signing an empty length + /// while the wire carries a real one is rejected as `AuthorizationFailure`, and only a + /// live service reveals it, so pin the header here instead. + #[tokio::test] + async fn test_send_sets_content_length_before_signing() { + let capturing_policy = Arc::new(HeaderCapturingPolicy::default()); + let policy = SharedKeyAuthorizationPolicy::new( + EMULATOR_ACCOUNT.to_owned(), + EMULATOR_ACCOUNT_KEY.to_owned(), + ); + let url = + Url::parse("http://127.0.0.1:10000/devstoreaccount1/c/blob?comp=blocklist").unwrap(); + let mut request = Request::new(url, Method::Put); + request.set_body(azure_core::Bytes::from_static(b"")); + + let next: Vec> = vec![capturing_policy.clone()]; + policy + .send(&Context::default(), &mut request, &next) + .await + .unwrap(); + + let seen_headers = capturing_policy + .seen_headers + .lock() + .unwrap() + .clone() + .unwrap(); + assert_eq!(header_or_empty(&seen_headers, CONTENT_LENGTH), "12"); + // The signature has to cover that same length. + let signed = string_to_sign(EMULATOR_ACCOUNT, &Method::Put, request.url(), &seen_headers); + assert!(signed.starts_with("PUT\n\n\n12\n")); + } + + /// A request the caller already sized keeps that value rather than being overwritten. + #[tokio::test] + async fn test_send_keeps_an_existing_content_length() { + let capturing_policy = Arc::new(HeaderCapturingPolicy::default()); + let policy = SharedKeyAuthorizationPolicy::new( + EMULATOR_ACCOUNT.to_owned(), + EMULATOR_ACCOUNT_KEY.to_owned(), + ); + let url = Url::parse("http://127.0.0.1:10000/devstoreaccount1/c/blob").unwrap(); + let mut request = Request::new(url, Method::Put); + request.insert_header(CONTENT_LENGTH, "5"); + request.set_body(azure_core::Bytes::from_static(b"hello")); + + let next: Vec> = vec![capturing_policy.clone()]; + policy + .send(&Context::default(), &mut request, &next) + .await + .unwrap(); + + let seen_headers = capturing_policy + .seen_headers + .lock() + .unwrap() + .clone() + .unwrap(); + assert_eq!(header_or_empty(&seen_headers, CONTENT_LENGTH), "5"); + } + + #[test] + fn test_debug_does_not_leak_the_key() { + let policy = SharedKeyAuthorizationPolicy::new( + EMULATOR_ACCOUNT.to_owned(), + EMULATOR_ACCOUNT_KEY.to_owned(), + ); + let rendered = format!("{policy:?}"); + assert!(rendered.contains(EMULATOR_ACCOUNT)); + assert!(!rendered.contains(EMULATOR_ACCOUNT_KEY)); + } +} diff --git a/quickwit/quickwit-storage/src/object_storage/mod.rs b/quickwit/quickwit-storage/src/object_storage/mod.rs index e914c107291..b19d66c343e 100644 --- a/quickwit/quickwit-storage/src/object_storage/mod.rs +++ b/quickwit/quickwit-storage/src/object_storage/mod.rs @@ -26,4 +26,8 @@ mod s3_compatible_storage_resolver; #[cfg(feature = "azure")] mod azure_blob_storage; #[cfg(feature = "azure")] +mod azure_credentials; +#[cfg(feature = "azure")] +mod azure_shared_key; +#[cfg(feature = "azure")] pub use self::azure_blob_storage::{AzureBlobStorage, AzureBlobStorageFactory}; diff --git a/quickwit/quickwit-storage/tests/azure_storage.rs b/quickwit/quickwit-storage/tests/azure_storage.rs index 1619596a822..a62fcd9f2c5 100644 --- a/quickwit/quickwit-storage/tests/azure_storage.rs +++ b/quickwit/quickwit-storage/tests/azure_storage.rs @@ -23,15 +23,13 @@ async fn azure_storage_test_suite() -> anyhow::Result<()> { use std::path::PathBuf; use anyhow::Context; - use azure_storage_blobs::prelude::ClientBuilder; use quickwit_common::rand::append_random_suffix; use quickwit_storage::{AzureBlobStorage, MultiPartPolicy}; let _ = tracing_subscriber::fmt::try_init(); // Setup container. let container_name = append_random_suffix("quickwit").to_lowercase(); - let container_client = ClientBuilder::emulator().container_client(&container_name); - container_client.create().into_future().await?; + AzureBlobStorage::create_emulated_container(&container_name).await?; let mut object_storage = AzureBlobStorage::new_emulated(&container_name); quickwit_storage::storage_test_suite(&mut object_storage).await?; @@ -56,6 +54,6 @@ async fn azure_storage_test_suite() -> anyhow::Result<()> { .context("test multipart upload failed")?; // Teardown container. - container_client.delete().into_future().await?; + AzureBlobStorage::delete_emulated_container(&container_name).await?; Ok(()) }