diff --git a/.cargo/config.toml b/.cargo/config.toml index ef954fd8fe73a..82df4ec732175 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -1,6 +1,12 @@ [alias] vdev = "run --quiet --package vdev --" +# The `sinks-iggy` feature depends on the private `obstack-wire` crate over SSH. +# Cargo's built-in git (libgit2) does not reuse ssh-agent/ssh-config the way the +# git CLI does, so delegate git fetches to the already-authenticated CLI. +[net] +git-fetch-with-cli = true + [env] # Build jemalloc assuming 64K pages so binaries also run on aarch64 hosts whose # kernels ship with a 64K page size (notably EL8 aarch64). On 4K-page hosts this diff --git a/.dockerignore b/.dockerignore index 02c60ccd5e905..abb2193cc5c6f 100644 --- a/.dockerignore +++ b/.dockerignore @@ -10,6 +10,10 @@ !rust-toolchain.toml !scripts !vdev +!distribution +!distribution/docker +!distribution/docker/obstack +!distribution/docker/obstack/entrypoint.sh # Exclude target directories to avoid copying build artifacts target diff --git a/Cargo.lock b/Cargo.lock index 606c3cb2882bc..7d9203a19fea2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -50,6 +50,20 @@ dependencies = [ "cpufeatures 0.3.0", ] +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead 0.5.2", + "aes 0.8.3", + "cipher 0.4.4", + "ctr 0.9.2", + "ghash", + "subtle", +] + [[package]] name = "aes-siv" version = "0.7.0" @@ -101,6 +115,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "aligned-vec" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc890384c8602f339876ded803c97ad529f3842aba97f6392b3dba0dd171769b" +dependencies = [ + "equator", +] + [[package]] name = "alloc-no-stdlib" version = "2.0.4" @@ -203,18 +226,33 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d96bd03f33fe50a863e394ee9718a706f988b9079b20c3784fb726e7678b62fb" dependencies = [ "anstyle", - "anstyle-parse", + "anstyle-parse 0.2.2", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "utf8parse", +] + +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse 1.0.0", "anstyle-query", "anstyle-wincon", "colorchoice", + "is_terminal_polyfill", "utf8parse", ] [[package]] name = "anstyle" -version = "1.0.8" +version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bec1de6f59aedf83baf9ff929c98f2ad654b97c9510f4e70cf6f661d49fd5b1" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" [[package]] name = "anstyle-parse" @@ -225,6 +263,15 @@ dependencies = [ "utf8parse", ] +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + [[package]] name = "anstyle-query" version = "1.0.0" @@ -236,12 +283,13 @@ dependencies = [ [[package]] name = "anstyle-wincon" -version = "3.0.1" +version = "3.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0699d10d2f4d628a98ee7b57b289abbc98ff3bad977cb3152709d4bf2330628" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", - "windows-sys 0.48.0", + "once_cell_polyfill", + "windows-sys 0.61.0", ] [[package]] @@ -362,6 +410,12 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "arrayref" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76a2e8124351fda1ef8aaaa3bbd7ebbcb486bbcd4225aca0aa0d84bb2db8fecb" + [[package]] name = "arrayvec" version = "0.7.4" @@ -621,6 +675,45 @@ dependencies = [ "term 1.0.1", ] +[[package]] +name = "asn1-rs" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom 7.1.3", + "num-traits", + "rusticata-macros", + "thiserror 2.0.18", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", + "synstructure", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", +] + [[package]] name = "assert-json-diff" version = "2.0.2" @@ -649,9 +742,9 @@ dependencies = [ [[package]] name = "async-broadcast" -version = "0.7.1" +version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "20cd0e2e25ea8e5f7e9df04578dc6cf5c83577fd09b1a46aaf5c85e1c33f2a7e" +checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532" dependencies = [ "event-listener 5.3.1", "event-listener-strategy", @@ -696,6 +789,45 @@ dependencies = [ "tokio", ] +[[package]] +name = "async-dropper" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d901072ae4dcdca2201b98beb02d31fb4b6b2472fbd0e870b12ec15b8b35b2d2" +dependencies = [ + "async-dropper-derive", + "async-dropper-simple", + "async-trait", + "futures", + "tokio", +] + +[[package]] +name = "async-dropper-derive" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a35cf17a37761f1c88b8e770b5956820fe84c12854165b6f930c604ea186e47e" +dependencies = [ + "async-trait", + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", + "tokio", +] + +[[package]] +name = "async-dropper-simple" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7c4748dfe8cd3d625ec68fc424fa80c134319881185866f9e173af9e5d8add8" +dependencies = [ + "async-scoped", + "async-trait", + "futures", + "rustc_version", + "tokio", +] + [[package]] name = "async-executor" version = "1.14.0" @@ -812,7 +944,7 @@ dependencies = [ "thiserror 2.0.18", "time", "tokio", - "tokio-rustls 0.26.2", + "tokio-rustls 0.26.4", "tokio-stream", "tokio-util", "tokio-websockets", @@ -878,6 +1010,17 @@ dependencies = [ "tokio-stream", ] +[[package]] +name = "async-scoped" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4042078ea593edffc452eef14e99fdb2b120caa4ad9618bcdeabc4a023b98740" +dependencies = [ + "futures", + "pin-project", + "tokio", +] + [[package]] name = "async-signal" version = "0.2.5" @@ -998,6 +1141,29 @@ dependencies = [ "zeroize", ] +[[package]] +name = "aws-lc-rs" +version = "1.17.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.43.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + [[package]] name = "aws-runtime" version = "1.7.2" @@ -1909,7 +2075,7 @@ dependencies = [ "bitflags 2.10.0", "cexpr", "clang-sys", - "itertools 0.10.5", + "itertools 0.13.0", "proc-macro2 1.0.106", "quote 1.0.45", "regex", @@ -1924,7 +2090,7 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "08807e080ed7f9d5433fa9b275196cfc35414f66a0c79d864dc51a0d825231a3" dependencies = [ - "bit-vec", + "bit-vec 0.8.0", ] [[package]] @@ -1933,6 +2099,15 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -1970,6 +2145,20 @@ dependencies = [ "wyz", ] +[[package]] +name = "blake3" +version = "1.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0aa83c34e62843d924f905e0f5c866eb1dd6545fc4d719e803d9ba6030371fce" +dependencies = [ + "arrayref", + "arrayvec", + "cc", + "cfg-if", + "constant_time_eq", + "cpufeatures 0.3.0", +] + [[package]] name = "block-buffer" version = "0.10.4" @@ -2042,7 +2231,7 @@ dependencies = [ "hyperlocal", "log", "pin-project-lite", - "rustls 0.23.37", + "rustls 0.23.42", "rustls-native-certs 0.8.1", "rustls-pki-types", "serde", @@ -2071,9 +2260,9 @@ dependencies = [ [[package]] name = "bon" -version = "3.8.1" +version = "3.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ebeb9aaf9329dff6ceb65c689ca3db33dbf15f324909c60e4e5eef5701ce31b1" +checksum = "a602c73c7b0148ec6d12af6fd5cc7a46e2eacc8878271a999abac56eed12f561" dependencies = [ "bon-macros", "rustversion", @@ -2081,11 +2270,11 @@ dependencies = [ [[package]] name = "bon-macros" -version = "3.8.1" +version = "3.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77e9d642a7e3a318e37c2c9427b5a6a48aa1ad55dcd986f3034ab2239045a645" +checksum = "6dee98b0db6a962de883bf5d20362dee4d7ca0d12fe39a7c6c73c844e1cd7c1f" dependencies = [ - "darling 0.21.3", + "darling 0.23.0", "ident_case", "prettyplease", "proc-macro2 1.0.106", @@ -2204,6 +2393,18 @@ version = "3.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f30e7476521f6f8af1a1c4c0b8cc94f0bee37d91763d0ca2665f299b6cd8aec" +[[package]] +name = "byte-unit" +version = "5.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a813de7f2bbedb7dce265b64f1cf5908ebe4d56281ece8d847e98113788b9b0" +dependencies = [ + "rust_decimal", + "schemars 1.0.3", + "serde", + "utf8-width", +] + [[package]] name = "bytecheck" version = "0.6.11" @@ -2234,9 +2435,23 @@ checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e" [[package]] name = "bytemuck" -version = "1.21.0" +version = "1.25.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef657dfab802224e671f5818e9a4935f9b1957ed18e58292690cc39e7a4092a3" +checksum = "95832e849adfb21180ccb6826a99da14e5d266ae5c2e668e1602cf234f153797" +dependencies = [ + "bytemuck_derive", +] + +[[package]] +name = "bytemuck_derive" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f65693059b6b9c588b9f62fed1cedbf0a8b805631457ea162d68f0de186f3de5" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", +] [[package]] name = "byteorder" @@ -2480,9 +2695,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.60" +version = "4.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2797f34da339ce31042b27d23607e051786132987f595b02ba4f6a6dffb7030a" +checksum = "d91e0c145792ef73a6ad36d27c75ac09f1832222a3c209689d90f534685ee5b7" dependencies = [ "clap_builder", "clap_derive", @@ -2500,11 +2715,11 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.5.60" +version = "4.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24a241312cea5059b13574bb9b3861cabf758b879c15190b37b6d6fd63ab6876" +checksum = "f09628afdcc538b57f3c6341e9c8e9970f18e4a481690a64974d7023bd33548b" dependencies = [ - "anstream", + "anstream 1.0.0", "anstyle", "clap_lex", "strsim", @@ -2522,14 +2737,14 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.5.55" +version = "4.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a92793da1a46a5f2a02a6f4c46c6496b28c43638adea8306fcb0caa1634f24e5" +checksum = "d012d2b9d65aca7f18f4d9878a045bc17899bba951561ba5ec3c2ba1eed9a061" dependencies = [ "heck 0.5.0", "proc-macro2 1.0.106", "quote 1.0.45", - "syn 2.0.117", + "syn 3.0.3", ] [[package]] @@ -2706,6 +2921,208 @@ dependencies = [ "static_assertions", ] +[[package]] +name = "compio" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b84ee96a86948d04388f3a0b8c36b9f0a6b40b3528ac0d65737e53632fb37fe" +dependencies = [ + "compio-buf", + "compio-driver", + "compio-fs", + "compio-io", + "compio-log", + "compio-macros", + "compio-net", + "compio-quic", + "compio-runtime", + "compio-tls", + "compio-ws", +] + +[[package]] +name = "compio-buf" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b51a2c35873865376ed4cdb6cfeb602b6cf569815f017ddd7bd8f86ad49b34c7" +dependencies = [ + "arrayvec", + "bytes", + "libc", +] + +[[package]] +name = "compio-driver" +version = "0.11.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "74d42d98dc890ee4db00c1e68a723391711aab6d67085880d716b72830f7c715" +dependencies = [ + "cfg-if", + "cfg_aliases", + "compio-buf", + "compio-log", + "crossbeam-queue", + "flume 0.12.0", + "futures-util", + "io-uring", + "io_uring_buf_ring", + "libc", + "once_cell", + "paste", + "pin-project-lite", + "polling", + "slab", + "smallvec", + "socket2 0.6.3", + "synchrony", + "thin-cell", + "windows-sys 0.61.0", +] + +[[package]] +name = "compio-fs" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65ee36e1acf2cec4835efe9a986c012b2462c5ef53580e4ee84ae6d5a3d8e3b3" +dependencies = [ + "cfg-if", + "cfg_aliases", + "compio-buf", + "compio-driver", + "compio-io", + "compio-runtime", + "libc", + "os_pipe", + "pin-project-lite", + "widestring 1.0.2", + "windows-sys 0.61.0", +] + +[[package]] +name = "compio-io" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "637522f28a64fd5f7dcceaa4ddef13fa8d8020025e8c993f7a069e237835580e" +dependencies = [ + "compio-buf", + "futures-util", + "paste", + "synchrony", +] + +[[package]] +name = "compio-log" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc4e560213c1996b618da369b7c9109564b41af9033802ae534465c4ee4e132f" +dependencies = [ + "tracing 0.1.44", +] + +[[package]] +name = "compio-macros" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f05ed201484967dc70de77a8f7a02b29aaa8e6c81cbea2e75492ee0c8d97766b" +dependencies = [ + "proc-macro-crate 3.2.0", + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", +] + +[[package]] +name = "compio-net" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "becd7d40522c885113752a3640cba9f9d347f205b646bb3f8ff3967173a228f2" +dependencies = [ + "cfg-if", + "compio-buf", + "compio-driver", + "compio-io", + "compio-runtime", + "either", + "libc", + "once_cell", + "socket2 0.6.3", + "widestring 1.0.2", + "windows-sys 0.61.0", +] + +[[package]] +name = "compio-quic" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad9efdad81b920108b9de57148e1b9d73dc408b6d06a59ee64836dde651cf026" +dependencies = [ + "cfg_aliases", + "compio-buf", + "compio-io", + "compio-log", + "compio-net", + "compio-runtime", + "flume 0.12.0", + "futures-util", + "libc", + "quinn-proto", + "rustc-hash", + "rustls 0.23.42", + "synchrony", + "thiserror 2.0.18", + "windows-sys 0.61.0", +] + +[[package]] +name = "compio-runtime" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6c1c71f011bdd9c8f30e97d877b606505ee6d241c7782cfaed172f66acbd9cd" +dependencies = [ + "async-task", + "cfg-if", + "compio-buf", + "compio-driver", + "compio-log", + "core_affinity", + "crossbeam-queue", + "futures-util", + "libc", + "once_cell", + "pin-project-lite", + "scoped-tls", + "slab", + "socket2 0.6.3", + "windows-sys 0.61.0", +] + +[[package]] +name = "compio-tls" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a7056da226af42cda4c83b00a021cce3e1ee5f4cffc8a0ff8801381e618cf1c" +dependencies = [ + "compio-buf", + "compio-io", + "futures-rustls", + "futures-util", + "rustls 0.23.42", +] + +[[package]] +name = "compio-ws" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "99d45f47c6e64babcaa6b8df1dffced56012e60e58401255e679f428ddbe9fb6" +dependencies = [ + "compio-buf", + "compio-io", + "compio-log", + "compio-net", + "compio-tls", + "tungstenite 0.28.0", +] + [[package]] name = "compression-codecs" version = "0.4.38" @@ -2837,6 +3254,12 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "18f12cc9948ed9604230cdddc7c86e270f9401ccbe3c2e98a4378c5e7632212f" +[[package]] +name = "constant_time_eq" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" + [[package]] name = "convert_case" version = "0.7.1" @@ -2925,6 +3348,17 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "core_affinity" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a034b3a7b624016c6e13f5df875747cc25f884156aad2abd12b6c46797971342" +dependencies = [ + "libc", + "num_cpus", + "winapi", +] + [[package]] name = "cpubits" version = "0.1.1" @@ -3027,23 +3461,34 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" +[[package]] +name = "crossbeam" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1137cd7e7fc0fb5d3c5a8678be38ec56e819125d8d7907411fe24ccb943faca8" +dependencies = [ + "crossbeam-channel", + "crossbeam-deque", + "crossbeam-epoch", + "crossbeam-queue", + "crossbeam-utils", +] + [[package]] name = "crossbeam-channel" -version = "0.5.8" +version = "0.5.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a33c2bf77f2df06183c3aa30d1e96c0695a313d4f9c453cc3762a6db39f99200" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" dependencies = [ - "cfg-if", "crossbeam-utils", ] [[package]] name = "crossbeam-deque" -version = "0.8.3" +version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce6fd6f855243022dcecf8702fef0c297d4338e226845fe067f6341ad9fa0cef" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" dependencies = [ - "cfg-if", "crossbeam-epoch", "crossbeam-utils", ] @@ -3299,7 +3744,6 @@ dependencies = [ "ident_case", "proc-macro2 1.0.106", "quote 1.0.45", - "strsim", "syn 2.0.117", ] @@ -3489,6 +3933,20 @@ dependencies = [ "zeroize", ] +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom 7.1.3", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.5" @@ -3803,6 +4261,12 @@ dependencies = [ "shared_child", ] +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "dyn-clone" version = "1.0.20" @@ -3974,29 +4438,50 @@ dependencies = [ name = "enum_dispatch" version = "0.3.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aa18ce2bc66555b3218614519ac839ddb759a7d6720732f979ef8d13be147ecd" +checksum = "aa18ce2bc66555b3218614519ac839ddb759a7d6720732f979ef8d13be147ecd" +dependencies = [ + "once_cell", + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", +] + +[[package]] +name = "enumflags2" +version = "0.7.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1027f7680c853e056ebcec683615fb6fbbc07dbaa13b4d5d9442b146ded4ecef" +dependencies = [ + "enumflags2_derive", +] + +[[package]] +name = "enumflags2_derive" +version = "0.7.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67c78a4d8fdf9953a5c9d458f9efe940fd97a0cab0941c075a813ac594733827" dependencies = [ - "once_cell", "proc-macro2 1.0.106", "quote 1.0.45", "syn 2.0.117", ] [[package]] -name = "enumflags2" -version = "0.7.12" +name = "enumset" +version = "1.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1027f7680c853e056ebcec683615fb6fbbc07dbaa13b4d5d9442b146ded4ecef" +checksum = "ccc5801fd11762e24d1e420d01d2ac518f2a2ca4329d4fbb6639f2412b6204e0" dependencies = [ - "enumflags2_derive", + "enumset_derive", ] [[package]] -name = "enumflags2_derive" -version = "0.7.12" +name = "enumset_derive" +version = "0.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67c78a4d8fdf9953a5c9d458f9efe940fd97a0cab0941c075a813ac594733827" +checksum = "4bd536557b58c682b217b8fb199afdff47cd3eff260623f19e77074eb073d63a" dependencies = [ + "darling 0.21.3", "proc-macro2 1.0.106", "quote 1.0.45", "syn 2.0.117", @@ -4024,13 +4509,33 @@ version = "0.11.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b2daee4ea451f429a58296525ddf28b45a3b64f1acf6587e2067437bb11e218d" dependencies = [ - "anstream", + "anstream 0.6.13", "anstyle", "env_filter", "jiff", "log", ] +[[package]] +name = "equator" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4711b213838dfee0117e3be6ac926007d7f433d7bbe33595975d4190cb07e6fc" +dependencies = [ + "equator-macro", +] + +[[package]] +name = "equator-macro" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44f23cf4b44bfce11a86ace86f8a73ffdec849c9fd00a386a53d278bd9e81fb3" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", +] + [[package]] name = "equivalent" version = "1.0.1" @@ -4046,6 +4551,27 @@ dependencies = [ "serde", ] +[[package]] +name = "err_trail" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b044506df3479abb857a3e1514fea858db128ea6fd95cc26088d9cf748f2a" +dependencies = [ + "err_trail_macros", + "tracing 0.1.44", +] + +[[package]] +name = "err_trail_macros" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d9ab0367afaae0ba1538512a0360b2b9fc737089a48612334200d815c7982f9" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", +] + [[package]] name = "errno" version = "0.3.10" @@ -4137,6 +4663,35 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "de853764b47027c2e862a995c34978ffa63c1501f2e15f987ba11bd4f9bba193" +[[package]] +name = "ext-trait" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d772df1c1a777963712fb68e014235e80863d6a91a85c4e06ba2d16243a310e5" +dependencies = [ + "ext-trait-proc_macros", +] + +[[package]] +name = "ext-trait-proc_macros" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ab7934152eaf26aa5aa9f7371408ad5af4c31357073c9e84c3b9d7f11ad639a" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 1.0.109", +] + +[[package]] +name = "extension-traits" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a296e5a895621edf9fa8329c83aa1cb69a964643e36cf54d8d7a69b789089537" +dependencies = [ + "ext-trait", +] + [[package]] name = "fakedata" version = "0.1.0" @@ -4173,11 +4728,26 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "fastbloom" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef975e30683b2d965054bb0a836f8973857c4ebf6acf274fe46617cd285060d8" +dependencies = [ + "foldhash 0.2.0", + "libm", + "portable-atomic", + "siphasher 1.0.1", +] + [[package]] name = "fastrand" version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" +dependencies = [ + "getrandom 0.2.15", +] [[package]] name = "ff" @@ -4329,6 +4899,7 @@ version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ + "fastrand", "futures-core", "futures-sink", "spin 0.9.8", @@ -4386,6 +4957,12 @@ dependencies = [ "num", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "fsevent-sys" version = "4.1.0" @@ -4494,6 +5071,17 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "futures-rustls" +version = "0.26.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8f2f12607f92c69b12ed746fabf9ca4f5c482cba46679c1a75b874ed7c26adb" +dependencies = [ + "futures-io", + "rustls 0.23.42", + "rustls-pki-types", +] + [[package]] name = "futures-sink" version = "0.3.32" @@ -4529,6 +5117,21 @@ dependencies = [ "slab", ] +[[package]] +name = "generator" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3b854b0e584ead1a33f18b2fcad7cf7be18b3875c78816b753639aa501513ae" +dependencies = [ + "cc", + "cfg-if", + "libc", + "log", + "rustversion", + "windows-link 0.2.0", + "windows-result 0.3.4", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -4583,6 +5186,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + [[package]] name = "git2" version = "0.20.4" @@ -5294,6 +5907,12 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "human-repr" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f58b778a5761513caf593693f8951c97a5b610841e754788400f32102eefdff1" + [[package]] name = "humantime" version = "2.3.0" @@ -5372,7 +5991,7 @@ dependencies = [ "pin-project-lite", "rustls-native-certs 0.7.0", "tokio", - "tokio-rustls 0.26.2", + "tokio-rustls 0.26.4", "tower-service", ] @@ -5470,11 +6089,11 @@ dependencies = [ "http 1.3.1", "hyper 1.7.0", "hyper-util", - "rustls 0.23.37", + "rustls 0.23.42", "rustls-native-certs 0.8.1", "rustls-pki-types", "tokio", - "tokio-rustls 0.26.2", + "tokio-rustls 0.26.4", "tower-service", "webpki-roots 0.26.1", ] @@ -5746,6 +6365,102 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "iggy" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a470a78ccd8a6602402817906d842a3f361f84516cb880282565365df4eba2b" +dependencies = [ + "async-broadcast", + "async-dropper", + "async-trait", + "bon", + "bytes", + "dashmap", + "flume 0.12.0", + "futures", + "futures-util", + "iggy_common", + "quinn", + "reqwest 0.13.3", + "reqwest-middleware 0.5.2", + "reqwest-retry 0.9.1", + "reqwest-tracing", + "rustls 0.23.42", + "secrecy", + "serde", + "tokio", + "tokio-rustls 0.26.4", + "tokio-tungstenite 0.29.0", + "tracing 0.1.44", + "trait-variant", + "tungstenite 0.29.0", + "webpki-roots 1.0.9", +] + +[[package]] +name = "iggy_binary_protocol" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e6da7f3a07797ef6a4248400bdf9d752e0eb7d83d14bf9cadcaae6bffdc9235" +dependencies = [ + "aligned-vec", + "bytemuck", + "bytes", + "compio-buf", + "enumset", + "smallvec", + "thiserror 2.0.18", +] + +[[package]] +name = "iggy_common" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6193adb66b1f12b6b2f9725573a6a6f69ce714c49830e122ed0d5ce7e7ffbde8" +dependencies = [ + "aes-gcm", + "aligned-vec", + "async-broadcast", + "async-trait", + "base64 0.22.1", + "blake3", + "bon", + "byte-unit", + "bytemuck", + "bytes", + "chrono", + "clap", + "comfy-table", + "compio", + "crossbeam", + "derive_more", + "err_trail", + "human-repr", + "humantime", + "iggy_binary_protocol", + "lending-iterator", + "moka", + "nix 0.31.2", + "once_cell", + "papaya", + "rcgen", + "ring", + "rustls 0.23.42", + "secrecy", + "serde", + "serde_json", + "serde_with", + "strum 0.28.0", + "thiserror 2.0.18", + "tokio", + "tracing 0.1.44", + "tungstenite 0.29.0", + "twox-hash", + "ulid", + "uuid", +] + [[package]] name = "indexmap" version = "1.9.3" @@ -5869,6 +6584,28 @@ dependencies = [ "rustversion", ] +[[package]] +name = "io-uring" +version = "0.7.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9080b15e63775b9a2ac7dca720f7050a8b955e092ea0f6020a4a80f69998cdc0" +dependencies = [ + "bitflags 2.10.0", + "cfg-if", + "libc", +] + +[[package]] +name = "io_uring_buf_ring" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1838759bb8c2f24cf05a35429d83145c4aa6af43f8ad38477295e12a7320a80e" +dependencies = [ + "bytes", + "io-uring", + "rustix 1.1.4", +] + [[package]] name = "ipconfig" version = "0.3.2" @@ -5932,6 +6669,12 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7655c9839580ee829dfacba1d1278c2b7883e50a277ff7541299489d6bdfdc45" +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + [[package]] name = "itertools" version = "0.10.5" @@ -6415,6 +7158,31 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" +[[package]] +name = "lending-iterator" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc07588c853b50689205fb5c00498aa681d89828e0ce8cbd965ebc7a5d8ae260" +dependencies = [ + "extension-traits", + "lending-iterator-proc_macros", + "macro_rules_attribute", + "never-say-never", + "nougat", + "polonius-the-crab", +] + +[[package]] +name = "lending-iterator-proc_macros" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5445dd1c0deb1e97b8a16561d17fc686ca83e8411128fb036e9668a72d51b1d" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 1.0.109", +] + [[package]] name = "lexical-core" version = "1.0.6" @@ -6663,6 +7431,20 @@ dependencies = [ "prost-types 0.12.6", ] +[[package]] +name = "loom" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca" +dependencies = [ + "cfg-if", + "generator", + "pin-utils", + "scoped-tls", + "tracing 0.1.44", + "tracing-subscriber", +] + [[package]] name = "lru" version = "0.16.3" @@ -6788,6 +7570,22 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "macro_rules_attribute" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf0c9b980bf4f3a37fd7b1c066941dd1b1d0152ce6ee6e8fe8c49b9f6810d862" +dependencies = [ + "macro_rules_attribute-proc_macro", + "paste", +] + +[[package]] +name = "macro_rules_attribute-proc_macro" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "58093314a45e00c77d5c508f76e77c3396afbbc0d01506e7fae47b018bac2b1d" + [[package]] name = "malloc_buf" version = "0.0.6" @@ -7065,21 +7863,21 @@ checksum = "dce6dd36094cac388f119d2e9dc82dc730ef91c32a6222170d630e5414b956e6" [[package]] name = "moka" -version = "0.12.7" +version = "0.12.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e0d88686dc561d743b40de8269b26eaf0dc58781bde087b0984646602021d08" +checksum = "957228ad12042ee839f93c8f257b62b4c0ab5eaae1d4fa60de53b27c9d7c5046" dependencies = [ + "async-lock 3.4.0", "crossbeam-channel", "crossbeam-epoch", "crossbeam-utils", - "once_cell", + "equivalent", + "event-listener 5.3.1", + "futures-util", "parking_lot", - "quanta", - "rustc_version", + "portable-atomic", "smallvec", "tagptr", - "thiserror 1.0.68", - "triomphe", "uuid", ] @@ -7128,7 +7926,7 @@ dependencies = [ "percent-encoding", "rand 0.9.4", "rustc_version_runtime", - "rustls 0.23.37", + "rustls 0.23.42", "serde", "serde_bytes", "serde_with", @@ -7140,11 +7938,11 @@ dependencies = [ "take_mut", "thiserror 2.0.18", "tokio", - "tokio-rustls 0.26.2", + "tokio-rustls 0.26.4", "tokio-util", "typed-builder", "uuid", - "webpki-roots 1.0.4", + "webpki-roots 1.0.9", ] [[package]] @@ -7224,6 +8022,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b" +[[package]] +name = "never-say-never" +version = "6.6.666" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf5a574dadd7941adeaa71823ecba5e28331b8313fb2e1c6a5c7e5981ea53ad6" + [[package]] name = "new_debug_unreachable" version = "1.0.4" @@ -7384,6 +8188,27 @@ version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e0826a989adedc2a244799e823aece04662b66609d96af8dff7ac6df9a8925d" +[[package]] +name = "nougat" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97b57b9ced431322f054fc673f1d3c7fa52d80efd9df74ad2fc759f044742510" +dependencies = [ + "macro_rules_attribute", + "nougat-proc_macros", +] + +[[package]] +name = "nougat-proc_macros" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c84f77a45e99a2f9b492695d99e1c23844619caa5f3e57647cffacad773ca257" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 1.0.109", +] + [[package]] name = "ntapi" version = "0.3.7" @@ -7653,6 +8478,18 @@ dependencies = [ "objc2-core-foundation", ] +[[package]] +name = "obstack-wire" +version = "0.1.0" +source = "git+ssh://git@github.com/alexpacio/obstack-wire.git?rev=817345d4cf563098d3b7e6ff49821231ee08a86d#817345d4cf563098d3b7e6ff49821231ee08a86d" +dependencies = [ + "blake3", + "rmp-serde", + "serde", + "thiserror 2.0.18", + "xxhash-rust", +] + [[package]] name = "octseq" version = "0.6.1" @@ -7673,16 +8510,31 @@ dependencies = [ "cipher 0.5.2", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" -version = "1.21.3" +version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" dependencies = [ "critical-section", "portable-atomic", ] +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + [[package]] name = "onig" version = "6.5.1" @@ -7938,6 +8790,16 @@ dependencies = [ "winapi", ] +[[package]] +name = "papaya" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "997ee03cd38c01469a7046643714f0ad28880bcb9e6679ff0666e24817ca19b7" +dependencies = [ + "equivalent", + "seize", +] + [[package]] name = "parking" version = "2.2.0" @@ -8188,9 +9050,9 @@ dependencies = [ [[package]] name = "pin-project-lite" -version = "0.2.14" +version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bda66fc9667c18cb2758a2ac84d1167245054bcf85d5d1aaa6923f45801bdd02" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" [[package]] name = "pin-utils" @@ -8285,6 +9147,12 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "polonius-the-crab" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a69ee997a6282f8462abf1e0d8c38c965e968799e912b3bed8c9e8a28c2f9f" + [[package]] name = "poly1305" version = "0.8.0" @@ -8306,6 +9174,18 @@ dependencies = [ "universal-hash 0.6.1", ] +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.11", + "opaque-debug", + "universal-hash 0.5.1", +] + [[package]] name = "portable-atomic" version = "1.13.1" @@ -8574,7 +9454,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744" dependencies = [ "bit-set", - "bit-vec", + "bit-vec 0.8.0", "bitflags 2.10.0", "num-traits", "rand 0.9.4", @@ -8666,7 +9546,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck 0.5.0", - "itertools 0.10.5", + "itertools 0.14.0", "log", "multimap", "once_cell", @@ -8686,7 +9566,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ "heck 0.5.0", - "itertools 0.10.5", + "itertools 0.14.0", "log", "multimap", "petgraph 0.8.3", @@ -8720,7 +9600,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2 1.0.106", "quote 1.0.45", "syn 2.0.117", @@ -8733,7 +9613,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2 1.0.106", "quote 1.0.45", "syn 2.0.117", @@ -9052,36 +9932,42 @@ dependencies = [ [[package]] name = "quinn" -version = "0.11.6" +version = "0.11.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62e96808277ec6f97351a2380e6c25114bc9e67037775464979f3037c92d05ef" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" dependencies = [ "bytes", + "cfg_aliases", "pin-project-lite", "quinn-proto", "quinn-udp", "rustc-hash", - "rustls 0.23.37", - "socket2 0.5.10", + "rustls 0.23.42", + "socket2 0.6.3", "thiserror 2.0.18", "tokio", "tracing 0.1.44", + "web-time", ] [[package]] name = "quinn-proto" -version = "0.11.14" +version = "0.11.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" +checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" dependencies = [ + "aws-lc-rs", "bytes", - "getrandom 0.3.4", + "fastbloom", + "getrandom 0.4.2", "lru-slab", - "rand 0.9.4", + "rand 0.10.1", + "rand_pcg", "ring", "rustc-hash", - "rustls 0.23.37", + "rustls 0.23.42", "rustls-pki-types", + "rustls-platform-verifier", "slab", "thiserror 2.0.18", "tinyvec", @@ -9242,6 +10128,15 @@ dependencies = [ "rand 0.9.4", ] +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core 0.10.0", +] + [[package]] name = "rand_xorshift" version = "0.4.0" @@ -9355,6 +10250,20 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rcgen" +version = "0.14.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57f6d249aad744e274e682777a50283a225a32705394ee6d5fcc01efa25e4055" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", +] + [[package]] name = "rdkafka" version = "0.39.0" @@ -9648,7 +10557,7 @@ dependencies = [ "percent-encoding", "pin-project-lite", "quinn", - "rustls 0.23.37", + "rustls 0.23.42", "rustls-native-certs 0.8.1", "rustls-pki-types", "serde", @@ -9657,7 +10566,7 @@ dependencies = [ "sync_wrapper 1.0.2", "tokio", "tokio-native-tls", - "tokio-rustls 0.26.2", + "tokio-rustls 0.26.4", "tokio-util", "tower 0.5.3", "tower-http 0.6.8", @@ -9667,7 +10576,7 @@ dependencies = [ "wasm-bindgen-futures", "wasm-streams 0.4.0", "web-sys", - "webpki-roots 1.0.4", + "webpki-roots 1.0.9", ] [[package]] @@ -9692,15 +10601,17 @@ dependencies = [ "native-tls", "percent-encoding", "pin-project-lite", - "rustls 0.23.37", + "quinn", + "rustls 0.23.42", "rustls-pki-types", "rustls-platform-verifier", "serde", "serde_json", + "serde_urlencoded", "sync_wrapper 1.0.2", "tokio", "tokio-native-tls", - "tokio-rustls 0.26.2", + "tokio-rustls 0.26.4", "tokio-util", "tower 0.5.3", "tower-http 0.6.8", @@ -9727,6 +10638,21 @@ dependencies = [ "tower-service", ] +[[package]] +name = "reqwest-middleware" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bc3f1384cffa4f274dad2d4ddd73aed32fed8f786d96c6be8aa4e5fd3c3b58" +dependencies = [ + "anyhow", + "async-trait", + "http 1.3.1", + "reqwest 0.13.3", + "serde", + "thiserror 2.0.18", + "tower-service", +] + [[package]] name = "reqwest-retry" version = "0.8.0" @@ -9740,13 +10666,50 @@ dependencies = [ "http 1.3.1", "hyper 1.7.0", "reqwest 0.12.28", - "reqwest-middleware", + "reqwest-middleware 0.4.2", + "retry-policies", + "thiserror 2.0.18", + "tokio", + "wasmtimer", +] + +[[package]] +name = "reqwest-retry" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe2412db2af7d2268e7a5406be0431f37d9eb67ff390f35b395716f5f06c2eaa" +dependencies = [ + "anyhow", + "async-trait", + "futures", + "getrandom 0.2.15", + "http 1.3.1", + "hyper 1.7.0", + "reqwest 0.13.3", + "reqwest-middleware 0.5.2", "retry-policies", "thiserror 2.0.18", "tokio", + "tracing 0.1.44", "wasmtimer", ] +[[package]] +name = "reqwest-tracing" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5e5af0cd6fc3d3c8f703d597af70b6e4e62432c63157b49419fa1ffaf481702" +dependencies = [ + "anyhow", + "async-trait", + "getrandom 0.2.15", + "http 1.3.1", + "matchit 0.8.4", + "reqwest 0.13.3", + "reqwest-middleware 0.5.2", + "tracing 0.1.44", +] + [[package]] name = "resolv-conf" version = "0.7.5" @@ -9972,6 +10935,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom 7.1.3", +] + [[package]] name = "rustix" version = "0.38.40" @@ -10026,10 +10998,11 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.37" +version = "0.23.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4" +checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" dependencies = [ + "aws-lc-rs", "log", "once_cell", "ring", @@ -10104,7 +11077,7 @@ dependencies = [ "jni 0.22.4", "log", "once_cell", - "rustls 0.23.37", + "rustls 0.23.42", "rustls-native-certs 0.8.1", "rustls-platform-verifier-android", "rustls-webpki 0.103.13", @@ -10147,6 +11120,7 @@ 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", @@ -10322,6 +11296,7 @@ version = "0.10.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e891af845473308773346dc847b2c23ee78fe442e0472ac50e22a18a93d3ae5a" dependencies = [ + "serde", "zeroize", ] @@ -10361,6 +11336,16 @@ dependencies = [ "libc", ] +[[package]] +name = "seize" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521" +dependencies = [ + "libc", + "windows-sys 0.61.0", +] + [[package]] name = "semver" version = "1.0.28" @@ -11051,7 +12036,7 @@ dependencies = [ "memchr", "once_cell", "percent-encoding", - "rustls 0.23.37", + "rustls 0.23.42", "serde", "serde_json", "sha2 0.10.9", @@ -11402,6 +12387,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "0.1.2" @@ -11417,6 +12413,16 @@ dependencies = [ "futures-core", ] +[[package]] +name = "synchrony" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d6d5fbc4583cf3e5eee953506f13853a42ee6b4a21a983dffffb10c765cb80" +dependencies = [ + "futures-util", + "loom", +] + [[package]] name = "synstructure" version = "0.13.1" @@ -11600,12 +12606,12 @@ dependencies = [ [[package]] name = "terminal_size" -version = "0.4.0" +version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f599bd7ca042cfdf8f4512b277c02ba102247820f9d9d4a9f521f496751a6ef" +checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ - "rustix 0.38.40", - "windows-sys 0.59.0", + "rustix 1.1.4", + "windows-sys 0.61.0", ] [[package]] @@ -11626,6 +12632,12 @@ dependencies = [ "syn 0.15.44", ] +[[package]] +name = "thin-cell" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4164c6c316ba9733b0ab021e7f9852c788a4b991b49c25820f1be48e1d41345b" + [[package]] name = "thiserror" version = "1.0.68" @@ -11903,11 +12915,11 @@ dependencies = [ [[package]] name = "tokio-rustls" -version = "0.26.2" +version = "0.26.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e727b36a1a0e8b74c376ac2211e40c2c8af09fb4013c60d910495810f008e9b" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" dependencies = [ - "rustls 0.23.37", + "rustls 0.23.42", "tokio", ] @@ -11943,7 +12955,23 @@ dependencies = [ "futures-util", "log", "tokio", - "tungstenite", + "tungstenite 0.20.1", +] + +[[package]] +name = "tokio-tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" +dependencies = [ + "futures-util", + "log", + "rustls 0.23.42", + "rustls-pki-types", + "tokio", + "tokio-rustls 0.26.4", + "tungstenite 0.29.0", + "webpki-roots 0.26.1", ] [[package]] @@ -11978,7 +13006,7 @@ dependencies = [ "ring", "rustls-pki-types", "tokio", - "tokio-rustls 0.26.2", + "tokio-rustls 0.26.4", "tokio-util", "webpki-roots 0.26.1", ] @@ -12157,7 +13185,7 @@ dependencies = [ "socket2 0.6.3", "sync_wrapper 1.0.2", "tokio", - "tokio-rustls 0.26.2", + "tokio-rustls 0.26.4", "tokio-stream", "tower 0.5.3", "tower-layer", @@ -12523,10 +13551,15 @@ dependencies = [ ] [[package]] -name = "triomphe" -version = "0.1.12" +name = "trait-variant" +version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b2cb4fbb9995eeb36ac86fadf24031ccd58f99d6b4b2d7b911db70bddb80d90" +checksum = "b19a4867a870f6edc4c283f2b455804b1879c0baf0e642f26b03ed8ee262d9d3" +dependencies = [ + "proc-macro2 1.0.106", + "quote 1.0.45", + "syn 2.0.117", +] [[package]] name = "try-lock" @@ -12563,11 +13596,51 @@ dependencies = [ "utf-8", ] +[[package]] +name = "tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +dependencies = [ + "bytes", + "data-encoding", + "http 1.3.1", + "httparse", + "log", + "rand 0.9.4", + "rustls 0.23.42", + "rustls-pki-types", + "sha1 0.10.6", + "thiserror 2.0.18", + "utf-8", +] + +[[package]] +name = "tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" +dependencies = [ + "bytes", + "data-encoding", + "http 1.3.1", + "httparse", + "log", + "rand 0.9.4", + "rustls 0.23.42", + "rustls-pki-types", + "sha1 0.10.6", + "thiserror 2.0.18", +] + [[package]] name = "twox-hash" version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ea3136b675547379c4bd395ca6b938e5ad3c3d20fad76e7fe85f9e0d011419c" +dependencies = [ + "rand 0.9.4", +] [[package]] name = "typed-builder" @@ -12687,6 +13760,16 @@ version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed646292ffc8188ef8ea4d1e0e0150fb15a5c2e12ad9b8fc191ae7a8a7f3c4b9" +[[package]] +name = "ulid" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "470dbf6591da1b39d43c14523b2b469c86879a53e8b758c8e090a470fe7b1fbe" +dependencies = [ + "rand 0.9.4", + "web-time", +] + [[package]] name = "unarray" version = "0.1.4" @@ -12863,21 +13946,22 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" [[package]] name = "utf8parse" -version = "0.2.1" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "uuid" -version = "1.22.0" +version = "1.24.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a68d3c8f01c0cfa54a75291d83601161799e4a89a39e0929f4b0354d88757a37" +checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" dependencies = [ "getrandom 0.4.2", "js-sys", "rand 0.10.1", "serde_core", "wasm-bindgen", + "zerocopy 0.8.16", ] [[package]] @@ -13033,6 +14117,7 @@ dependencies = [ "hyper 0.14.32", "hyper-openssl 0.9.2", "hyper-proxy", + "iggy", "indexmap 2.12.0", "indoc", "inventory", @@ -13056,6 +14141,7 @@ dependencies = [ "nkeys", "nom 8.0.0", "notify", + "obstack-wire", "opendal", "openssl", "openssl-probe", @@ -13120,7 +14206,7 @@ dependencies = [ "tokio-postgres", "tokio-stream", "tokio-test", - "tokio-tungstenite", + "tokio-tungstenite 0.20.1", "tokio-util", "toml", "tonic 0.11.0", @@ -13672,8 +14758,8 @@ dependencies = [ "regex", "relative-path 2.0.1", "reqwest 0.12.28", - "reqwest-middleware", - "reqwest-retry", + "reqwest-middleware 0.4.2", + "reqwest-retry 0.8.0", "roxmltree", "rust_decimal", "rustyline", @@ -14010,9 +15096,9 @@ dependencies = [ [[package]] name = "webpki-roots" -version = "1.0.4" +version = "1.0.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b2878ef029c47c6e8cf779119f20fcf52bde7ad42a731b2a304bc221df17571e" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" dependencies = [ "rustls-pki-types", ] @@ -14765,6 +15851,24 @@ dependencies = [ "tap", ] +[[package]] +name = "x509-parser" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom 7.1.3", + "oid-registry", + "ring", + "rusticata-macros", + "thiserror 2.0.18", + "time", +] + [[package]] name = "xmlparser" version = "0.13.6" @@ -14777,6 +15881,16 @@ version = "0.8.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fdd20c5420375476fbd4394763288da7eb0cc0b8c11deed431a91562af7335d3" +[[package]] +name = "yasna" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282" +dependencies = [ + "bit-vec 0.9.1", + "time", +] + [[package]] name = "yoke" version = "0.7.4" diff --git a/Cargo.toml b/Cargo.toml index 28e7a4d2e8f0d..139367d1b5c0b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -345,6 +345,13 @@ serde_yaml.workspace = true rmp-serde = { version = "1.3.0", default-features = false, optional = true } rmpv = { version = "1.3.0", default-features = false, features = ["with-serde"], optional = true } +# Obstack `iggy` sink: Apache Iggy client + the shared on-wire contract. The +# label/row model, shard placement, and envelope codec (previously vendored in +# src/sinks/iggy/proto.rs) live in obstack-wire so producer and consumer share +# one definition; keep the rev in lock-step with obstack's Cargo.toml. +iggy = { version = "=0.10.0", optional = true } +obstack-wire = { git = "ssh://git@github.com/alexpacio/obstack-wire.git", rev = "817345d4cf563098d3b7e6ff49821231ee08a86d", optional = true } + # Prost / Protocol Buffers prost = { workspace = true, optional = true } prost-reflect = { workspace = true, optional = true } @@ -891,6 +898,7 @@ sinks-logs = [ "sinks-http", "sinks-humio", "sinks-influxdb", + "sinks-iggy", "sinks-kafka", "sinks-keep", "sinks-loki", @@ -921,6 +929,7 @@ sinks-metrics = [ "sinks-greptimedb_metrics", "sinks-humio", "sinks-influxdb", + "sinks-iggy", "sinks-kafka", "sinks-prometheus", "sinks-sematext", @@ -964,6 +973,7 @@ sinks-http = [] sinks-humio = ["sinks-splunk_hec", "transforms-metric_to_log"] sinks-influxdb = [] sinks-kafka = ["dep:rdkafka"] +sinks-iggy = ["dep:iggy", "dep:obstack-wire", "dep:hex"] sinks-keep = [] sinks-mezmo = [] sinks-loki = ["loki-logproto"] @@ -1086,6 +1096,7 @@ humio-integration-tests = ["sinks-humio"] http-client-integration-tests = ["sources-http_client", "vrl-functions-crypto"] influxdb-integration-tests = ["sinks-influxdb"] kafka-integration-tests = ["sinks-kafka", "sources-kafka"] +iggy-integration-tests = ["sinks-iggy"] logstash-integration-tests = ["docker", "sources-logstash"] loki-integration-tests = ["sinks-loki"] mongodb_metrics-integration-tests = ["sources-mongodb_metrics"] diff --git a/changelog.d/iggy_sink.feature.md b/changelog.d/iggy_sink.feature.md new file mode 100644 index 0000000000000..79a0d066892b8 --- /dev/null +++ b/changelog.d/iggy_sink.feature.md @@ -0,0 +1,11 @@ +Added a new `iggy` sink that publishes OTLP logs, metrics, and traces to +[Apache Iggy](https://iggy.apache.org/) topics. + +The sink shards, encodes, and durably appends messages using a compact, +versioned wire format designed for a queue-only observability storage +ingest path, letting a single Vector agent replace a reference +OpenTelemetry Collector plus a bespoke OTLP-to-Iggy adapter. Pair it with +the `opentelemetry` source configured with `use_otlp_decoding: true` so +the OTLP structure is preserved end to end. + +authors: alexpacio diff --git a/distribution/docker/obstack/Dockerfile b/distribution/docker/obstack/Dockerfile new file mode 100644 index 0000000000000..336a8952b5d91 --- /dev/null +++ b/distribution/docker/obstack/Dockerfile @@ -0,0 +1,29 @@ +# syntax=docker/dockerfile:1.7 +FROM rust:1.97-bookworm AS build +WORKDIR /src +RUN apt-get update \ + && apt-get install -y --no-install-recommends protobuf-compiler openssh-client \ + && rm -rf /var/lib/apt/lists/* \ + && mkdir -p -m 0700 /root/.ssh \ + && ssh-keyscan -t ed25519,rsa github.com >> /root/.ssh/known_hosts +COPY . . +# obstack-wire (sinks-iggy feature) is a private git dependency fetched over +# SSH; forward the host's ssh-agent via buildx --ssh default. +RUN --mount=type=ssh cargo build --locked --release --no-default-features \ + --features sources-datadog_agent,sources-opentelemetry,sinks-iggy \ + --bin vector + +FROM debian:bookworm-slim +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates \ + && rm -rf /var/lib/apt/lists/* \ + && useradd -r -u 10001 vector \ + && mkdir -p /etc/vector /var/lib/vector \ + && chown -R vector:vector /var/lib/vector +COPY --from=build /src/target/release/vector /usr/local/bin/vector +COPY --chmod=0755 distribution/docker/obstack/entrypoint.sh /usr/local/bin/obstack-vector-entrypoint +USER vector +VOLUME ["/var/lib/vector"] +EXPOSE 4317 4318 8080 +ENTRYPOINT ["/usr/local/bin/obstack-vector-entrypoint"] +CMD ["/usr/local/bin/vector", "--config", "/etc/vector/vector.yaml"] diff --git a/distribution/docker/obstack/entrypoint.sh b/distribution/docker/obstack/entrypoint.sh new file mode 100644 index 0000000000000..60b49cc0c2daf --- /dev/null +++ b/distribution/docker/obstack/entrypoint.sh @@ -0,0 +1,16 @@ +#!/bin/sh +set -eu + +if [ -z "${VECTOR_IGGY_URL:-}" ] \ + && [ -n "${IGGY_USERNAME_FILE:-}" ] \ + && [ -n "${IGGY_PASSWORD_FILE:-}" ]; then + VECTOR_IGGY_URL="iggy://$(cat "$IGGY_USERNAME_FILE"):$(cat "$IGGY_PASSWORD_FILE")@${IGGY_ADDRESS:-iggy:8090}" + export VECTOR_IGGY_URL +fi + +if [ -z "${VECTOR_IGGY_URL:-}" ]; then + echo "VECTOR_IGGY_URL or Iggy credential files are required" >&2 + exit 1 +fi + +exec "$@" diff --git a/src/sinks/file/mod.rs b/src/sinks/file/mod.rs index 65453c98fc218..0ed5a004ea405 100644 --- a/src/sinks/file/mod.rs +++ b/src/sinks/file/mod.rs @@ -517,14 +517,14 @@ impl FileSink { .modified() .map_err(|_| ()) .and_then(|t| t.elapsed().map_err(|_| ())) - && time.as_secs() > after_close_time_secs.into() + && time.as_secs() > after_close_time_secs.get() { truncate = true; } if let Some(after_secs) = self.truncation_config.after_secs && let Some(file) = self.files.get(path) - && (file.created_at().elapsed().as_secs() > after_secs.into()) + && (file.created_at().elapsed().as_secs() > after_secs.get()) { truncate = true; } @@ -534,7 +534,7 @@ impl FileSink { .files .get_with_deadline(path) .and_then(|(_, deadline)| deadline.checked_sub(self.idle_timeout)) - && previous_modification.elapsed().as_secs() > after_modified_time_secs.into() + && previous_modification.elapsed().as_secs() > after_modified_time_secs.get() { truncate = true; } diff --git a/src/sinks/iggy/config.rs b/src/sinks/iggy/config.rs new file mode 100644 index 0000000000000..7573776dccf43 --- /dev/null +++ b/src/sinks/iggy/config.rs @@ -0,0 +1,185 @@ +//! Configuration for the `iggy` sink. + +use std::time::Duration; + +use serde_with::serde_as; +use vector_lib::configurable::configurable_component; + +use crate::{ + config::{AcknowledgementsConfig, DataType, Input, SinkConfig, SinkContext}, + sinks::{Healthcheck, VectorSink}, +}; + +use super::sink::IggySink; + +/// Default maximum encoded message size (Iggy's own hard limit is 64 MB). +const fn default_max_message_bytes() -> usize { + 16 * 1024 * 1024 +} + +const fn default_partitions() -> u32 { + 8 +} + +fn default_stream() -> String { + "obstack".to_string() +} + +fn default_topic_prefix() -> String { + super::proto::DEFAULT_TOPIC_PREFIX.to_string() +} + +fn default_tenant_attribute() -> String { + "obstack.tenant.id".to_string() +} + +const fn default_batch_events() -> usize { + 8_192 +} + +const fn default_batch_bytes() -> usize { + 8 * 1024 * 1024 +} + +const fn default_batch_timeout() -> Duration { + Duration::from_millis(200) +} + +const fn default_replication_factor() -> u8 { + 1 +} + +const fn default_max_active_topics() -> usize { + 1_000 +} + +const fn default_bootstrap_timeout() -> Duration { + Duration::from_secs(10) +} + +/// Publish OTLP telemetry to an Obstack storage cluster through Apache Iggy. +/// +/// This sink is the producer half of Obstack's queue-only ingest path: it +/// converts OTLP events (from an `opentelemetry` source configured with +/// `use_otlp_decoding: true`) into Obstack's partitioned, generation-stamped +/// queue envelopes and durably appends them to a topic per `(tenant, +/// producer)`. It replaces the former Collector adapter with one strict, +/// retry-aware producer boundary. +#[serde_as] +#[configurable_component(sink( + "iggy", + "Publish OTLP telemetry to an Obstack cluster through Apache Iggy." +))] +#[derive(Clone, Debug)] +#[serde(deny_unknown_fields)] +pub struct IggySinkConfig { + /// Iggy connection string, e.g. `iggy://user:pass@host:8090`. + #[configurable(metadata(docs::examples = "iggy://iggy:iggy@127.0.0.1:8090"))] + pub connection_string: String, + + /// Iggy stream name. The sink creates it when absent. + #[serde(default = "default_stream")] + #[configurable(metadata(docs::examples = "obstack"))] + pub stream: String, + + /// Prefix for deterministic tenant/producer topic names. + #[serde(default = "default_topic_prefix")] + #[configurable(metadata(docs::examples = "obstack-p-"))] + pub topic_prefix: String, + + /// Partitions per producer topic. Obstack requires exactly eight. + #[serde(default = "default_partitions")] + pub partitions: u32, + + /// Required string resource attribute containing the tenant. + #[serde(default = "default_tenant_attribute")] + #[configurable(metadata(docs::examples = "obstack.tenant.id"))] + pub tenant_attribute: String, + + /// Iggy replication factor used when creating producer topics. + #[serde(default = "default_replication_factor")] + pub replication_factor: u8, + + /// Maximum producer topics cached by this sink instance. + #[serde(default = "default_max_active_topics")] + pub max_active_topics: usize, + + /// How many seconds a provisioning loser waits for the topic creator's bootstrap. + #[serde_as(as = "serde_with::DurationSeconds")] + #[serde(default = "default_bootstrap_timeout")] + pub bootstrap_timeout: Duration, + + /// Maximum encoded envelope size before a batch is split. Capped at + /// Iggy's 64 MB limit. + #[serde(default = "default_max_message_bytes")] + pub max_message_bytes: usize, + + /// Number of shared Iggy connection lanes (defaults to 16). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lanes: Option, + + /// Maximum number of source events coalesced into one publish pass. + #[serde(default = "default_batch_events")] + pub batch_events: usize, + + /// Approximate maximum source bytes coalesced into one publish pass. + #[serde(default = "default_batch_bytes")] + pub batch_bytes: usize, + + /// Maximum milliseconds to wait for a partially full publish pass. + #[serde_as(as = "serde_with::DurationMilliSeconds")] + #[serde(default = "default_batch_timeout")] + pub batch_timeout: Duration, + + #[configurable(derived)] + #[serde( + default, + deserialize_with = "crate::serde::bool_or_struct", + skip_serializing_if = "crate::serde::is_default" + )] + pub acknowledgements: AcknowledgementsConfig, +} + +impl_generate_config_from_default!(IggySinkConfig); + +impl Default for IggySinkConfig { + fn default() -> Self { + Self { + connection_string: "iggy://iggy:iggy@127.0.0.1:8090".to_string(), + stream: default_stream(), + topic_prefix: default_topic_prefix(), + partitions: default_partitions(), + tenant_attribute: default_tenant_attribute(), + replication_factor: default_replication_factor(), + max_active_topics: default_max_active_topics(), + bootstrap_timeout: default_bootstrap_timeout(), + max_message_bytes: default_max_message_bytes(), + lanes: None, + batch_events: default_batch_events(), + batch_bytes: default_batch_bytes(), + batch_timeout: default_batch_timeout(), + acknowledgements: AcknowledgementsConfig::default(), + } + } +} + +#[async_trait::async_trait] +#[typetag::serde(name = "iggy")] +impl SinkConfig for IggySinkConfig { + async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { + let sink = IggySink::connect(self.clone()).await?; + let healthcheck = futures::future::ok::<(), crate::Error>(()).boxed(); + Ok((VectorSink::from_event_streamsink(sink), healthcheck)) + } + + fn input(&self) -> Input { + // OTLP-decoding logs/metrics arrive as Log events, traces as Trace. + Input::new(DataType::Log | DataType::Trace) + } + + fn acknowledgements(&self) -> &AcknowledgementsConfig { + &self.acknowledgements + } +} + +use futures::FutureExt; diff --git a/src/sinks/iggy/datadog.rs b/src/sinks/iggy/datadog.rs new file mode 100644 index 0000000000000..68b5a9aebae3e --- /dev/null +++ b/src/sinks/iggy/datadog.rs @@ -0,0 +1,638 @@ +//! Datadog trace events -> canonical OTLP `resourceSpans`. +//! +//! Vector's Datadog Agent source deliberately preserves the tracer payload, +//! including unsigned IDs represented as signed `i64` values. Normalize that +//! source shape once, then use the same OTLP decoder and storage mapping as +//! native OTLP traces. + +use std::collections::{BTreeMap, BTreeSet}; + +use serde_json::{Value as JsonValue, json}; +use vrl::value::Value; + +pub(super) fn normalize(event: &Value, tenant_attribute: &str) -> Result, String> { + let Some(spans) = field(event, "spans").and_then(Value::as_array) else { + return Ok(None); + }; + if spans.is_empty() { + return Err("Datadog trace event has no spans".into()); + } + + let tenant = unique_tag(event, spans, tenant_attribute)? + .ok_or_else(|| format!("Datadog trace requires exactly one {tenant_attribute} tag"))?; + let high_tid = unique_tag(event, spans, "_dd.p.tid")? + .map(|value| parse_high_tid(&value)) + .transpose()?; + let runtime_id = string(field(event, "runtime_id")); + let app_version = string(field(event, "app_version")); + let payload_env = string(field(event, "env")); + let priority = integer(field(event, "priority")).unwrap_or(0); + + let mut groups: BTreeMap> = BTreeMap::new(); + for span in spans { + let meta = field(span, "meta").and_then(Value::as_object); + let service = + nonempty(string(field(span, "service"))).unwrap_or_else(|| "unknown_service".into()); + let instance = nonempty(tag(meta, "runtime-id")) + .or_else(|| nonempty(tag(meta, "runtime_id"))) + .or_else(|| nonempty(runtime_id.clone())); + let version = nonempty(tag(meta, "version")) + .or_else(|| nonempty(tag(meta, "service.version"))) + .or_else(|| nonempty(app_version.clone())); + let environment = nonempty(tag(meta, "env")) + .or_else(|| nonempty(tag(meta, "deployment.environment.name"))) + .or_else(|| nonempty(payload_env.clone())); + let resource = ResourceKey { + service, + instance, + version, + environment, + tenant: tenant.clone(), + }; + groups + .entry(resource) + .or_default() + .push(normalize_span(span, high_tid, priority)?); + } + + let resource_spans = groups + .into_iter() + .map(|(resource, spans)| { + json!({ + "resource": {"attributes": resource.attributes()}, + "scopeSpans": [{ + "scope": {"name": "dd-trace", "version": string(field(event, "tracer_version"))}, + "spans": spans + }] + }) + }) + .collect::>(); + Ok(Some(Value::from(json!({"resourceSpans": resource_spans})))) +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +struct ResourceKey { + service: String, + instance: Option, + version: Option, + environment: Option, + tenant: String, +} + +impl ResourceKey { + fn attributes(&self) -> Vec { + let mut attrs = vec![ + otlp_string("service.name", &self.service), + otlp_string("obstack.tenant.id", &self.tenant), + ]; + if let Some(value) = &self.instance { + attrs.push(otlp_string("service.instance.id", value)); + } + if let Some(value) = &self.version { + attrs.push(otlp_string("service.version", value)); + } + if let Some(value) = &self.environment { + attrs.push(otlp_string("deployment.environment.name", value)); + } + attrs + } +} + +fn normalize_span( + span: &Value, + event_high: Option, + priority: i64, +) -> Result { + let low = integer(field(span, "trace_id")) + .ok_or_else(|| "Datadog span has no trace_id".to_string())? as u64; + let span_id = integer(field(span, "span_id")) + .ok_or_else(|| "Datadog span has no span_id".to_string())? as u64; + let parent_id = integer(field(span, "parent_id")).unwrap_or(0) as u64; + let meta = field(span, "meta").and_then(Value::as_object); + let span_high = nonempty(tag(meta, "_dd.p.tid")) + .map(|value| parse_high_tid(&value)) + .transpose()? + .or(event_high) + .unwrap_or(0); + let start_ns = timestamp_ns(field(span, "start")) + .ok_or_else(|| "Datadog span has no start timestamp".to_string())?; + let duration_ns = integer(field(span, "duration")).unwrap_or(0).max(0); + let operation = string(field(span, "name")); + let resource = string(field(span, "resource")); + let span_type = string(field(span, "type")); + let error = integer(field(span, "error")).unwrap_or(0) != 0; + + let mut attrs = Vec::new(); + if !operation.is_empty() { + attrs.push(otlp_string("datadog.operation.name", &operation)); + } + if !resource.is_empty() { + attrs.push(otlp_string("datadog.resource.name", &resource)); + } + if !span_type.is_empty() { + attrs.push(otlp_string("datadog.span.type", &span_type)); + } + attrs.push(otlp_i64("datadog.sampling.priority", priority)); + + // `meta` is mostly string tags, but the tracer also folds two structured + // payloads into it as JSON strings: `events` (span events) and + // `_dd.span_links` (span links). Lift those into first-class OTLP + // `events`/`links` so TraceQL and the Tempo trace API expose them, and + // coerce well-known numeric semantic keys (e.g. `http.status_code`) back to + // integers -- dd-trace stringifies them on the wire, which would otherwise + // defeat numeric TraceQL predicates like `span.http.status_code = 500`. + let mut events = Vec::new(); + let mut links = Vec::new(); + if let Some(meta) = meta { + for (key, value) in meta { + match key.as_str() { + "_dd.p.tid" => {} + "events" => match parse_span_events(&string(Some(value))) { + Some(parsed) => events = parsed, + None => attrs.push(otlp_string(key.as_str(), &string(Some(value)))), + }, + "_dd.span_links" => match parse_span_links(&string(Some(value))) { + Some(parsed) => links = parsed, + None => attrs.push(otlp_string(key.as_str(), &string(Some(value)))), + }, + key if numeric_meta_key(key) => { + let raw = string(Some(value)); + match raw.parse::() { + Ok(number) => attrs.push(otlp_i64(key, number)), + Err(_) => attrs.push(otlp_string(key, &raw)), + } + } + key => attrs.push(otlp_string(key, &string(Some(value)))), + } + } + } + + // Datadog `metrics` are numeric tags. Skip the tracer's internal + // bookkeeping (`_sampling_priority_v1`, `_dd.measured`, `_top_level`, span + // sampling, ...): the sampling decision already rides on + // `datadog.sampling.priority`, and the rest is noise in the attribute set. + if let Some(metrics) = field(span, "metrics").and_then(Value::as_object) { + for (key, value) in metrics { + if key.starts_with('_') { + continue; + } + if let Some(value) = number(Some(value)) { + attrs.push(otlp_f64(&format!("datadog.metric.{}", key.as_str()), value)); + } + } + } + + let mut normalized = json!({ + "traceId": format!("{span_high:016x}{low:016x}"), + "spanId": format!("{span_id:016x}"), + "name": if resource.is_empty() { operation } else { resource }, + "kind": span_kind(meta, &span_type), + "startTimeUnixNano": start_ns.to_string(), + "endTimeUnixNano": start_ns.saturating_add(duration_ns).to_string(), + "attributes": attrs, + "flags": if priority > 0 { 1 } else { 0 }, + "status": { + "code": if error { 2 } else { 0 }, + "message": tag(meta, "error.msg") + } + }); + if parent_id != 0 { + normalized["parentSpanId"] = json!(format!("{parent_id:016x}")); + } + if !events.is_empty() { + normalized["events"] = JsonValue::Array(events); + } + if !links.is_empty() { + normalized["links"] = JsonValue::Array(links); + } + Ok(normalized) +} + +fn span_kind(meta: Option<&vrl::value::ObjectMap>, span_type: &str) -> i64 { + // An explicit `span.kind` tag (set by OTel-shaped integrations) always + // wins; otherwise infer from the Datadog span `type`. Inbound web requests + // are servers; outbound HTTP and every datastore/cache client type is a + // client. Anything ambiguous (e.g. `grpc`, `queue`, `custom`) stays + // unspecified and relies on `span.kind` when the integration sets it. + match tag(meta, "span.kind").to_ascii_lowercase().as_str() { + "internal" => 1, + "server" => 2, + "client" => 3, + "producer" => 4, + "consumer" => 5, + _ => match span_type { + "web" => 2, + "http" | "sql" | "db" | "cassandra" | "redis" | "memcached" | "mongodb" + | "elasticsearch" | "leveldb" | "dns" | "consul" => 3, + _ => 0, + }, + } +} + +/// Semantic-convention keys the Datadog tracer emits as strings in `meta` but +/// that OTLP and TraceQL treat as integers. Coerce these back so numeric +/// predicates such as `span.http.status_code = 500` match. dd-trace stringifies +/// `http.status_code` deliberately (see its span formatter), so without this +/// every DD span carries the status code as an unmatchable string. +fn numeric_meta_key(key: &str) -> bool { + matches!( + key, + "http.status_code" + | "http.response.status_code" + | "http.request.status_code" + | "rpc.grpc.status_code" + ) +} + +/// Parse the tracer's `meta.events` JSON array into OTLP span events. Returns +/// `None` when the payload is absent or malformed so the caller preserves the +/// raw string instead of silently dropping it. +fn parse_span_events(raw: &str) -> Option> { + let entries = serde_json::from_str::(raw).ok()?; + let events = entries + .as_array()? + .iter() + .filter_map(|entry| { + let object = entry.as_object()?; + let name = object.get("name").and_then(JsonValue::as_str).unwrap_or(""); + let time = object + .get("time_unix_nano") + .and_then(JsonValue::as_i64) + .unwrap_or(0); + Some(json!({ + "name": name, + "timeUnixNano": time.to_string(), + "attributes": object.get("attributes").map(json_attrs).unwrap_or_default(), + })) + }) + .collect(); + Some(events) +} + +/// Parse the tracer's `meta._dd.span_links` JSON array into OTLP span links. +/// The tracer emits 32-/16-char lowercase-hex ids; links with unusable ids are +/// skipped. Returns `None` on an absent or malformed payload. +fn parse_span_links(raw: &str) -> Option> { + let entries = serde_json::from_str::(raw).ok()?; + let links = entries + .as_array()? + .iter() + .filter_map(|entry| { + let object = entry.as_object()?; + let trace_id = + normalize_link_id(object.get("trace_id").and_then(JsonValue::as_str)?, 32)?; + let span_id = + normalize_link_id(object.get("span_id").and_then(JsonValue::as_str)?, 16)?; + Some(json!({ + "traceId": trace_id, + "spanId": span_id, + "attributes": object.get("attributes").map(json_attrs).unwrap_or_default(), + })) + }) + .collect(); + Some(links) +} + +/// Normalize a span-link id to canonical lowercase hex of `hex_len` characters, +/// or `None` if it is not valid hex. +fn normalize_link_id(raw: &str, hex_len: usize) -> Option { + let trimmed = raw.strip_prefix("0x").unwrap_or(raw); + if trimmed.is_empty() + || trimmed.len() > hex_len + || !trimmed.bytes().all(|byte| byte.is_ascii_hexdigit()) + { + return None; + } + Some(format!( + "{:0>width$}", + trimmed.to_ascii_lowercase(), + width = hex_len + )) +} + +/// Convert a plain JSON object -- the shape span-event / span-link attributes +/// take in the tracer's JSON -- into an OTLP `[{key, value}]` attribute array. +fn json_attrs(value: &JsonValue) -> Vec { + let Some(object) = value.as_object() else { + return Vec::new(); + }; + object + .iter() + .filter_map(|(key, value)| Some(json!({"key": key, "value": json_scalar(value)?}))) + .collect() +} + +/// Map a JSON scalar (or array of scalars) to an OTLP `AnyValue`. Nulls and +/// nested objects are dropped. +fn json_scalar(value: &JsonValue) -> Option { + match value { + JsonValue::String(value) => Some(json!({ "stringValue": value })), + JsonValue::Bool(value) => Some(json!({ "boolValue": value })), + JsonValue::Number(value) => match value.as_i64() { + Some(value) => Some(json!({"intValue": value.to_string()})), + None => value.as_f64().map(|value| json!({ "doubleValue": value })), + }, + JsonValue::Array(items) => { + let values = items.iter().filter_map(json_scalar).collect::>(); + Some(json!({ "arrayValue": { "values": values } })) + } + JsonValue::Null | JsonValue::Object(_) => None, + } +} + +/// Resolve a single value for `key` across the payload-level `tags` and every +/// span's `meta`. The Datadog Agent's v1 trace payload carries no event-level +/// tags, so for that path the tenant is recovered from span `meta` -- where the +/// tracer's global `obstack.tenant.id` tag always lands; v2 payloads may also +/// carry it in chunk/payload tags. Conflicting values are rejected. +fn unique_tag(event: &Value, spans: &[Value], key: &str) -> Result, String> { + let mut values = BTreeSet::new(); + if let Some(value) = nonempty(tag(field(event, "tags").and_then(Value::as_object), key)) { + values.insert(value); + } + for span in spans { + if let Some(value) = nonempty(tag(field(span, "meta").and_then(Value::as_object), key)) { + values.insert(value); + } + } + if values.len() > 1 { + return Err(format!("Datadog trace has conflicting {key} tags")); + } + Ok(values.into_iter().next()) +} + +fn field<'a>(value: &'a Value, key: &str) -> Option<&'a Value> { + value.as_object().and_then(|object| object.get(key)) +} + +fn tag(object: Option<&vrl::value::ObjectMap>, key: &str) -> String { + object + .and_then(|object| object.get(key)) + .map_or_else(String::new, |value| string(Some(value))) +} + +fn string(value: Option<&Value>) -> String { + match value { + Some(Value::Bytes(value)) => String::from_utf8_lossy(value).into_owned(), + Some(Value::Integer(value)) => value.to_string(), + Some(Value::Float(value)) => value.to_string(), + Some(Value::Boolean(value)) => value.to_string(), + _ => String::new(), + } +} + +fn nonempty(value: String) -> Option { + (!value.is_empty()).then_some(value) +} + +fn integer(value: Option<&Value>) -> Option { + match value { + Some(Value::Integer(value)) => Some(*value), + Some(Value::Bytes(value)) => String::from_utf8_lossy(value).parse().ok(), + _ => None, + } +} + +fn number(value: Option<&Value>) -> Option { + match value { + Some(Value::Float(value)) => Some(value.into_inner()), + Some(Value::Integer(value)) => Some(*value as f64), + Some(Value::Bytes(value)) => String::from_utf8_lossy(value).parse().ok(), + _ => None, + } +} + +fn timestamp_ns(value: Option<&Value>) -> Option { + match value { + Some(Value::Timestamp(value)) => value.timestamp_nanos_opt(), + value => integer(value), + } +} + +fn parse_high_tid(value: &str) -> Result { + let value = value.strip_prefix("0x").unwrap_or(value); + if value.len() != 16 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err("_dd.p.tid must contain exactly 16 hexadecimal characters".into()); + } + u64::from_str_radix(value, 16).map_err(|error| error.to_string()) +} + +fn otlp_string(key: &str, value: &str) -> JsonValue { + json!({"key": key, "value": {"stringValue": value}}) +} + +fn otlp_i64(key: &str, value: i64) -> JsonValue { + json!({"key": key, "value": {"intValue": value.to_string()}}) +} + +fn otlp_f64(key: &str, value: f64) -> JsonValue { + json!({"key": key, "value": {"doubleValue": value}}) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn first_span(normalized: &Value) -> Value { + field(normalized, "resourceSpans") + .and_then(Value::as_array) + .and_then(|groups| groups.first()) + .and_then(|group| field(group, "scopeSpans")) + .and_then(Value::as_array) + .and_then(|scopes| scopes.first()) + .and_then(|scope| field(scope, "spans")) + .and_then(Value::as_array) + .and_then(|spans| spans.first()) + .cloned() + .unwrap() + } + + fn attr<'a>(span: &'a Value, key: &str) -> Option<&'a Value> { + field(span, "attributes") + .and_then(Value::as_array)? + .iter() + .find(|attribute| string(field(attribute, "key")) == key) + .and_then(|attribute| field(attribute, "value")) + } + + #[test] + fn reconstructs_unsigned_128_bit_ids_and_canonical_resources() { + let event = Value::from(json!({ + "runtime_id": "node-1", + "app_version": "1.2.3", + "env": "e2e", + "priority": 1, + "tags": {"obstack.tenant.id": "tenant-a"}, + "spans": [{ + "service": "checkout", + "name": "express.request", + "resource": "GET /checkout", + "trace_id": -1, + "span_id": -2, + "parent_id": 3, + "start": 1_000_000_000_i64, + "duration": 20_i64, + "error": 0, + "type": "web", + "meta": {"_dd.p.tid": "0123456789ABCDEF", "http.method": "GET"}, + "metrics": {} + }] + })); + let normalized = normalize(&event, "obstack.tenant.id").unwrap().unwrap(); + let span = first_span(&normalized); + assert_eq!( + string(field(&span, "traceId")), + "0123456789abcdefffffffffffffffff" + ); + assert_eq!(string(field(&span, "spanId")), "fffffffffffffffe"); + assert_eq!(integer(field(&span, "kind")), Some(2)); + } + + #[test] + fn rejects_conflicting_tenants() { + let event = Value::from(json!({ + "tags": {"obstack.tenant.id": "a"}, + "spans": [{"meta": {"obstack.tenant.id": "b"}}] + })); + assert!(normalize(&event, "obstack.tenant.id").is_err()); + } + + fn single_span_event(meta: JsonValue, metrics: JsonValue, span_type: &str) -> Value { + let event = Value::from(json!({ + "tags": {"obstack.tenant.id": "t"}, + "spans": [{ + "service": "checkout", + "name": "op", + "resource": "GET /x", + "trace_id": 1, + "span_id": 2, + "start": 1_000_000_000_i64, + "duration": 5_i64, + "type": span_type, + "meta": meta, + "metrics": metrics + }] + })); + first_span(&normalize(&event, "obstack.tenant.id").unwrap().unwrap()) + } + + #[test] + fn coerces_known_numeric_meta_keys_to_integers() { + let span = single_span_event( + json!({"http.status_code": "500", "http.method": "GET"}), + json!({}), + "web", + ); + // dd-trace stringifies the status code; it must come back as an int so + // `span.http.status_code = 500` matches, while free-form tags stay strings. + assert_eq!( + integer(attr(&span, "http.status_code").and_then(|v| field(v, "intValue"))), + Some(500) + ); + assert!( + attr(&span, "http.status_code") + .and_then(|v| field(v, "stringValue")) + .is_none() + ); + assert_eq!( + string(attr(&span, "http.method").and_then(|v| field(v, "stringValue"))), + "GET" + ); + } + + #[test] + fn lifts_meta_events_into_otlp_span_events() { + let span = single_span_event( + json!({ + "events": r#"[{"name":"exception","time_unix_nano":1700000000000000000,"attributes":{"exception.type":"Error","retries":2,"flaky":true}}]"# + }), + json!({}), + "web", + ); + let events = field(&span, "events").and_then(Value::as_array).unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(string(field(&events[0], "name")), "exception"); + assert_eq!( + string(field(&events[0], "timeUnixNano")), + "1700000000000000000" + ); + let event_attrs = field(&events[0], "attributes") + .and_then(Value::as_array) + .unwrap(); + let by_key = |key: &str| { + event_attrs + .iter() + .find(|a| string(field(a, "key")) == key) + .and_then(|a| field(a, "value")) + .cloned() + }; + assert_eq!( + string(field(&by_key("exception.type").unwrap(), "stringValue")), + "Error" + ); + assert_eq!( + integer(field(&by_key("retries").unwrap(), "intValue")), + Some(2) + ); + assert!( + field(&span, "attributes") + .and_then(Value::as_array) + .unwrap() + .iter() + .all(|a| string(field(a, "key")) != "events") + ); + } + + #[test] + fn lifts_dd_span_links_into_otlp_links() { + let span = single_span_event( + json!({ + "_dd.span_links": r#"[{"trace_id":"0123456789ABCDEF0123456789ABCDEF","span_id":"FEDCBA9876543210","attributes":{"link.kind":"child"},"flags":1}]"# + }), + json!({}), + "web", + ); + let links = field(&span, "links").and_then(Value::as_array).unwrap(); + assert_eq!(links.len(), 1); + assert_eq!( + string(field(&links[0], "traceId")), + "0123456789abcdef0123456789abcdef" + ); + assert_eq!(string(field(&links[0], "spanId")), "fedcba9876543210"); + let link_attrs = field(&links[0], "attributes") + .and_then(Value::as_array) + .unwrap(); + assert_eq!(string(field(&link_attrs[0], "key")), "link.kind"); + } + + #[test] + fn malformed_structured_meta_falls_back_to_string_attribute() { + let span = single_span_event(json!({"events": "not json"}), json!({}), "web"); + // Rather than drop it, an unparseable payload survives as a raw attribute. + assert!(field(&span, "events").is_none()); + assert_eq!( + string(attr(&span, "events").and_then(|v| field(v, "stringValue"))), + "not json" + ); + } + + #[test] + fn drops_internal_metrics_but_keeps_user_metrics() { + let span = single_span_event( + json!({}), + json!({"_sampling_priority_v1": 1, "_dd.measured": 1, "queue.depth": 7}), + "web", + ); + assert!(attr(&span, "datadog.metric._sampling_priority_v1").is_none()); + assert!(attr(&span, "datadog.metric._dd.measured").is_none()); + assert!(attr(&span, "datadog.metric.queue.depth").is_some()); + } + + #[test] + fn infers_client_kind_for_datastore_span_types() { + let span = single_span_event(json!({}), json!({}), "redis"); + assert_eq!(integer(field(&span, "kind")), Some(3)); + let explicit = single_span_event(json!({"span.kind": "producer"}), json!({}), "redis"); + assert_eq!(integer(field(&explicit, "kind")), Some(4)); + } +} diff --git a/src/sinks/iggy/mod.rs b/src/sinks/iggy/mod.rs new file mode 100644 index 0000000000000..56ada4944a83b --- /dev/null +++ b/src/sinks/iggy/mod.rs @@ -0,0 +1,18 @@ +//! The `iggy` sink: publish OTLP telemetry to an Obstack cluster through +//! Apache Iggy, replacing the reference OpenTelemetry Collector + Obstack's +//! `obstack-otel-iggy` adapter with a single Vector component. +//! +//! Pair it with an `opentelemetry` source configured with +//! `use_otlp_decoding: true` so the OTLP structure is preserved end to end. + +mod config; +mod datadog; +mod otlp; +mod proto; +mod publisher; +mod sink; + +pub use config::IggySinkConfig; + +#[cfg(test)] +mod tests; diff --git a/src/sinks/iggy/otlp.rs b/src/sinks/iggy/otlp.rs new file mode 100644 index 0000000000000..a92b653986479 --- /dev/null +++ b/src/sinks/iggy/otlp.rs @@ -0,0 +1,629 @@ +//! OTLP `Value` (from the `opentelemetry` source with `use_otlp_decoding`) +//! → Obstack [`WriteBatch`] rows. +//! +//! The source's OTLP-decoding mode emits one event per request whose payload +//! mirrors the OTLP protobuf with camelCase JSON field names: attributes are +//! arrays of `{key, value: {stringValue|intValue|...}}`, timestamps are +//! numeric (or numeric-strings), and ids are hex/byte strings. This is a 1:1 +//! shape with Obstack's `obstack-ingest` decoders, whose semantics are ported +//! here field-for-field (severity→level, resource→labels, metric name +//! mapping + histogram/summary explosion, span structure). + +use vrl::value::Value; + +use super::proto::{ + Label, Labels, LogRow, MetricExemplar, SampleRow, ScalarValue, SpanEvent, SpanKind, SpanLink, + SpanRow, StatusCode, WriteBatch, +}; + +const RESOURCE_LOGS: &str = "resourceLogs"; +const RESOURCE_METRICS: &str = "resourceMetrics"; +const RESOURCE_SPANS: &str = "resourceSpans"; + +/// Which OTLP signal an event carries, by top-level field. +pub fn signal_of(event: &Value) -> Option { + let obj = event.as_object()?; + if obj.contains_key(RESOURCE_LOGS) { + Some(Signal::Logs) + } else if obj.contains_key(RESOURCE_METRICS) { + Some(Signal::Metrics) + } else if obj.contains_key(RESOURCE_SPANS) { + Some(Signal::Traces) + } else { + None + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Signal { + Logs, + Metrics, + Traces, +} + +/// Append every row in one OTLP-decoded event into `out`. +pub fn decode_event(event: &Value, out: &mut WriteBatch) { + match signal_of(event) { + Some(Signal::Logs) => decode_logs(field(event, RESOURCE_LOGS), out), + Some(Signal::Metrics) => decode_metrics(field(event, RESOURCE_METRICS), out), + Some(Signal::Traces) => decode_traces(field(event, RESOURCE_SPANS), out), + None => {} + } +} + +// ----------------------------------------------------------- accessors -- + +fn field<'a>(v: &'a Value, key: &str) -> Option<&'a Value> { + v.as_object().and_then(|o| o.get(key)) +} + +fn arr(v: Option<&Value>) -> &[Value] { + match v.and_then(Value::as_array) { + Some(a) => a, + None => &[], + } +} + +/// A VRL string (VRL stores strings as bytes). +fn as_string(v: Option<&Value>) -> String { + match v { + Some(Value::Bytes(b)) => String::from_utf8_lossy(b).into_owned(), + Some(Value::Integer(i)) => i.to_string(), + Some(Value::Float(f)) => f.to_string(), + Some(Value::Boolean(b)) => b.to_string(), + _ => String::new(), + } +} + +/// A numeric field that may arrive as an integer or a numeric string +/// (protobuf JSON encodes 64-bit ints as strings). +fn as_i64(v: Option<&Value>) -> i64 { + match v { + Some(Value::Integer(i)) => *i, + Some(Value::Float(f)) => f.into_inner() as i64, + Some(Value::Bytes(b)) => String::from_utf8_lossy(b).parse::().unwrap_or(0), + _ => 0, + } +} + +fn as_f64(v: Option<&Value>) -> Option { + match v { + Some(Value::Float(f)) => Some(f.into_inner()), + Some(Value::Integer(i)) => Some(*i as f64), + Some(Value::Bytes(b)) => String::from_utf8_lossy(b).parse::().ok(), + _ => None, + } +} + +/// Render an OTLP protobuf ID as canonical lower-case hexadecimal. Vector's +/// OTLP reflection decoder preserves protobuf `bytes` verbatim, while JSON +/// fixtures commonly contain the already-hex-encoded representation. +pub(super) fn as_id(v: Option<&Value>, raw_len: usize) -> String { + match v { + Some(Value::Bytes(bytes)) if bytes.len() == raw_len => hex::encode(bytes), + _ => as_string(v).to_ascii_lowercase(), + } +} + +// ------------------------------------------------------------- anyvalue -- + +/// Render an OTLP `AnyValue` object (`{stringValue|intValue|...}`) as a +/// display string. Mirrors `obstack_ingest::attrs::any_value_to_string`. +fn any_value_to_string(v: &Value) -> String { + let Some(obj) = v.as_object() else { + return as_string(Some(v)); + }; + if let Some(s) = obj.get("stringValue") { + as_string(Some(s)) + } else if let Some(b) = obj.get("boolValue") { + as_string(Some(b)) + } else if let Some(i) = obj.get("intValue") { + as_string(Some(i)) + } else if let Some(d) = obj.get("doubleValue") { + as_string(Some(d)) + } else if let Some(b) = obj.get("bytesValue") { + as_string(Some(b)) + } else { + any_value_to_json(v).to_string() + } +} + +/// Typed scalar for span/resource attributes. Mirrors +/// `obstack_ingest::attrs::flatten_one` for the scalar cases. +fn any_value_scalar(v: &Value) -> Option { + let obj = v.as_object()?; + if let Some(s) = obj.get("stringValue") { + Some(ScalarValue::Str(as_string(Some(s)))) + } else if let Some(b) = obj.get("boolValue") { + Some(ScalarValue::Bool(matches!(b, Value::Boolean(true)))) + } else if let Some(i) = obj.get("intValue") { + Some(ScalarValue::Int(as_i64(Some(i)))) + } else if let Some(d) = obj.get("doubleValue") { + as_f64(Some(d)).map(ScalarValue::Float) + } else { + obj.get("bytesValue") + .map(|b| ScalarValue::Str(as_string(Some(b)))) + } +} + +fn any_value_to_json(v: &Value) -> serde_json::Value { + serde_json::to_value(v).unwrap_or(serde_json::Value::Null) +} + +/// Flatten an `attributes` array (`[{key, value:{..}}]`) into typed pairs, +/// nested kvlists dotted. Mirrors `obstack_ingest::attrs::flatten_attrs`. +fn flatten_attrs(attrs: &[Value]) -> Vec<(String, ScalarValue)> { + let mut out = Vec::with_capacity(attrs.len()); + for kv in attrs { + let (Some(key), Some(value)) = (field(kv, "key"), field(kv, "value")) else { + continue; + }; + flatten_one(&as_string(Some(key)), value, &mut out); + } + out +} + +fn flatten_one(key: &str, value: &Value, out: &mut Vec<(String, ScalarValue)>) { + let Some(obj) = value.as_object() else { return }; + if let Some(kv) = obj.get("kvlistValue") { + for e in arr(field(kv, "values")) { + if let (Some(k), Some(v)) = (field(e, "key"), field(e, "value")) { + flatten_one(&format!("{key}.{}", as_string(Some(k))), v, out); + } + } + } else if obj.contains_key("arrayValue") { + out.push(( + key.into(), + ScalarValue::Str(any_value_to_json(value).to_string()), + )); + } else if let Some(scalar) = any_value_scalar(value) { + out.push((key.into(), scalar)); + } +} + +fn attr<'a>(attrs: &'a [(String, ScalarValue)], key: &str) -> Option<&'a ScalarValue> { + attrs.iter().find(|(k, _)| k == key).map(|(_, v)| v) +} + +fn service_name(resource_attrs: &[(String, ScalarValue)]) -> String { + attr(resource_attrs, "service.name") + .and_then(scalar_str) + .filter(|s| !s.is_empty()) + .unwrap_or("unknown_service") + .to_string() +} + +fn scalar_str(v: &ScalarValue) -> Option<&str> { + match v { + ScalarValue::Str(s) => Some(s), + _ => None, + } +} + +fn scalar_render(v: &ScalarValue) -> String { + match v { + ScalarValue::Str(s) => s.clone(), + ScalarValue::Int(i) => i.to_string(), + ScalarValue::Float(f) => f.to_string(), + ScalarValue::Bool(b) => b.to_string(), + } +} + +fn resource_attrs_of(resource: Option<&Value>) -> Vec<(String, ScalarValue)> { + flatten_attrs(arr(resource.and_then(|r| field(r, "attributes")))) +} + +/// Sanitize an attribute/metric name into a Loki-safe label name. +/// Mirrors `obstack_model::labels::sanitize_label_name`. +fn sanitize_label_name(name: &str) -> String { + sanitize(name, false) +} + +/// Metric names additionally allow `:`. Mirrors +/// `obstack_ingest::otlp::sanitize_metric_name`. +fn sanitize_metric_name(name: &str) -> String { + sanitize(name, true) +} + +fn sanitize(name: &str, allow_colon: bool) -> String { + let mut out = String::with_capacity(name.len()); + for (i, c) in name.chars().enumerate() { + let ok = c.is_ascii_alphanumeric() || c == '_' || (allow_colon && c == ':'); + let ok_first = c.is_ascii_alphabetic() || c == '_' || (allow_colon && c == ':'); + if (i == 0 && ok_first) || (i > 0 && ok) { + out.push(c); + } else { + out.push('_'); + } + } + if out.is_empty() { "_".to_string() } else { out } +} + +fn now_ns() -> i64 { + std::time::UNIX_EPOCH + .elapsed() + .map(|d| d.as_nanos() as i64) + .unwrap_or(0) +} + +// ---------------------------------------------------------------- logs -- + +const fn severity_level(num: i64) -> &'static str { + match num { + 1..=4 => "trace", + 5..=8 => "debug", + 9..=12 => "info", + 13..=16 => "warn", + 17..=20 => "error", + 21..=24 => "fatal", + _ => "", + } +} + +fn decode_logs(resource_logs: Option<&Value>, out: &mut WriteBatch) { + for rl in arr(resource_logs) { + let resource_attrs = resource_attrs_of(field(rl, "resource")); + let svc = service_name(&resource_attrs); + let mut base_labels: Vec