diff --git a/asap-precompute-rs/Cargo.lock b/asap-precompute-rs/Cargo.lock index 6d1167a..f9a50b4 100644 --- a/asap-precompute-rs/Cargo.lock +++ b/asap-precompute-rs/Cargo.lock @@ -63,6 +63,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 = "allocator-api2" version = "0.2.21" @@ -350,6 +359,7 @@ dependencies = [ "libc", "linkme", "otel-arrow-dfe-config", + "otel-arrow-dfe-controller", "otel-arrow-dfe-core-nodes", "otel-arrow-dfe-engine", "otel-arrow-dfe-otap", @@ -488,11 +498,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", + "base64 0.22.1", "bytes", + "form_urlencoded", "futures-util", "http", "http-body", "http-body-util", + "hyper", + "hyper-util", "itoa", "matchit", "memchr", @@ -500,10 +514,17 @@ dependencies = [ "percent-encoding", "pin-project-lite", "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sha1", "sync_wrapper", + "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -522,6 +543,7 @@ dependencies = [ "sync_wrapper", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -1016,6 +1038,26 @@ 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 = "cpp_demangle" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2bb79cb74d735044c972aae58ed0aaa9a837e85b01106a54c39e42e97f62253" +dependencies = [ + "cfg-if", +] + [[package]] name = "cpu-time" version = "1.0.0" @@ -1683,7 +1725,7 @@ dependencies = [ "itertools 0.14.0", "parking_lot", "paste", - "petgraph", + "petgraph 0.8.3", "tokio", ] @@ -1818,6 +1860,15 @@ dependencies = [ "sqlparser", ] +[[package]] +name = "debugid" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef552e6f588e446098f6ba40d89ac146c8c7b64aade83c051ee00bb5d2bc18d" +dependencies = [ + "uuid", +] + [[package]] name = "deflate64" version = "0.1.12" @@ -1927,7 +1978,17 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.60.2", + "windows-sys 0.61.2", +] + +[[package]] +name = "dispatch2" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0e367e4e7da84520dedcac1901e4da967309406d1e51017ae1abfb97adbd38" +dependencies = [ + "bitflags 2.13.0", + "objc2", ] [[package]] @@ -1977,6 +2038,26 @@ dependencies = [ "cfg-if", ] +[[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", + "quote", + "syn 2.0.118", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -2058,6 +2139,24 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "findshlibs" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40b9e59cd0f7e0806cca4be089683ecb6434e602038df21fe6bf6711b2f07f64" +dependencies = [ + "cc", + "lazy_static", + "libc", + "winapi", +] + +[[package]] +name = "fixedbitset" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80" + [[package]] name = "fixedbitset" version = "0.5.7" @@ -2104,7 +2203,7 @@ checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ "futures-core", "futures-sink", - "spin", + "spin 0.9.9", ] [[package]] @@ -3246,6 +3345,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17592d60ebacc7d5e169f4663c5f84f9161cc90328abcfe8456f41e4dfcb284" + [[package]] name = "hex" version = "0.4.3" @@ -3659,9 +3764,9 @@ dependencies = [ [[package]] name = "itertools" -version = "0.13.0" +version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +checksum = "ba291022dbbd398a455acf126c1e341954079855bc60dfdda641363bd6922569" dependencies = [ "either", ] @@ -4213,6 +4318,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "mime_guess" +version = "2.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e" +dependencies = [ + "mime", + "unicase", +] + [[package]] name = "minijinja" version = "2.24.0" @@ -4281,6 +4396,17 @@ dependencies = [ "tempfile", ] +[[package]] +name = "nix" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "598beaf3cc6fdd9a5dfb1630c2800c7acd31df7aaf0f565796fba2b53ca1af1b" +dependencies = [ + "bitflags 1.3.2", + "cfg-if", + "libc", +] + [[package]] name = "nix" version = "0.31.3" @@ -4335,6 +4461,15 @@ dependencies = [ "bitflags 2.13.0", ] +[[package]] +name = "ntapi" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3b335231dfd352ffb0f8017f3b6027a4917f7df785ea2143d8af2adc66980ae" +dependencies = [ + "winapi", +] + [[package]] name = "num" version = "0.4.3" @@ -4420,6 +4555,16 @@ dependencies = [ "libm", ] +[[package]] +name = "num_cpus" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" +dependencies = [ + "hermit-abi", + "libc", +] + [[package]] name = "num_enum" version = "0.7.6" @@ -4442,6 +4587,63 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "objc2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08849bbd4767dfae9457696856ae1c84fe4e0281bbe4a7abff2d0e06fb7981f8" +dependencies = [ + "objc2-encode", +] + +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags 2.13.0", + "dispatch2", + "objc2", +] + +[[package]] +name = "objc2-encode" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33" + +[[package]] +name = "objc2-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3e0adef53c21f888deb4fa59fc59f7eb17404926ee8a6f59f5df0fd7f9f3272" +dependencies = [ + "bitflags 2.13.0", + "objc2", +] + +[[package]] +name = "objc2-io-kit" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33fafba39597d6dc1fb709123dfa8289d39406734be322956a69f0931c73bb15" +dependencies = [ + "libc", + "objc2-core-foundation", +] + +[[package]] +name = "objc2-open-directory" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb82bed227edf5201dfedf072bba4015a33d3d4a98519837295a90f0a23f676d" +dependencies = [ + "objc2", + "objc2-core-foundation", + "objc2-foundation", +] + [[package]] name = "object" version = "0.37.3" @@ -4604,6 +4806,39 @@ dependencies = [ "num-traits", ] +[[package]] +name = "otel-arrow-dfe-admin" +version = "0.51.0" +source = "git+https://github.com/open-telemetry/otel-arrow?rev=3e85c3460361446ebfce99e9f35fffd2dd5ab740#3e85c3460361446ebfce99e9f35fffd2dd5ab740" +dependencies = [ + "axum", + "chrono", + "include_dir", + "mime_guess", + "otel-arrow-dfe-admin-types", + "otel-arrow-dfe-config", + "otel-arrow-dfe-engine", + "otel-arrow-dfe-state", + "otel-arrow-dfe-telemetry", + "pprof", + "serde", + "serde_json", + "thiserror 2.0.20", + "tokio", + "tokio-util", + "tower", +] + +[[package]] +name = "otel-arrow-dfe-admin-types" +version = "0.51.0" +source = "git+https://github.com/open-telemetry/otel-arrow?rev=3e85c3460361446ebfce99e9f35fffd2dd5ab740#3e85c3460361446ebfce99e9f35fffd2dd5ab740" +dependencies = [ + "otel-arrow-dfe-config", + "serde", + "serde_json", +] + [[package]] name = "otel-arrow-dfe-channel" version = "0.51.0" @@ -4639,6 +4874,38 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "otel-arrow-dfe-controller" +version = "0.51.0" +source = "git+https://github.com/open-telemetry/otel-arrow?rev=3e85c3460361446ebfce99e9f35fffd2dd5ab740#3e85c3460361446ebfce99e9f35fffd2dd5ab740" +dependencies = [ + "bytes", + "chrono", + "core_affinity", + "futures-util", + "humantime-serde", + "linkme", + "miette", + "otel-arrow-dfe-admin", + "otel-arrow-dfe-config", + "otel-arrow-dfe-engine", + "otel-arrow-dfe-state", + "otel-arrow-dfe-telemetry", + "otel-arrow-dfe-telemetry-macros", + "prost 0.14.4", + "serde", + "serde_json", + "serde_yaml", + "smallvec", + "sysinfo", + "thiserror 2.0.20", + "tokio", + "tokio-tungstenite", + "tokio-util", + "url", + "uuid", +] + [[package]] name = "otel-arrow-dfe-core-nodes" version = "0.51.0" @@ -4663,7 +4930,7 @@ dependencies = [ "humantime-serde", "libc", "linkme", - "nix", + "nix 0.31.3", "object_store", "otel-arrow-dfe-channel", "otel-arrow-dfe-config", @@ -4679,7 +4946,7 @@ dependencies = [ "parking_lot", "parquet", "prost 0.14.4", - "prost-types", + "prost-types 0.14.4", "rand 0.10.2", "regex", "reqwest", @@ -4719,7 +4986,7 @@ dependencies = [ "futures-core", "linkme", "memory-stats", - "nix", + "nix 0.31.3", "otel-arrow-dfe-channel", "otel-arrow-dfe-component-inventory-syntax", "otel-arrow-dfe-config", @@ -4787,7 +5054,7 @@ dependencies = [ "otel-arrow-dfe-telemetry-macros", "parking_lot", "prost 0.14.4", - "prost-types", + "prost-types 0.14.4", "reqwest", "rustls", "rustls-native-certs", @@ -4826,7 +5093,7 @@ dependencies = [ "datafusion", "hex", "itertools 0.15.0", - "nix", + "nix 0.31.3", "num_enum", "otel-arrow-dfe-config", "otel-arrow-dfe-pdata-otlp-macros", @@ -4923,7 +5190,7 @@ dependencies = [ "bytes", "crc32fast", "memmap2", - "nix", + "nix 0.31.3", "parking_lot", "thiserror 2.0.20", "tokio", @@ -5127,13 +5394,23 @@ dependencies = [ "pest", ] +[[package]] +name = "petgraph" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4c5cc86750666a3ed20bdaf5ca2a0344f9c67674cae0515bec2da16fbaa47db" +dependencies = [ + "fixedbitset 0.4.2", + "indexmap", +] + [[package]] name = "petgraph" version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8701b58ea97060d5e5b155d383a69952a60943f0e6dfe30b04c287beb0b27455" dependencies = [ - "fixedbitset", + "fixedbitset 0.5.7", "hashbrown 0.15.5", "indexmap", "serde", @@ -5225,6 +5502,31 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "196a7c80b9a7652aba7cc070827516c2abe4ccdf53d128e1944003cf5726cff1" +[[package]] +name = "pprof" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38a01da47675efa7673b032bf8efd8214f1917d89685e07e395ab125ea42b187" +dependencies = [ + "aligned-vec", + "backtrace", + "cfg-if", + "findshlibs", + "libc", + "log", + "nix 0.26.4", + "once_cell", + "prost 0.12.6", + "prost-build 0.12.6", + "prost-derive 0.12.6", + "sha2 0.10.9", + "smallvec", + "spin 0.10.1", + "symbolic-demangle", + "tempfile", + "thiserror 2.0.20", +] + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -5283,6 +5585,16 @@ dependencies = [ "parking_lot", ] +[[package]] +name = "prost" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "deb1435c188b76130da55f17a466d252ff7b1418b2ad3e037d127b94e3411f29" +dependencies = [ + "bytes", + "prost-derive 0.12.6", +] + [[package]] name = "prost" version = "0.13.5" @@ -5303,6 +5615,27 @@ dependencies = [ "prost-derive 0.14.4", ] +[[package]] +name = "prost-build" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22505a5c94da8e3b7c2996394d1c933236c4d743e81a410bcca4e6989fc066a4" +dependencies = [ + "bytes", + "heck", + "itertools 0.12.1", + "log", + "multimap", + "once_cell", + "petgraph 0.6.5", + "prettyplease", + "prost 0.12.6", + "prost-types 0.12.6", + "regex", + "syn 2.0.118", + "tempfile", +] + [[package]] name = "prost-build" version = "0.14.4" @@ -5310,13 +5643,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.13.0", + "itertools 0.14.0", "log", "multimap", - "petgraph", + "petgraph 0.8.3", "prettyplease", "prost 0.14.4", - "prost-types", + "prost-types 0.14.4", "pulldown-cmark", "pulldown-cmark-to-cmark", "regex", @@ -5324,6 +5657,19 @@ dependencies = [ "tempfile", ] +[[package]] +name = "prost-derive" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81bddcdb20abf9501610992b6759a4c888aef7d1a7247ef75e2404275ac24af1" +dependencies = [ + "anyhow", + "itertools 0.12.1", + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "prost-derive" version = "0.13.5" @@ -5331,7 +5677,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.13.0", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.118", @@ -5344,12 +5690,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.13.0", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.118", ] +[[package]] +name = "prost-types" +version = "0.12.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9091c90b0a32608e984ff2fa4091273cbdd755d54935c51d520887f4a1dbd5b0" +dependencies = [ + "prost 0.12.6", +] + [[package]] name = "prost-types" version = "0.14.4" @@ -6097,6 +6452,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -6276,6 +6642,15 @@ dependencies = [ "lock_api", ] +[[package]] +name = "spin" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3" +dependencies = [ + "lock_api", +] + [[package]] name = "sqlparser" version = "0.61.0" @@ -6336,6 +6711,29 @@ version = "3.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7401a30af6cb5818bb64852270bb722533397edcfc7344954a38f420819ece2" +[[package]] +name = "symbolic-common" +version = "12.18.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "332615d90111d8eeaf86a84dc9bbe9f65d0d8c5cf11b4caccedc37754eb0dcfd" +dependencies = [ + "debugid", + "memmap2", + "stable_deref_trait", + "uuid", +] + +[[package]] +name = "symbolic-demangle" +version = "12.18.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "912017718eb4d21930546245af9a3475c9dccf15675a5c215664e76621afc471" +dependencies = [ + "cpp_demangle", + "rustc-demangle", + "symbolic-common", +] + [[package]] name = "syn" version = "1.0.109" @@ -6389,6 +6787,21 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "sysinfo" +version = "0.39.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2071df9448915b71c4fe6d25deaf1c22f12bd234f01540b77312bb8e41361e6" +dependencies = [ + "libc", + "memchr", + "ntapi", + "objc2-core-foundation", + "objc2-io-kit", + "objc2-open-directory", + "windows", +] + [[package]] name = "system-configuration" version = "0.7.0" @@ -6634,6 +7047,18 @@ dependencies = [ "tokio-util", ] +[[package]] +name = "tokio-tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.19" @@ -6755,8 +7180,8 @@ checksum = "654e5643eff75d7f8c99197ce1440ed19a3474eada74c12bbac488b2cafdae27" dependencies = [ "prettyplease", "proc-macro2", - "prost-build", - "prost-types", + "prost-build 0.14.4", + "prost-types 0.14.4", "quote", "syn 2.0.118", "tempfile", @@ -6823,6 +7248,7 @@ version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ + "log", "pin-project-lite", "tracing-attributes", "tracing-core", @@ -6869,6 +7295,22 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.4", + "sha1", + "thiserror 2.0.20", +] + [[package]] name = "twox-hash" version = "2.1.2" @@ -7401,6 +7843,27 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windows" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580" +dependencies = [ + "windows-collections", + "windows-core", + "windows-future", + "windows-numerics", +] + +[[package]] +name = "windows-collections" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610" +dependencies = [ + "windows-core", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -7414,6 +7877,17 @@ dependencies = [ "windows-strings", ] +[[package]] +name = "windows-future" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb" +dependencies = [ + "windows-core", + "windows-link", + "windows-threading", +] + [[package]] name = "windows-implement" version = "0.60.2" @@ -7442,6 +7916,16 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-numerics" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26" +dependencies = [ + "windows-core", + "windows-link", +] + [[package]] name = "windows-registry" version = "0.6.1" @@ -7531,6 +8015,15 @@ dependencies = [ "windows_x86_64_msvc 0.53.1", ] +[[package]] +name = "windows-threading" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37" +dependencies = [ + "windows-link", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" diff --git a/asap-precompute-rs/Cargo.toml b/asap-precompute-rs/Cargo.toml index 274a537..99a8f55 100644 --- a/asap-precompute-rs/Cargo.toml +++ b/asap-precompute-rs/Cargo.toml @@ -43,6 +43,7 @@ otel-arrow-dfe-pdata-views = { git = "https://github.com/open-telemetry/otel-arr otel-arrow-dfe-state = { git = "https://github.com/open-telemetry/otel-arrow", rev = "3e85c3460361446ebfce99e9f35fffd2dd5ab740", package = "otel-arrow-dfe-state", optional = true } otel-arrow-dfe-telemetry = { git = "https://github.com/open-telemetry/otel-arrow", rev = "3e85c3460361446ebfce99e9f35fffd2dd5ab740", package = "otel-arrow-dfe-telemetry", optional = true } otel-arrow-dfe-core-nodes = { git = "https://github.com/open-telemetry/otel-arrow", rev = "3e85c3460361446ebfce99e9f35fffd2dd5ab740", package = "otel-arrow-dfe-core-nodes", optional = true } +otel-arrow-dfe-controller = { git = "https://github.com/open-telemetry/otel-arrow", rev = "3e85c3460361446ebfce99e9f35fffd2dd5ab740", package = "otel-arrow-dfe-controller", optional = true } linkme = { version = "0.3", optional = true } async-trait = { version = "0.1", optional = true } humantime-serde = { version = "1", optional = true } @@ -85,6 +86,14 @@ otap-engine = [ "dep:tokio", ] +# The deployable collector: the official OTAP engine controller with the +# ASAP nodes linked in. Picks a TLS provider, so it is not part of the library. +collector = [ + "otap-engine", + "dep:otel-arrow-dfe-controller", + "otel-arrow-dfe-otap/crypto-ring", +] + # Network benchmark only: do not impose a TLS provider on library consumers. stream-benchmark = [ "otap-engine", @@ -92,6 +101,11 @@ stream-benchmark = [ "otel-arrow-dfe-core-nodes/dev-tools", ] +[[bin]] +name = "asap-collector" +path = "src/bin/asap_collector.rs" +required-features = ["collector"] + [[bin]] name = "asap-stream-bench" path = "src/bin/asap_stream_bench.rs" diff --git a/asap-precompute-rs/src/answers.rs b/asap-precompute-rs/src/answers.rs new file mode 100644 index 0000000..a423908 --- /dev/null +++ b/asap-precompute-rs/src/answers.rs @@ -0,0 +1,357 @@ +//! Answers to the three telemetry questions, from sketches or from exact data, +//! in one output shape so dashboards can overlay `method="sketch"` and +//! `method="exact"`. +//! +//! | metric | labels (besides the series labels) | value | +//! | --- | --- | --- | +//! | `asap.quantile` | `quantile`, `bound` | quantile of the window's values | +//! | `asap.distinct` | `bound` | distinct values of the series' attribute | +//! | `asap.points` | | data points in the window | +//! | `asap.topk` | `key`, `rank`, `bound` | count of a heavy-hitter value | +//! +//! Every answer also carries `metric` (the source metric) and `method`. +//! `bound` is `estimate`, `lower` or `upper`; exact answers only carry +//! `estimate`. Bounds come from each sketch's configuration: +//! +//! * HLL with 2^14 registers: standard error 1.04/sqrt(2^14) (0.81%), shown at +//! 95% (±1.96 standard errors). +//! * KLL: normalized rank error 2.296 / k^0.9723 (the DataSketches KLL +//! formula, 99% confidence; 1.3% at k = 200), turned into the quantiles at +//! q ± error. +//! * Count-Min: counts never underestimate; with width w and N points the +//! overestimate is at most e/w * N with probability 1 - e^-depth. +use std::collections::HashMap; + +use crate::envelope::{SketchEnvelope, SketchType}; +use crate::observation::KeyValue; +use crate::precompute::{CardinalitySketch, FrequencySketch, PrecomputeError, QuantileSketch}; +use crate::sketches::{CMSHeapWrapper, HLLWrapper, KLLWrapper}; + +/// HLL standard error with 2^14 registers. +pub const HLL_RELATIVE_ERROR: f64 = 1.04 / 128.0; +/// Two-sided 95% normal quantile. +pub const Z95: f64 = 1.96; + +/// KLL normalized rank error for `k` (99% confidence). +pub fn kll_rank_error(k: usize) -> f64 { + 2.296 / (k as f64).powf(0.9723) +} + +/// One scalar answer for a window, ready for the OTAP encoder. +pub fn gauge( + template: &SketchEnvelope, + metric: &str, + method: &str, + extra: &[(&str, String)], + value: f64, +) -> SketchEnvelope { + let mut labels = template.labels.clone(); + labels.push(KeyValue::new("metric", template.metric_name.clone())); + labels.push(KeyValue::new("method", method)); + labels.extend(extra.iter().map(|(k, v)| KeyValue::new(*k, v.clone()))); + SketchEnvelope { + schema_version: 1, + sketch_type: SketchType::Unspecified, + agg_id: template.agg_id, + resource_labels: template.resource_labels.clone(), + labels, + window_start_ms: template.window_start_ms, + window_end_ms: template.window_end_ms, + payload: Vec::new(), + metric_name: metric.into(), + count: template.count, + aggregation_temporality: 0, + value, + } +} + +fn bounded( + template: &SketchEnvelope, + metric: &str, + extra: &[(&str, String)], + [estimate, lower, upper]: [f64; 3], +) -> Vec { + [("estimate", estimate), ("lower", lower), ("upper", upper)] + .into_iter() + .map(|(bound, v)| { + let mut labels = extra.to_vec(); + labels.push(("bound", bound.into())); + gauge(template, metric, "sketch", &labels, v) + }) + .collect() +} + +/// Answers from one merged sketch window. `quantiles` applies to KLL, `k` to +/// top-k. Other sketch types produce no answers. +pub fn sketch_answers( + env: &SketchEnvelope, + quantiles: &[f64], + k: usize, +) -> Result, PrecomputeError> { + let mut out = vec![gauge(env, "asap.points", "sketch", &[], env.count as f64)]; + match env.sketch_type { + SketchType::KLLSketch => { + let sk = KLLWrapper::from_snapshot(&env.payload)?; + let eps = kll_rank_error(sk.inner().k()); + for &q in quantiles { + let at = |p: f64| sk.quantile(p.clamp(0.0, 1.0)); + out.extend(bounded( + env, + "asap.quantile", + &[("quantile", q.to_string())], + [at(q), at(q - eps), at(q + eps)], + )); + } + } + SketchType::HLLSketch => { + let d = HLLWrapper::from_snapshot(&env.payload)?.estimate_cardinality(); + let e = Z95 * HLL_RELATIVE_ERROR * d; + out.extend(bounded(env, "asap.distinct", &[], [d, d - e, d + e])); + } + SketchType::CountMinSketchWithHeap => { + let sk = CMSHeapWrapper::from_snapshot(&env.payload)?; + let over = std::f64::consts::E / sk.inner().cols() as f64 * env.count as f64; + for (rank, entry) in sk.top_k(k).into_iter().enumerate() { + let key = String::from_utf8_lossy(&entry.key).into_owned(); + out.extend(bounded( + env, + "asap.topk", + &[("key", key), ("rank", (rank + 1).to_string())], + [entry.count, (entry.count - over).max(0.0), entry.count], + )); + } + } + _ => {} + } + Ok(out) +} + +/// Exact data for one series and window: what an exact branch builds and an +/// exact merge adds up. +#[derive(Debug, Default, PartialEq)] +pub struct ExactWindow { + /// Every value, for quantiles. + pub values: Vec, + /// Occurrences per attribute value; its size is the distinct count. + pub counts: HashMap, + /// Data points. + pub points: u64, +} + +const EXACT_MAGIC: &[u8; 4] = b"EXv1"; + +impl ExactWindow { + /// Adds another source's window. + pub fn merge(&mut self, other: ExactWindow) { + self.values.extend(other.values); + for (k, c) in other.counts { + *self.counts.entry(k).or_default() += c; + } + self.points += other.points; + } + + /// `EXv1`, points, values, then (key, count) pairs; little-endian. + pub fn encode(&self) -> Vec { + let keys: usize = self.counts.keys().map(|k| 12 + k.len()).sum(); + let mut out = Vec::with_capacity(28 + 8 * self.values.len() + keys); + out.extend_from_slice(EXACT_MAGIC); + out.extend_from_slice(&self.points.to_le_bytes()); + out.extend_from_slice(&(self.values.len() as u64).to_le_bytes()); + for v in &self.values { + out.extend_from_slice(&v.to_le_bytes()); + } + out.extend_from_slice(&(self.counts.len() as u64).to_le_bytes()); + for (k, c) in &self.counts { + out.extend_from_slice(&(k.len() as u32).to_le_bytes()); + out.extend_from_slice(k.as_bytes()); + out.extend_from_slice(&c.to_le_bytes()); + } + out + } + + /// Inverse of [`Self::encode`]. + pub fn decode(bytes: &[u8]) -> Result { + struct Cursor<'a>(&'a [u8]); + impl<'a> Cursor<'a> { + fn take(&mut self, n: usize) -> Result<&'a [u8], PrecomputeError> { + if self.0.len() < n { + return Err(PrecomputeError::Other( + "truncated exact window payload".into(), + )); + } + let (head, tail) = self.0.split_at(n); + self.0 = tail; + Ok(head) + } + fn u64(&mut self) -> Result { + Ok(u64::from_le_bytes(self.take(8)?.try_into().unwrap())) + } + } + let rest = bytes + .strip_prefix(EXACT_MAGIC) + .ok_or_else(|| PrecomputeError::Other("not an exact window payload".into()))?; + let mut c = Cursor(rest); + let points = c.u64()?; + let n = c.u64()? as usize; + let values = (0..n) + .map(|_| c.u64().map(f64::from_bits)) + .collect::, _>>()?; + let n = c.u64()? as usize; + let mut counts = HashMap::with_capacity(n.min(1 << 24)); + for _ in 0..n { + let len = u32::from_le_bytes(c.take(4)?.try_into().unwrap()) as usize; + let key = std::str::from_utf8(c.take(len)?) + .map_err(|_| PrecomputeError::Other("exact window key is not UTF-8".into()))?; + counts.insert(key.to_owned(), c.u64()?); + } + Ok(Self { + values, + counts, + points, + }) + } +} + +/// Exact answers in the same shape as [`sketch_answers`]. +pub fn exact_answers( + template: &SketchEnvelope, + window: &mut ExactWindow, + quantiles: &[f64], + k: usize, +) -> Vec { + let e = |metric: &str, extra: &[(&str, String)], v: f64| { + let mut labels = extra.to_vec(); + labels.push(("bound", "estimate".into())); + gauge(template, metric, "exact", &labels, v) + }; + let mut out = vec![gauge( + template, + "asap.points", + "exact", + &[], + window.points as f64, + )]; + if !window.values.is_empty() { + window.values.sort_unstable_by(f64::total_cmp); + for &q in quantiles { + let i = ((window.values.len() - 1) as f64 * q).round() as usize; + out.push(e( + "asap.quantile", + &[("quantile", q.to_string())], + window.values[i], + )); + } + } + if !window.counts.is_empty() { + out.push(e("asap.distinct", &[], window.counts.len() as f64)); + } + let mut top: Vec<_> = window.counts.iter().collect(); + top.sort_unstable_by(|a, b| b.1.cmp(a.1).then(a.0.cmp(b.0))); + for (rank, (key, count)) in top.into_iter().take(k).enumerate() { + out.push(e( + "asap.topk", + &[("key", key.clone()), ("rank", (rank + 1).to_string())], + *count as f64, + )); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::precompute::Sketch; + + fn find<'a>( + out: &'a [SketchEnvelope], + metric: &str, + labels: &[(&str, &str)], + ) -> &'a SketchEnvelope { + out.iter() + .find(|e| { + e.metric_name == metric + && labels + .iter() + .all(|(k, v)| e.labels.iter().any(|kv| kv.key == *k && kv.value == *v)) + }) + .unwrap_or_else(|| panic!("missing {metric} {labels:?}")) + } + + #[test] + fn kll_bounds_contain_the_exact_quantile() { + let mut sk = KLLWrapper::new(200, Some(1)); + let mut exact = ExactWindow::default(); + for i in 0..100_000u64 { + let v = ((i * 7919) % 100_000) as f64; + sk.update(v); + exact.values.push(v); + } + let env = SketchEnvelope { + sketch_type: SketchType::KLLSketch, + payload: sk.snapshot().unwrap(), + count: 100_000, + ..Default::default() + }; + let out = sketch_answers(&env, &[0.5, 0.99], 10).unwrap(); + let truth = exact_answers(&env, &mut exact, &[0.5, 0.99], 10); + for q in ["0.5", "0.99"] { + let t = find(&truth, "asap.quantile", &[("quantile", q)]).value; + let lo = find( + &out, + "asap.quantile", + &[("quantile", q), ("bound", "lower")], + ) + .value; + let hi = find( + &out, + "asap.quantile", + &[("quantile", q), ("bound", "upper")], + ) + .value; + assert!(lo <= t && t <= hi, "{q}: {lo} <= {t} <= {hi}"); + } + } + + #[test] + fn hll_and_topk_answers_carry_bounds() { + let mut hll = HLLWrapper::new(); + let mut heap = CMSHeapWrapper::new(4, 2048, 3); + for i in 0..20_000u32 { + hll.update(&i.to_le_bytes()); + heap.update(format!("k{}", i % 7 % 3).as_bytes(), 1); + } + let base = SketchEnvelope { + count: 20_000, + ..Default::default() + }; + let hll_env = SketchEnvelope { + sketch_type: SketchType::HLLSketch, + payload: hll.snapshot().unwrap(), + ..base.clone() + }; + let out = sketch_answers(&hll_env, &[], 3).unwrap(); + let lo = find(&out, "asap.distinct", &[("bound", "lower")]).value; + let hi = find(&out, "asap.distinct", &[("bound", "upper")]).value; + assert!(lo < 20_000.0 && 20_000.0 < hi, "{lo} {hi}"); + + let heap_env = SketchEnvelope { + sketch_type: SketchType::CountMinSketchWithHeap, + payload: heap.snapshot().unwrap(), + ..base + }; + let out = sketch_answers(&heap_env, &[], 3).unwrap(); + assert_eq!( + find(&out, "asap.topk", &[("rank", "1"), ("bound", "estimate")]) + .labels + .iter() + .find(|kv| kv.key == "key") + .unwrap() + .value, + "k0" + ); + assert!(out.iter().all(|e| e + .labels + .iter() + .any(|kv| kv.key == "method" && kv.value == "sketch"))); + } +} diff --git a/asap-precompute-rs/src/bin/asap_collector.rs b/asap-precompute-rs/src/bin/asap_collector.rs new file mode 100644 index 0000000..dc0efa1 --- /dev/null +++ b/asap-precompute-rs/src/bin/asap_collector.rs @@ -0,0 +1,37 @@ +//! The official OTAP dataflow engine with the ASAP nodes linked in. +//! +//! `asap-collector --config pipeline.yaml` runs any pipeline the upstream +//! `df_engine` runs, plus `urn:asap:processor:asap_sketches` and +//! `urn:asap:receiver:scenario_generator`. Startup, admin endpoints +//! (`/metrics`, `/readyz`) and signal handling are the upstream controller's. +use asap_precompute_rs as _; +use otel_arrow_dfe_config::config_provider::{resolve_config, ConfigFormat}; +use otel_arrow_dfe_config::engine::OtelDataflowSpec; +use otel_arrow_dfe_controller::{startup, BuildInfo, Controller, ControllerRunOptions}; +use otel_arrow_dfe_core_nodes as _; +use otel_arrow_dfe_otap::OTAP_PIPELINE_FACTORY; + +fn main() -> Result<(), Box> { + otel_arrow_dfe_otap::crypto::install_crypto_provider()?; + let config = std::env::args() + .skip_while(|a| a != "--config" && a != "-c") + .nth(1) + .ok_or("usage: asap-collector --config ")?; + let resolved = resolve_config(Some(&config))?; + let spec = match resolved.format { + ConfigFormat::Json => OtelDataflowSpec::from_json(&resolved.content)?, + ConfigFormat::Yaml => OtelDataflowSpec::from_yaml(&resolved.content)?, + }; + let options = ControllerRunOptions { + build_info: BuildInfo { + service_name: Some(env!("CARGO_BIN_NAME").to_string()), + service_version: Some(env!("CARGO_PKG_VERSION").to_string()), + }, + handle_os_signals: true, + ..Default::default() + }; + startup::validate_engine_components(&spec, &OTAP_PIPELINE_FACTORY)?; + startup::validate_controller_extensions(&spec, &options.extensions)?; + Controller::new(&OTAP_PIPELINE_FACTORY).run_forever_with_options(spec, options)?; + Ok(()) +} diff --git a/asap-precompute-rs/src/config.rs b/asap-precompute-rs/src/config.rs index a4b7d92..dc45427 100644 --- a/asap-precompute-rs/src/config.rs +++ b/asap-precompute-rs/src/config.rs @@ -176,7 +176,7 @@ pub fn sketch_size_string(sketch_type: SketchType, params: &SketchParams) -> Opt _ => None, } } - SketchType::Unspecified => None, + SketchType::Unspecified | SketchType::Exact => None, } } diff --git a/asap-precompute-rs/src/envelope.rs b/asap-precompute-rs/src/envelope.rs index b09549a..0aed562 100644 --- a/asap-precompute-rs/src/envelope.rs +++ b/asap-precompute-rs/src/envelope.rs @@ -28,6 +28,8 @@ pub enum SketchType { CountMinSketch, /// Count-Min Sketch plus a top-k heavy-hitter heap (ASAPv1 kind `0x03 0x00`). CountMinSketchWithHeap, + /// Not a sketch: an exact baseline window ([`crate::answers::ExactWindow`]). + Exact, } impl SketchType { @@ -40,6 +42,7 @@ impl SketchType { Self::CountSketch => "CountSketch", Self::CountMinSketch => "CountMinSketch", Self::CountMinSketchWithHeap => "CountMinSketchWithHeap", + Self::Exact => "Exact", Self::Unspecified => "Unspecified", } } @@ -53,6 +56,7 @@ impl SketchType { Self::CountSketch, Self::CountMinSketch, Self::CountMinSketchWithHeap, + Self::Exact, Self::Unspecified, ] .into_iter() diff --git a/asap-precompute-rs/src/lib.rs b/asap-precompute-rs/src/lib.rs index 2a931a9..e55c2fb 100644 --- a/asap-precompute-rs/src/lib.rs +++ b/asap-precompute-rs/src/lib.rs @@ -13,6 +13,7 @@ // rewrite behavior-sensitive comparisons. #![allow(clippy::neg_cmp_op_on_partial_ord)] +pub mod answers; pub mod config; pub mod envelope; pub mod kll_windows; diff --git a/asap-precompute-rs/src/otap/answer_nodes.rs b/asap-precompute-rs/src/otap/answer_nodes.rs new file mode 100644 index 0000000..5cb7ace --- /dev/null +++ b/asap-precompute-rs/src/otap/answer_nodes.rs @@ -0,0 +1,236 @@ +//! `urn:asap:processor:sketch_detector` reads merged sketch windows (the +//! output of `window_merge`) and emits each window's answers with bounds. The +//! exact and raw baselines are in [`super::baseline_nodes`]. +//! +//! ```yaml +//! detector: +//! type: urn:asap:processor:sketch_detector +//! config: {quantiles: [0.5, 0.99], k: 10} +//! ``` +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use async_trait::async_trait; +use linkme::distributed_slice; +use otel_arrow_dfe_config::node::NodeUserConfig; +use otel_arrow_dfe_engine::config::ProcessorConfig; +use otel_arrow_dfe_engine::context::PipelineContext; +use otel_arrow_dfe_engine::error::Error; +use otel_arrow_dfe_engine::local::processor as local; +use otel_arrow_dfe_engine::message::Message; +use otel_arrow_dfe_engine::node::NodeId; +use otel_arrow_dfe_engine::processor::ProcessorWrapper; +use otel_arrow_dfe_engine::{MessageSourceLocalEffectHandlerExtension, ProcessorFactory}; +use otel_arrow_dfe_otap::pdata::OtapPdata; +use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; +use serde::Deserialize; + +use otel_arrow_dfe_pdata::proto::opentelemetry::common::v1::{ + AnyValue, InstrumentationScope, KeyValue as OtlpKeyValue, +}; +use otel_arrow_dfe_pdata::proto::opentelemetry::metrics::v1::{ + Gauge, Metric, MetricsData, NumberDataPoint, ResourceMetrics, ScopeMetrics, +}; +use otel_arrow_dfe_pdata::proto::opentelemetry::resource::v1::Resource; +use otel_arrow_dfe_pdata::proto::OtlpProtoMessage; +use otel_arrow_dfe_pdata::OtapPayload; + +use super::codec::decode_pdata_to_observations; +use crate::answers::sketch_answers; +use crate::envelope::SketchEnvelope; +use crate::observation::{KeyValue, Observation}; + +/// Registered URN of the sketch detector. +pub const SKETCH_DETECTOR_URN: &str = "urn:asap:processor:sketch_detector"; + +pub(super) fn default_quantiles() -> Vec { + vec![0.5, 0.99] +} +pub(super) fn default_k() -> usize { + 10 +} +pub(super) fn default_window() -> Duration { + Duration::from_secs(60) +} +pub(super) fn default_grace() -> Duration { + Duration::from_secs(5) +} + +/// `sketch_detector` configuration. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DetectorConfig { + /// Quantiles answered from KLL windows. + #[serde(default = "default_quantiles")] + pub quantiles: Vec, + /// Heavy hitters answered from CMS + heap windows. + #[serde(default = "default_k")] + pub k: usize, +} + +pub(super) fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// Adds `asap.answer.delay` (seconds from a window's end to its answers) once +/// per window and method: the end-to-end latency of each answer path. +fn with_delay(envs: &[SketchEnvelope], now_ms: u64) -> Vec { + let mut out = envs.to_vec(); + let mut seen = std::collections::BTreeSet::new(); + for e in envs { + let method = e + .labels + .iter() + .find(|kv| kv.key == "method") + .map(|kv| kv.value.clone()); + if let Some(method) = method { + if seen.insert((e.window_end_ms, method.clone())) { + out.push(SketchEnvelope { + labels: vec![KeyValue::new("method", method)], + metric_name: "asap.answer.delay".into(), + window_start_ms: e.window_start_ms, + window_end_ms: e.window_end_ms, + value: now_ms.saturating_sub(e.window_end_ms) as f64 / 1000.0, + ..Default::default() + }); + } + } + } + out +} + +/// Plain OTLP gauges with every label on the data point, so backends such +/// as Prometheus keep them as series labels. +pub(super) fn gauges_pdata(envs: &[SketchEnvelope]) -> Result { + let mut by_metric: BTreeMap<&str, Vec> = BTreeMap::new(); + for e in envs { + let attributes: Vec = e + .labels + .iter() + .map(|kv| OtlpKeyValue::new(kv.key.clone(), AnyValue::new_string(kv.value.clone()))) + .collect(); + by_metric.entry(&e.metric_name).or_default().push( + NumberDataPoint::build() + .start_time_unix_nano(e.window_start_ms * 1_000_000) + .time_unix_nano(e.window_end_ms * 1_000_000) + .value_double(e.value) + .attributes(attributes) + .finish(), + ); + } + let metrics = by_metric + .into_iter() + .map(|(name, points)| { + Metric::build() + .name(name.to_string()) + .data_gauge(Gauge::new(points)) + .finish() + }) + .collect::>(); + let data = MetricsData::new(vec![ResourceMetrics::new( + Resource::build().finish(), + vec![ScopeMetrics::new( + InstrumentationScope::build().name("asap.answers").finish(), + metrics, + )], + )]); + let payload: OtapPayload = OtlpProtoMessage::Metrics(data) + .try_into() + .map_err(|e| format!("{e:?}"))?; + Ok(OtapPdata::new_todo_context(payload)) +} + +pub(super) async fn emit( + envs: &[SketchEnvelope], + effects: &mut local::EffectHandler, +) -> Result<(), Error> { + if envs.is_empty() { + return Ok(()); + } + let envs = with_delay(envs, now_ms()); + match gauges_pdata(&envs) { + Ok(pdata) => effects.send_message_with_source_node(pdata).await?, + Err(e) => effects.info(&format!("answers: encode failed: {e}")).await, + } + Ok(()) +} + +/// Identity of a series across windows: metric and labels. +pub(super) fn series_key(env: &SketchEnvelope) -> String { + format!("{}|{:?}", env.metric_name, env.labels) +} + +pub(super) async fn observations( + pdata: OtapPdata, + effects: &mut local::EffectHandler, +) -> Vec { + match decode_pdata_to_observations(pdata) { + Ok(outcome) => outcome.observations, + Err(e) => { + effects + .info(&format!("answers: undecodable input: {e}")) + .await; + Vec::new() + } + } +} + +struct SketchDetector { + config: DetectorConfig, +} + +#[async_trait(?Send)] +impl local::Processor for SketchDetector { + async fn process( + &mut self, + msg: Message, + effects: &mut local::EffectHandler, + ) -> Result<(), Error> { + let Message::PData(pdata) = msg else { + return Ok(()); + }; + let mut out = Vec::new(); + for obs in observations(pdata, effects).await { + let Some(env) = obs.value.envelope.filter(|e| !e.payload.is_empty()) else { + continue; + }; + match sketch_answers(&env, &self.config.quantiles, self.config.k) { + Ok(answers) => out.extend(answers), + Err(e) => effects.info(&format!("sketch_detector: {e}")).await, + } + } + emit(&out, effects).await + } +} + +pub(super) fn invalid(e: impl std::fmt::Display) -> otel_arrow_dfe_config::error::Error { + otel_arrow_dfe_config::error::Error::InvalidUserConfig { + error: e.to_string(), + } +} + +#[allow(unsafe_code)] +#[distributed_slice(OTAP_PROCESSOR_FACTORIES)] +static SKETCH_DETECTOR: ProcessorFactory = ProcessorFactory { + name: SKETCH_DETECTOR_URN, + create: |_: PipelineContext, + node: NodeId, + node_config: Arc, + processor_config: &ProcessorConfig, + _: &otel_arrow_dfe_engine::capability::registry::Capabilities| { + let config: DetectorConfig = + serde_json::from_value(node_config.config.clone()).map_err(invalid)?; + Ok(ProcessorWrapper::local( + SketchDetector { config }, + node, + node_config, + processor_config, + )) + }, + wiring_contract: otel_arrow_dfe_engine::wiring_contract::WiringContract::UNRESTRICTED, + validate_config: otel_arrow_dfe_config::validation::validate_typed_config::, +}; diff --git a/asap-precompute-rs/src/otap/baseline_nodes.rs b/asap-precompute-rs/src/otap/baseline_nodes.rs new file mode 100644 index 0000000..a38d151 --- /dev/null +++ b/asap-precompute-rs/src/otap/baseline_nodes.rs @@ -0,0 +1,881 @@ +//! The exact baseline and the raw baseline, as separate pipelines that share +//! no processor with the sketch pipeline; only the generated input is the same. +//! +//! * `urn:asap:processor:exact_branch` (one per source) keeps every raw point +//! of each event-time window as an [`ExactWindow`]: the values of each +//! quantile series and a value -> count map per attribute key. It emits a +//! window when data crosses its end, or 5 s after its end if the stream idles. +//! * `urn:asap:processor:exact_merge` adds up the sources' windows, `grace` +//! after each window's end. +//! * `urn:asap:processor:exact_detector` computes each merged window's exact +//! answers and events, with `method="exact"`: the ground truth. +//! * `urn:asap:processor:raw_sink` ends the raw pipeline, which only forwards +//! raw points: it writes every point as one log record to a store through +//! its `logs` output and watches the store's acknowledgements. For each +//! event-time window it reports on its `latency` output when the store +//! acknowledged the window's last point (`asap.answer.delay`, +//! `method="raw"`), the per-point time from event to acknowledgement +//! (`asap.raw.transit`) and rejected points (`asap.raw.failed`). +//! +//! ```yaml +//! exact_branch: +//! type: urn:asap:processor:exact_branch +//! config: {window: 60s, group_by: [http.route], per_attribute: [user.id]} +//! exact_merge: +//! type: urn:asap:processor:exact_merge +//! config: {grace: 5s} +//! exact_detector: +//! type: urn:asap:processor:exact_detector +//! config: {quantiles: [0.5, 0.99], k: 10} +//! raw_sink: +//! type: urn:asap:processor:raw_sink +//! outputs: [logs, latency] +//! config: {window: 60s, grace: 5s} +//! ``` +use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use linkme::distributed_slice; +use otel_arrow_dfe_config::node::NodeUserConfig; +use otel_arrow_dfe_engine::config::ProcessorConfig; +use otel_arrow_dfe_engine::context::PipelineContext; +use otel_arrow_dfe_engine::control::{AckMsg, CallData, Context8u8, NodeControlMsg}; +use otel_arrow_dfe_engine::error::Error; +use otel_arrow_dfe_engine::local::processor as local; +use otel_arrow_dfe_engine::message::Message; +use otel_arrow_dfe_engine::node::NodeId; +use otel_arrow_dfe_engine::processor::ProcessorWrapper; +use otel_arrow_dfe_engine::{ + ConsumerEffectHandlerExtension, Interests, ProducerEffectHandlerExtension, +}; +use otel_arrow_dfe_engine::{MessageSourceLocalEffectHandlerExtension, ProcessorFactory}; +use otel_arrow_dfe_otap::pdata::OtapPdata; +use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; +use otel_arrow_dfe_pdata::proto::opentelemetry::common::v1::{ + AnyValue, InstrumentationScope, KeyValue as OtlpKeyValue, +}; +use otel_arrow_dfe_pdata::proto::opentelemetry::logs::v1::{ + LogRecord, LogsData, ResourceLogs, ScopeLogs, +}; +use otel_arrow_dfe_pdata::proto::opentelemetry::resource::v1::Resource; +use otel_arrow_dfe_pdata::proto::OtlpProtoMessage; +use otel_arrow_dfe_pdata::OtapPayload; +use serde::Deserialize; + +use super::answer_nodes::{ + default_grace, default_k, default_quantiles, default_window, emit, gauges_pdata, invalid, + now_ms, observations, series_key, +}; +use super::codec::OtapSketchEncoder; +use crate::answers::{exact_answers, gauge, ExactWindow}; +use crate::envelope::{SketchEnvelope, SketchType}; + +use crate::observation::{KeyValue, Observation, ObservationValueKind}; + +/// Registered URN of the exact branch. +pub const EXACT_BRANCH_URN: &str = "urn:asap:processor:exact_branch"; +/// Registered URN of the exact merge. +pub const EXACT_MERGE_URN: &str = "urn:asap:processor:exact_merge"; +/// Registered URN of the exact detector. +pub const EXACT_DETECTOR_URN: &str = "urn:asap:processor:exact_detector"; +/// Registered URN of the raw pipeline's sink. +pub const RAW_SINK_URN: &str = "urn:asap:processor:raw_sink"; +/// `raw_sink` output with the log records for the store. +const LOGS_PORT: &str = "logs"; +/// `raw_sink` output with the latency gauges. +const LATENCY_PORT: &str = "latency"; + +/// Wait after a window's end before the idle timer closes it, as in +/// `asap_sketches`: data crossing the end closes it on time. +const IDLE_CLOSE_GRACE: Duration = Duration::from_secs(5); + +/// `exact_branch` configuration. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExactBranchConfig { + /// Event-time window size. + #[serde(default = "default_window", with = "humantime_serde")] + pub window: Duration, + /// Labels identifying a quantile series, as `aggregate_by`; empty + /// disables quantiles. + #[serde(default)] + pub group_by: Vec, + /// Attribute keys whose value counts are kept; `*` for every key. + #[serde(default)] + pub per_attribute: Vec, +} + +/// `exact_merge` configuration. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExactMergeConfig { + /// How long after a window's end to wait for the last source. + #[serde(default = "default_grace", with = "humantime_serde")] + pub grace: Duration, +} + +/// `exact_detector` configuration. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExactDetectorConfig { + /// Quantiles to answer. + #[serde(default = "default_quantiles")] + pub quantiles: Vec, + /// Heavy hitters per attribute key. + #[serde(default = "default_k")] + pub k: usize, +} + +/// `raw_sink` configuration. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RawSinkConfig { + /// Event-time window size. + #[serde(default = "default_window", with = "humantime_serde")] + pub window: Duration, + /// Wait after a window's end before reporting it. + #[serde(default = "default_grace", with = "humantime_serde")] + pub grace: Duration, +} + +/// Open windows by window end, then series. +type Windows = BTreeMap>; + +/// Removes and encodes every window ending at or before `until`. +fn close(windows: &mut Windows, until: u64) -> Vec { + let open = windows.split_off(&(until.saturating_add(1))); + std::mem::replace(windows, open) + .into_values() + .flat_map(|series| series.into_values()) + .map(|(mut env, w)| { + env.count = w.points; + env.payload = w.encode(); + env + }) + .collect() +} + +async fn send( + encoder: &mut OtapSketchEncoder, + envs: &[SketchEnvelope], + effects: &mut local::EffectHandler, +) -> Result<(), Error> { + if envs.is_empty() { + return Ok(()); + } + match encoder.encode(envs) { + Ok(pdata) => effects.send_message_with_source_node(pdata).await?, + Err(e) => effects.info(&format!("exact: encode failed: {e}")).await, + } + Ok(()) +} + +struct ExactBranch { + config: ExactBranchConfig, + windows: Windows, + /// Every window ending at or before this has been emitted. + emitted_until: u64, + encoder: OtapSketchEncoder, + timer_started: bool, +} + +impl ExactBranch { + fn observe(&mut self, obs: &Observation) { + if obs.value.kind != ObservationValueKind::Float { + return; + } + let size = self.config.window.as_millis().max(1) as u64; + let start = obs.timestamp_ms / size * size; + let end = start + size; + if end <= self.emitted_until { + return; + } + let state = self.windows.entry(end).or_default(); + let template = |labels: Vec| SketchEnvelope { + schema_version: 1, + sketch_type: SketchType::Exact, + labels, + window_start_ms: start, + window_end_ms: end, + metric_name: obs.metric.clone(), + ..Default::default() + }; + let label = |k: &str| { + obs.labels + .iter() + .find(|kv| kv.key == k) + .map(|kv| kv.value.clone()) + }; + if !self.config.group_by.is_empty() { + let labels: Vec = self + .config + .group_by + .iter() + .filter_map(|k| label(k).map(|v| KeyValue::new(k.clone(), v))) + .collect(); + let (_, w) = state + .entry(format!("q|{labels:?}")) + .or_insert_with(|| (template(labels), ExactWindow::default())); + w.values.push(obs.value.float); + w.points += 1; + } + let all = self.config.per_attribute.iter().any(|k| k == "*"); + for kv in &obs.labels { + if !all && !self.config.per_attribute.contains(&kv.key) { + continue; + } + let (_, w) = state.entry(format!("a|{}", kv.key)).or_insert_with(|| { + let labels = vec![KeyValue::new("attribute.key", kv.key.clone())]; + (template(labels), ExactWindow::default()) + }); + *w.counts.entry(kv.value.clone()).or_default() += 1; + w.points += 1; + } + } + + fn close(&mut self, until: u64) -> Vec { + let out = close(&mut self.windows, until); + if let Some(end) = out.iter().map(|e| e.window_end_ms).max() { + self.emitted_until = self.emitted_until.max(end); + } + out + } +} + +#[async_trait(?Send)] +impl local::Processor for ExactBranch { + async fn process( + &mut self, + msg: Message, + effects: &mut local::EffectHandler, + ) -> Result<(), Error> { + if !self.timer_started { + self.timer_started = effects + .start_periodic_timer(Duration::from_secs(1)) + .await + .is_ok(); + } + let out = match msg { + Message::PData(pdata) => { + // Consumed into state here: ack so upstream can move on. + effects.notify_ack(AckMsg::new(pdata.clone())).await?; + let points = observations(pdata, effects).await; + // A window closes as soon as a point's event time passes its end. + let out = match points.first() { + Some(first) => self.close(first.timestamp_ms), + None => Vec::new(), + }; + points.iter().for_each(|obs| self.observe(obs)); + out + } + Message::Control(NodeControlMsg::TimerTick { .. }) => { + self.close(now_ms().saturating_sub(IDLE_CLOSE_GRACE.as_millis() as u64)) + } + Message::Control(NodeControlMsg::Shutdown { .. }) => self.close(u64::MAX), + _ => Vec::new(), + }; + send(&mut self.encoder, &out, effects).await + } +} + +struct ExactMerge { + config: ExactMergeConfig, + windows: Windows, + emitted_until: u64, + encoder: OtapSketchEncoder, + timer_started: bool, +} + +impl ExactMerge { + /// Adds one source's window; windows already emitted drop it as late. + fn add(&mut self, env: SketchEnvelope) -> Result<(), crate::precompute::PrecomputeError> { + if env.window_end_ms <= self.emitted_until { + return Ok(()); + } + let w = ExactWindow::decode(&env.payload)?; + let key = format!("{}|{}", series_key(&env), env.window_start_ms); + let window = self.windows.entry(env.window_end_ms).or_default(); + match window.get_mut(&key) { + Some((_, merged)) => merged.merge(w), + None => { + window.insert(key, (env, w)); + } + } + Ok(()) + } + + fn due(&mut self, now: u64) -> Vec { + let cutoff = now.saturating_sub(self.config.grace.as_millis() as u64); + let out = close(&mut self.windows, cutoff); + if let Some(end) = out.iter().map(|e| e.window_end_ms).max() { + self.emitted_until = self.emitted_until.max(end); + } + out + } +} + +#[async_trait(?Send)] +impl local::Processor for ExactMerge { + async fn process( + &mut self, + msg: Message, + effects: &mut local::EffectHandler, + ) -> Result<(), Error> { + if !self.timer_started { + self.timer_started = effects + .start_periodic_timer(Duration::from_secs(1)) + .await + .is_ok(); + } + let out = match msg { + Message::PData(pdata) => { + effects.notify_ack(AckMsg::new(pdata.clone())).await?; + for obs in observations(pdata, effects).await { + let Some(env) = obs.value.envelope else { + continue; + }; + if let Err(e) = self.add(*env) { + effects.info(&format!("exact_merge: {e}")).await; + } + } + Vec::new() + } + Message::Control(NodeControlMsg::TimerTick { .. }) => self.due(now_ms()), + Message::Control(NodeControlMsg::Shutdown { .. }) => self.due(u64::MAX), + _ => Vec::new(), + }; + send(&mut self.encoder, &out, effects).await + } +} + +struct ExactDetector { + config: ExactDetectorConfig, +} + +impl ExactDetector { + /// Answers of one merged window. + fn answer(&mut self, mut template: SketchEnvelope) -> Vec { + let mut out = Vec::new(); + let Ok(mut w) = ExactWindow::decode(&template.payload) else { + return out; + }; + template.payload = Vec::new(); + template.sketch_type = SketchType::Unspecified; + template.count = w.points; + out.extend(exact_answers( + &template, + &mut w, + &self.config.quantiles, + self.config.k, + )); + out + } +} + +#[async_trait(?Send)] +impl local::Processor for ExactDetector { + async fn process( + &mut self, + msg: Message, + effects: &mut local::EffectHandler, + ) -> Result<(), Error> { + let Message::PData(pdata) = msg else { + return Ok(()); + }; + effects.notify_ack(AckMsg::new(pdata.clone())).await?; + let mut out = Vec::new(); + for obs in observations(pdata, effects).await { + let Some(env) = obs.value.envelope.filter(|e| !e.payload.is_empty()) else { + continue; + }; + out.extend(self.answer(*env)); + } + emit(&out, effects).await + } +} + +/// What the raw pipeline stored for one event-time window. +#[derive(Default)] +struct RawWindow { + /// When the store acknowledged the window's last point (ms). + last_ack_ms: u64, + /// Store acknowledgement minus event time of every stored point (s). + transit_s: Vec, + /// Points the store rejected. + failed: u64, + /// Messages with points of this window still waiting for the store. + in_flight: usize, +} + +struct RawSink { + config: RawSinkConfig, + windows: BTreeMap, + /// Event times (ms) by window end of every message the store has not + /// acknowledged yet, by message id. + in_flight: HashMap)>>, + next_id: u64, + emitted_until: u64, + timer_started: bool, +} + +/// One log record per raw point: event time, the value as the body, the +/// point's attributes. +fn logs_pdata(points: &[Observation]) -> Result { + let mut by_resource: BTreeMap, Vec> = BTreeMap::new(); + for p in points { + let resource = p + .resource_labels + .iter() + .map(|kv| (kv.key.clone(), kv.value.clone())) + .collect(); + let attributes: Vec = p + .labels + .iter() + .map(|kv| OtlpKeyValue::new(kv.key.clone(), AnyValue::new_string(kv.value.clone()))) + .collect(); + by_resource.entry(resource).or_default().push( + LogRecord::build() + .time_unix_nano(p.timestamp_ms * 1_000_000) + .body(AnyValue::new_double(p.value.float)) + .attributes(attributes) + .finish(), + ); + } + let resource_logs: Vec = by_resource + .into_iter() + .map(|(resource, records)| { + let attributes: Vec = resource + .into_iter() + .map(|(k, v)| OtlpKeyValue::new(k, AnyValue::new_string(v))) + .collect(); + ResourceLogs::new( + Resource::build().attributes(attributes).finish(), + vec![ScopeLogs::new( + InstrumentationScope::build().name("asap.raw").finish(), + records, + )], + ) + }) + .collect(); + let payload: OtapPayload = OtlpProtoMessage::Logs(LogsData::new(resource_logs)) + .try_into() + .map_err(|e| format!("{e:?}"))?; + Ok(OtapPdata::new_todo_context(payload)) +} + +impl RawSink { + /// Registers a message about to go to the store; returns its id. + fn send(&mut self, points: &[Observation]) -> u64 { + let size = self.config.window.as_millis().max(1) as u64; + let mut by_window: BTreeMap> = BTreeMap::new(); + for p in points { + let end = p.timestamp_ms / size * size + size; + if end > self.emitted_until { + by_window.entry(end).or_default().push(p.timestamp_ms); + } + } + for end in by_window.keys() { + self.windows.entry(*end).or_default().in_flight += 1; + } + let id = self.next_id; + self.next_id += 1; + let _ = self.in_flight.insert(id, by_window.into_iter().collect()); + id + } + + /// The store answered message `id` at `now`. + fn answered(&mut self, id: u64, now: u64, stored: bool) { + for (end, times) in self.in_flight.remove(&id).unwrap_or_default() { + let Some(w) = self.windows.get_mut(&end) else { + continue; + }; + w.in_flight = w.in_flight.saturating_sub(1); + if stored { + w.last_ack_ms = w.last_ack_ms.max(now); + w.transit_s + .extend(times.iter().map(|t| now.saturating_sub(*t) as f64 / 1000.0)); + } else { + w.failed += times.len() as u64; + } + } + } + + /// Gauges of every window that ended `grace` or more before `now` and + /// has no message waiting for the store (or ended 10 graces ago). + fn due(&mut self, now: u64) -> Vec { + let grace = self.config.grace.as_millis() as u64; + let size = self.config.window.as_millis() as u64; + let ready: Vec = self + .windows + .iter() + .filter(|(end, w)| { + **end + grace <= now && (w.in_flight == 0 || **end + 10 * grace <= now) + }) + .map(|(end, _)| *end) + .collect(); + let mut out = Vec::new(); + for end in ready { + let mut w = self.windows.remove(&end).unwrap_or_default(); + self.emitted_until = self.emitted_until.max(end); + let template = SketchEnvelope { + window_start_ms: end - size, + window_end_ms: end, + count: w.transit_s.len() as u64, + ..Default::default() + }; + let metric = |name: &str, extra: &[(&str, String)], v: f64| { + let mut e = gauge(&template, name, "raw", extra, v); + // Window-level gauges: no source metric label. + e.labels.retain(|kv| kv.key != "metric"); + e + }; + // Complete once the last point is stored, and not before the + // window has ended. + out.push(metric( + "asap.answer.delay", + &[], + w.last_ack_ms.saturating_sub(end) as f64 / 1000.0, + )); + out.push(metric("asap.points", &[], w.transit_s.len() as f64)); + out.push(metric("asap.raw.failed", &[], w.failed as f64)); + if !w.transit_s.is_empty() { + w.transit_s.sort_unstable_by(f64::total_cmp); + let at = + |q: f64| w.transit_s[((w.transit_s.len() - 1) as f64 * q).round() as usize]; + for q in [0.5, 0.99] { + out.push(metric( + "asap.raw.transit", + &[("quantile", q.to_string())], + at(q), + )); + } + } + } + out + } +} + +#[async_trait(?Send)] +impl local::Processor for RawSink { + async fn process( + &mut self, + msg: Message, + effects: &mut local::EffectHandler, + ) -> Result<(), Error> { + if !self.timer_started { + self.timer_started = effects + .start_periodic_timer(Duration::from_secs(1)) + .await + .is_ok(); + } + let out = match msg { + Message::PData(pdata) => { + effects.notify_ack(AckMsg::new(pdata.clone())).await?; + let points: Vec = observations(pdata, effects) + .await + .into_iter() + .filter(|o| o.value.kind == ObservationValueKind::Float) + .collect(); + if points.is_empty() { + return Ok(()); + } + match logs_pdata(&points) { + Ok(mut logs) => { + let id = self.send(&points); + let calldata = CallData::from_vec(vec![Context8u8::from(id)]); + effects.subscribe_to(Interests::ACKS_OR_NACKS, calldata, &mut logs); + effects + .send_message_to(LOGS_PORT, logs) + .await + .map_err(Error::from)?; + } + Err(e) => effects.info(&format!("raw_sink: encode failed: {e}")).await, + } + return Ok(()); + } + Message::Control(NodeControlMsg::Ack(ack)) => { + if let Some(id) = ack.unwind.route.calldata.first() { + self.answered(u64::from(*id), now_ms(), true); + } + return Ok(()); + } + Message::Control(NodeControlMsg::Nack(nack)) => { + if let Some(id) = nack.unwind.route.calldata.first() { + self.answered(u64::from(*id), now_ms(), false); + } + return Ok(()); + } + Message::Control(NodeControlMsg::TimerTick { .. }) => self.due(now_ms()), + Message::Control(NodeControlMsg::Shutdown { .. }) => self.due(u64::MAX), + _ => return Ok(()), + }; + if out.is_empty() { + return Ok(()); + } + // Not through `emit`: the delay here is the last store + // acknowledgement, not the time this gauge is sent. + match gauges_pdata(&out) { + Ok(pdata) => effects + .send_message_to(LATENCY_PORT, pdata) + .await + .map_err(Error::from)?, + Err(e) => effects.info(&format!("raw_sink: encode failed: {e}")).await, + } + Ok(()) + } +} + +fn config( + node_config: &NodeUserConfig, +) -> Result { + serde_json::from_value(node_config.config.clone()).map_err(invalid) +} + +#[allow(unsafe_code)] +#[distributed_slice(OTAP_PROCESSOR_FACTORIES)] +static EXACT_BRANCH: ProcessorFactory = ProcessorFactory { + name: EXACT_BRANCH_URN, + create: |_: PipelineContext, + node: NodeId, + node_config: Arc, + processor_config: &ProcessorConfig, + _: &otel_arrow_dfe_engine::capability::registry::Capabilities| { + Ok(ProcessorWrapper::local( + ExactBranch { + config: config(&node_config)?, + windows: Windows::new(), + emitted_until: 0, + encoder: OtapSketchEncoder::default(), + timer_started: false, + }, + node, + node_config, + processor_config, + )) + }, + wiring_contract: otel_arrow_dfe_engine::wiring_contract::WiringContract::UNRESTRICTED, + validate_config: otel_arrow_dfe_config::validation::validate_typed_config::, +}; + +#[allow(unsafe_code)] +#[distributed_slice(OTAP_PROCESSOR_FACTORIES)] +static EXACT_MERGE: ProcessorFactory = ProcessorFactory { + name: EXACT_MERGE_URN, + create: |_: PipelineContext, + node: NodeId, + node_config: Arc, + processor_config: &ProcessorConfig, + _: &otel_arrow_dfe_engine::capability::registry::Capabilities| { + Ok(ProcessorWrapper::local( + ExactMerge { + config: config(&node_config)?, + windows: Windows::new(), + emitted_until: 0, + encoder: OtapSketchEncoder::default(), + timer_started: false, + }, + node, + node_config, + processor_config, + )) + }, + wiring_contract: otel_arrow_dfe_engine::wiring_contract::WiringContract::UNRESTRICTED, + validate_config: otel_arrow_dfe_config::validation::validate_typed_config::, +}; + +#[allow(unsafe_code)] +#[distributed_slice(OTAP_PROCESSOR_FACTORIES)] +static EXACT_DETECTOR: ProcessorFactory = ProcessorFactory { + name: EXACT_DETECTOR_URN, + create: |_: PipelineContext, + node: NodeId, + node_config: Arc, + processor_config: &ProcessorConfig, + _: &otel_arrow_dfe_engine::capability::registry::Capabilities| { + let config: ExactDetectorConfig = config(&node_config)?; + Ok(ProcessorWrapper::local( + ExactDetector { config }, + node, + node_config, + processor_config, + )) + }, + wiring_contract: otel_arrow_dfe_engine::wiring_contract::WiringContract::UNRESTRICTED, + validate_config: otel_arrow_dfe_config::validation::validate_typed_config::, +}; + +#[allow(unsafe_code)] +#[distributed_slice(OTAP_PROCESSOR_FACTORIES)] +static RAW_SINK: ProcessorFactory = ProcessorFactory { + name: RAW_SINK_URN, + create: |_: PipelineContext, + node: NodeId, + node_config: Arc, + processor_config: &ProcessorConfig, + _: &otel_arrow_dfe_engine::capability::registry::Capabilities| { + Ok(ProcessorWrapper::local( + RawSink { + config: config(&node_config)?, + windows: BTreeMap::new(), + in_flight: HashMap::new(), + next_id: 0, + emitted_until: 0, + timer_started: false, + }, + node, + node_config, + processor_config, + )) + }, + wiring_contract: otel_arrow_dfe_engine::wiring_contract::WiringContract::UNRESTRICTED, + validate_config: otel_arrow_dfe_config::validation::validate_typed_config::, +}; + +#[cfg(test)] +mod tests { + use super::*; + use crate::observation::ObservationValue; + + fn point(ts: u64, route: &str, user: &str, v: f64) -> Observation { + Observation { + timestamp_ms: ts, + metric: "latency".into(), + resource_labels: vec![], + labels: vec![ + KeyValue::new("http.route", route), + KeyValue::new("user.id", user), + ], + value: ObservationValue::float(v), + } + } + + fn branch() -> ExactBranch { + ExactBranch { + config: serde_json::from_value(serde_json::json!({ + "group_by": ["http.route"], "per_attribute": ["user.id"] + })) + .unwrap(), + windows: Windows::new(), + emitted_until: 0, + encoder: OtapSketchEncoder::default(), + timer_started: false, + } + } + + fn get(out: &[SketchEnvelope], metric: &str, extra: (&str, &str)) -> Option { + out.iter() + .find(|e| { + e.metric_name == metric + && e.labels + .iter() + .any(|kv| kv.key == extra.0 && kv.value == extra.1) + }) + .map(|e| e.value) + } + + #[test] + fn exact_window_round_trips() { + let mut w = ExactWindow { + values: vec![1.5, -2.0], + points: 3, + ..Default::default() + }; + w.counts.insert("a".into(), 2); + w.counts.insert("bé".into(), 1); + assert_eq!(ExactWindow::decode(&w.encode()).unwrap(), w); + assert!(ExactWindow::decode(b"EXv1\x01").is_err()); + } + + #[test] + fn branches_merge_and_answer_like_one_exact_window() { + // Two sources, 50 points each in window [0, 60 s); a later point + // closes the window in each branch. + let (mut a, mut b) = (branch(), branch()); + for i in 0..50u64 { + a.observe(&point(1_000 + i, "/a", &format!("u{}", i % 4), i as f64)); + b.observe(&point( + 1_000 + i, + "/a", + &format!("u{}", i % 4), + (50 + i) as f64, + )); + } + let mut sources = a.close(61_000); + sources.extend(b.close(61_000)); + assert_eq!( + sources.len(), + 4, + "a quantile and an attribute series per source" + ); + let mut merge = ExactMerge { + config: ExactMergeConfig { + grace: Duration::from_secs(5), + }, + windows: Windows::new(), + emitted_until: 0, + encoder: OtapSketchEncoder::default(), + timer_started: false, + }; + sources.into_iter().for_each(|e| merge.add(e).unwrap()); + assert!(merge.due(64_000).is_empty()); + let merged = merge.due(65_000); + assert_eq!(merged.len(), 2); + assert!(merged.iter().all(|e| e.count == 100)); + let config: ExactDetectorConfig = + serde_json::from_value(serde_json::json!({"k": 2})).unwrap(); + let mut detector = ExactDetector { config }; + let out: Vec<_> = merged + .into_iter() + .flat_map(|e| detector.answer(e)) + .collect(); + assert_eq!(get(&out, "asap.quantile", ("quantile", "0.5")), Some(50.0)); + assert_eq!( + get(&out, "asap.distinct", ("attribute.key", "user.id")), + Some(4.0) + ); + assert_eq!(get(&out, "asap.topk", ("rank", "1")), Some(26.0)); + assert!(out.iter().all(|e| e.window_end_ms == 60_000)); + // A late window is dropped. + merge.add(branch_window(2_000)).unwrap(); + assert!(merge.due(200_000).is_empty()); + } + + fn branch_window(ts: u64) -> SketchEnvelope { + let mut b = branch(); + b.observe(&point(ts, "/a", "u0", 1.0)); + b.close(u64::MAX).remove(0) + } + + #[test] + fn raw_sink_reports_the_last_acknowledgement_after_grace() { + let mut raw = RawSink { + config: serde_json::from_value(serde_json::json!({})).unwrap(), + windows: BTreeMap::new(), + in_flight: HashMap::new(), + next_id: 0, + emitted_until: 0, + timer_started: false, + }; + let first = raw.send(&[point(10_000, "/a", "u0", 1.0)]); + let last = raw.send(&[ + point(59_900, "/a", "u1", 1.0), + point(59_950, "/b", "u2", 1.0), + ]); + let lost = raw.send(&[point(30_000, "/a", "u3", 1.0)]); + raw.answered(first, 10_500, true); + raw.answered(lost, 30_100, false); + // Past the grace, but one message is still waiting for the store. + assert!(raw.due(65_000).is_empty()); + raw.answered(last, 61_200, true); + let out = raw.due(65_000); + assert_eq!(get(&out, "asap.answer.delay", ("method", "raw")), Some(1.2)); + assert_eq!(get(&out, "asap.points", ("method", "raw")), Some(3.0)); + assert_eq!(get(&out, "asap.raw.failed", ("method", "raw")), Some(1.0)); + assert_eq!( + get(&out, "asap.raw.transit", ("quantile", "0.99")), + Some(1.3) + ); + } +} diff --git a/asap-precompute-rs/src/otap/codec.rs b/asap-precompute-rs/src/otap/codec.rs index 6deafcb..61dbbaf 100644 --- a/asap-precompute-rs/src/otap/codec.rs +++ b/asap-precompute-rs/src/otap/codec.rs @@ -69,10 +69,6 @@ use crate::config::{sketch_size_string, SketchParams}; use crate::envelope::{SketchEnvelope, SketchType}; use crate::observation::{KeyValue, Observation, ObservationValue}; -use crate::sketches::{ - CMSHeapWrapper, CMSWrapper, CountSketchWrapper, DDSketchWrapper, HLLWrapper, KLLWrapper, -}; - use super::schema::{ ATTR_AGG_ID, ATTR_ENVELOPE, ATTR_SCHEMA_VERSION, ATTR_SERIES_ID, ATTR_SKETCH_SIZE, ATTR_SKETCH_TYPE, ATTR_WINDOW_END_MS, ATTR_WINDOW_START_MS, @@ -184,17 +180,14 @@ fn validate_self_describing_payload(sketch_type: SketchType, payload: &[u8]) -> if payload.is_empty() { return Ok(()); } - let decoded = match sketch_type { - SketchType::DDSketch => DDSketchWrapper::from_snapshot(payload).map(drop), - SketchType::KLLSketch => KLLWrapper::from_snapshot(payload).map(drop), - SketchType::HLLSketch => HLLWrapper::from_snapshot(payload).map(drop), - SketchType::CountSketch => CountSketchWrapper::from_snapshot(payload).map(drop), - SketchType::CountMinSketch => CMSWrapper::from_snapshot(payload).map(drop), - SketchType::CountMinSketchWithHeap => CMSHeapWrapper::from_snapshot(payload).map(drop), - SketchType::Unspecified => { - return Err("sketch.envelope requires a sketch.type".into()); - } - }; + if sketch_type == SketchType::Unspecified { + return Err("sketch.envelope requires a sketch.type".into()); + } + if sketch_type == SketchType::Exact { + // Exact windows can be large; their consumer decodes them once. + return Ok(()); + } + let decoded = crate::sketches::decode_snapshot(sketch_type, payload).map(drop); decoded.map_err(|error| format!("invalid {} sketch.envelope: {error}", sketch_type.name())) } diff --git a/asap-precompute-rs/src/otap/config.rs b/asap-precompute-rs/src/otap/config.rs index a48497a..d268d2e 100644 --- a/asap-precompute-rs/src/otap/config.rs +++ b/asap-precompute-rs/src/otap/config.rs @@ -114,6 +114,8 @@ pub struct PluginConfig { /// Quantiles emitted per series when `transmit_sketch = false` and the /// sketch is a quantile sketch (DDSketch / KLL). Ignored otherwise. pub quantiles: Vec, + /// Labels that identify a series; empty keys series by every label. + pub aggregate_by: Vec, } impl Default for PluginConfig { @@ -131,6 +133,7 @@ impl Default for PluginConfig { emit_window_stats: false, transmit_sketch: true, quantiles: Vec::new(), + aggregate_by: Vec::new(), } } } @@ -184,7 +187,7 @@ pub fn resolve(config: &PluginConfig) -> Result<(PrecomputeConfig, SketchDispatc ..Default::default() }, matchers: Vec::new(), - aggregate_by: Vec::new(), + aggregate_by: config.aggregate_by.clone(), transmit_sketch: config.transmit_sketch, quantiles: config.quantiles.clone(), sketch_params: config.sketch_params.clone(), diff --git a/asap-precompute-rs/src/otap/mod.rs b/asap-precompute-rs/src/otap/mod.rs index 71f02b3..838e0ba 100644 --- a/asap-precompute-rs/src/otap/mod.rs +++ b/asap-precompute-rs/src/otap/mod.rs @@ -10,12 +10,18 @@ mod schema; pub mod config; +#[cfg(feature = "otap-engine")] +pub mod answer_nodes; +#[cfg(feature = "otap-engine")] +pub mod baseline_nodes; #[cfg(feature = "otap-engine")] pub mod codec; #[cfg(feature = "otap-engine")] pub mod processor; #[cfg(feature = "otap-engine")] pub mod scenario_generator; +#[cfg(feature = "otap-engine")] +pub mod window_merge; pub use config::{ConfigError, PluginConfig, SketchDispatch}; pub use schema::{ diff --git a/asap-precompute-rs/src/otap/processor.rs b/asap-precompute-rs/src/otap/processor.rs index 5b899a9..52a107f 100644 --- a/asap-precompute-rs/src/otap/processor.rs +++ b/asap-precompute-rs/src/otap/processor.rs @@ -93,6 +93,7 @@ use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; use crate::config::PrecomputeConfigSet; use crate::envelope::SketchEnvelope; +use crate::observation::ObservationValueKind; use crate::otap::config::resolve as resolve_plugin_config; use crate::otap::PluginConfig; use crate::precompute::{Precompute, PrecomputeError, PrecomputeImpl}; @@ -151,6 +152,9 @@ pub struct AsapSketchesUserConfig { /// Quantiles emitted by DDSketch/KLL in estimate mode. #[serde(default)] pub quantiles: Vec, + /// Labels that identify a series; empty keys series by every label. + #[serde(default)] + pub aggregate_by: Vec, } impl AsapSketchesUserConfig { @@ -188,6 +192,7 @@ impl AsapSketchesUserConfig { emit_window_stats: self.emit_window_stats, transmit_sketch: self.transmit_sketch, quantiles: self.quantiles, + aggregate_by: self.aggregate_by, }; // Bounce the config through the resolver so config mistakes // (unsupported sketch_type, zero window) surface here rather @@ -203,9 +208,11 @@ const fn default_transmit_sketch() -> bool { true } +/// Tumbling windows close within a second of their end even when idle; +/// sliding windows emit once per slide. fn timer_period(config: &PluginConfig) -> Duration { if config.window_slide.is_zero() { - config.window_size + config.window_size.min(Duration::from_secs(1)) } else { config.window_slide } @@ -423,6 +430,18 @@ impl local::Processor for AsapSketchesProcessor { )) .await; } + // A tumbling window closes as soon as a point's event time + // passes its end, so no point lands in an earlier window. + if self.plugin_config.window_slide.is_zero() { + if let Some(first) = outcome + .observations + .iter() + .find(|o| o.value.kind != ObservationValueKind::Envelope) + { + let envs = self.precompute.tick(first.timestamp_ms); + self.emit_envelopes(&envs, effect_handler).await?; + } + } for obs in &outcome.observations { // Note: `observe` here handles both "genuine OTLP // metric" and "sketch shipped as binary inside an @@ -555,6 +574,16 @@ mod tests { assert!(parse("60s", "5s", true).is_err()); assert!(parse("5s", "5s", true).is_ok()); assert!(parse("5s", "0s", false).is_err()); + let tumbling = serde_json::from_value::( + serde_json::json!({"sketch_type":"kll","window_size":"60s","output_metric_name":"t"}), + ) + .unwrap() + .into_plugin_config() + .unwrap(); + assert_eq!( + super::timer_period(&tumbling), + std::time::Duration::from_secs(1) + ); assert!(parse("3ms", "1.5ms", false).is_err()); } // Unit tests confined to config-shape validation — full lifecycle diff --git a/asap-precompute-rs/src/otap/window_merge.rs b/asap-precompute-rs/src/otap/window_merge.rs new file mode 100644 index 0000000..6a5534b --- /dev/null +++ b/asap-precompute-rs/src/otap/window_merge.rs @@ -0,0 +1,291 @@ +//! `urn:asap:processor:window_merge`: merges sketches from many sources by +//! the window they describe, not by when they arrive. +//! +//! Each input envelope names its series (labels, sketch type, agg id) and its +//! window. Envelopes with the same series and window are merged. A window is +//! emitted once, `grace` after its end on this host's clock; envelopes for a +//! window already emitted are counted as late and dropped. Scalar (non-sketch) +//! inputs pass through unchanged. +//! +//! ```yaml +//! type: urn:asap:processor:window_merge +//! config: {grace: 5s} +//! ``` +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use async_trait::async_trait; +use linkme::distributed_slice; +use otel_arrow_dfe_config::node::NodeUserConfig; +use otel_arrow_dfe_engine::config::ProcessorConfig; +use otel_arrow_dfe_engine::context::PipelineContext; +use otel_arrow_dfe_engine::control::NodeControlMsg; +use otel_arrow_dfe_engine::error::Error; +use otel_arrow_dfe_engine::local::processor as local; +use otel_arrow_dfe_engine::message::Message; +use otel_arrow_dfe_engine::node::NodeId; +use otel_arrow_dfe_engine::processor::ProcessorWrapper; +use otel_arrow_dfe_engine::{MessageSourceLocalEffectHandlerExtension, ProcessorFactory}; +use otel_arrow_dfe_otap::pdata::OtapPdata; +use otel_arrow_dfe_otap::OTAP_PROCESSOR_FACTORIES; +use serde::Deserialize; + +use super::codec::{decode_pdata_to_observations, OtapSketchEncoder}; +use crate::envelope::SketchEnvelope; +use crate::observation::ObservationValueKind; +use crate::precompute::{PrecomputeError, Sketch}; +use crate::sketches::decode_snapshot; + +/// Registered URN. +pub const WINDOW_MERGE_URN: &str = "urn:asap:processor:window_merge"; + +/// Processor configuration. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct WindowMergeConfig { + /// How long after a window's end to wait for the last source. + #[serde(default = "default_grace", with = "humantime_serde")] + pub grace: Duration, +} + +fn default_grace() -> Duration { + Duration::from_secs(5) +} + +struct Pending { + envelope: SketchEnvelope, + sketch: Box, +} + +/// Windows waiting for their grace period, keyed by window end then series. +#[derive(Default)] +pub struct WindowMerger { + windows: BTreeMap>, + /// Every window ending at or before this has been emitted. + emitted_until: u64, + /// Envelopes that arrived after their window was emitted. + pub late: u64, +} + +fn series_key(e: &SketchEnvelope) -> String { + let mut labels: Vec<_> = e.labels.iter().map(|kv| (&kv.key, &kv.value)).collect(); + labels.sort(); + let mut resource: Vec<_> = e + .resource_labels + .iter() + .map(|kv| (&kv.key, &kv.value)) + .collect(); + resource.sort(); + format!( + "{}|{}|{}|{}|{:?}|{:?}", + e.metric_name, + e.sketch_type.name(), + e.agg_id, + e.window_start_ms, + resource, + labels + ) +} + +impl WindowMerger { + /// Adds one source's sketch for a window. + pub fn add(&mut self, envelope: SketchEnvelope) -> Result<(), PrecomputeError> { + if envelope.window_end_ms <= self.emitted_until { + self.late += 1; + return Ok(()); + } + let window = self.windows.entry(envelope.window_end_ms).or_default(); + match window.entry(series_key(&envelope)) { + std::collections::btree_map::Entry::Occupied(mut slot) => { + let pending = slot.get_mut(); + pending.sketch.merge_snapshot(&envelope.payload)?; + pending.envelope.count += envelope.count; + } + std::collections::btree_map::Entry::Vacant(slot) => { + let sketch = decode_snapshot(envelope.sketch_type, &envelope.payload)?; + slot.insert(Pending { envelope, sketch }); + } + } + Ok(()) + } + + /// Merged envelopes of every window that ended `grace` or more before `now_ms`. + pub fn due(&mut self, now_ms: u64, grace: Duration) -> Vec { + let cutoff = now_ms.saturating_sub(grace.as_millis() as u64); + let open = self.windows.split_off(&(cutoff + 1)); + let ready = std::mem::replace(&mut self.windows, open); + let mut out = Vec::new(); + for (end, series) in ready { + self.emitted_until = self.emitted_until.max(end); + for (_, mut p) in series { + if let Ok(payload) = p.sketch.snapshot() { + p.envelope.payload = payload; + out.push(p.envelope); + } + } + } + out + } +} + +struct WindowMergeProcessor { + config: WindowMergeConfig, + merger: WindowMerger, + encoder: OtapSketchEncoder, + timer_started: bool, +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +impl WindowMergeProcessor { + async fn emit( + &mut self, + envs: Vec, + effects: &mut local::EffectHandler, + ) -> Result<(), Error> { + if envs.is_empty() { + return Ok(()); + } + match self.encoder.encode(&envs) { + Ok(pdata) => effects.send_message_with_source_node(pdata).await?, + Err(e) => { + effects + .info(&format!( + "window_merge: dropping window after encode failure: {e}" + )) + .await + } + } + Ok(()) + } +} + +#[async_trait(?Send)] +impl local::Processor for WindowMergeProcessor { + async fn process( + &mut self, + msg: Message, + effects: &mut local::EffectHandler, + ) -> Result<(), Error> { + if !self.timer_started { + self.timer_started = effects + .start_periodic_timer(Duration::from_secs(1)) + .await + .is_ok(); + } + match msg { + Message::PData(pdata) => { + let observations = match decode_pdata_to_observations(pdata) { + Ok(outcome) => outcome.observations, + Err(e) => { + effects + .info(&format!("window_merge: dropping undecodable input: {e}")) + .await; + return Ok(()); + } + }; + let mut passthrough = Vec::new(); + for obs in observations { + match (obs.value.kind, obs.value.envelope) { + (ObservationValueKind::Envelope, Some(env)) if !env.payload.is_empty() => { + if let Err(e) = self.merger.add(*env) { + effects.info(&format!("window_merge: {e}")).await; + } + } + (ObservationValueKind::Envelope, Some(env)) => passthrough.push(*env), + _ => {} + } + } + self.emit(passthrough, effects).await + } + Message::Control(NodeControlMsg::TimerTick { .. }) => { + let due = self.merger.due(now_ms(), self.config.grace); + self.emit(due, effects).await + } + Message::Control(NodeControlMsg::Shutdown { .. }) => { + let all = self.merger.due(u64::MAX, Duration::ZERO); + self.emit(all, effects).await + } + _ => Ok(()), + } + } +} + +#[allow(unsafe_code)] +#[distributed_slice(OTAP_PROCESSOR_FACTORIES)] +static WINDOW_MERGE: ProcessorFactory = ProcessorFactory { + name: WINDOW_MERGE_URN, + create: |_: PipelineContext, + node: NodeId, + node_config: Arc, + processor_config: &ProcessorConfig, + _: &otel_arrow_dfe_engine::capability::registry::Capabilities| { + let config: WindowMergeConfig = serde_json::from_value(node_config.config.clone()) + .map_err(|e| otel_arrow_dfe_config::error::Error::InvalidUserConfig { + error: format!("window_merge: {e}"), + })?; + Ok(ProcessorWrapper::local( + WindowMergeProcessor { + config, + merger: WindowMerger::default(), + encoder: OtapSketchEncoder::default(), + timer_started: false, + }, + node, + node_config, + processor_config, + )) + }, + wiring_contract: otel_arrow_dfe_engine::wiring_contract::WiringContract::UNRESTRICTED, + validate_config: otel_arrow_dfe_config::validation::validate_typed_config::, +}; + +#[cfg(test)] +mod tests { + use super::*; + use crate::envelope::SketchType; + use crate::observation::KeyValue; + use crate::precompute::CardinalitySketch; + use crate::sketches::HLLWrapper; + + fn hll(values: std::ops::Range, start: u64) -> SketchEnvelope { + let mut sk = HLLWrapper::new(); + values.for_each(|v| sk.update(&v.to_le_bytes())); + SketchEnvelope { + sketch_type: SketchType::HLLSketch, + labels: vec![KeyValue::new("attribute.key", "user.id")], + window_start_ms: start, + window_end_ms: start + 60_000, + payload: sk.snapshot().unwrap(), + metric_name: "asap.distinct".into(), + count: 1000, + ..Default::default() + } + } + + #[test] + fn merges_sources_per_window_after_grace() { + let mut m = WindowMerger::default(); + m.add(hll(0..1000, 0)).unwrap(); + m.add(hll(500..1500, 0)).unwrap(); + m.add(hll(0..10, 60_000)).unwrap(); + let grace = Duration::from_secs(5); + assert!(m.due(64_000, grace).is_empty()); + let out = m.due(65_000, grace); + assert_eq!(out.len(), 1); + assert_eq!(out[0].count, 2000); + let merged = HLLWrapper::from_snapshot(&out[0].payload).unwrap(); + let d = merged.estimate_cardinality(); + assert!((d - 1500.0).abs() < 1500.0 * 0.05, "{d}"); + // A source arriving after its window was emitted is late. + m.add(hll(0..5, 0)).unwrap(); + assert_eq!(m.late, 1); + assert_eq!(m.due(125_000, grace).len(), 1); + } +} diff --git a/asap-precompute-rs/src/sketches/mod.rs b/asap-precompute-rs/src/sketches/mod.rs index 023b070..8a77948 100644 --- a/asap-precompute-rs/src/sketches/mod.rs +++ b/asap-precompute-rs/src/sketches/mod.rs @@ -19,3 +19,31 @@ pub use countsketch::{CountSketchObserver, CountSketchWrapper}; pub use ddsketch::{DDSketchObserver, DDSketchWrapper}; pub use hll::{HLLObserver, HLLWrapper}; pub use kll::{KLLObserver, KLLWrapper}; + +use crate::envelope::SketchType; +use crate::precompute::{PrecomputeError, Sketch}; + +/// Decodes an ASAPv1 snapshot into the wrapper for `sketch_type`. +pub fn decode_snapshot( + sketch_type: SketchType, + bytes: &[u8], +) -> Result, PrecomputeError> { + Ok(match sketch_type { + SketchType::DDSketch => Box::new(DDSketchWrapper::from_snapshot(bytes)?), + SketchType::KLLSketch => Box::new(KLLWrapper::from_snapshot(bytes)?), + SketchType::HLLSketch => Box::new(HLLWrapper::from_snapshot(bytes)?), + SketchType::CountSketch => Box::new(CountSketchWrapper::from_snapshot(bytes)?), + SketchType::CountMinSketch => Box::new(CMSWrapper::from_snapshot(bytes)?), + SketchType::CountMinSketchWithHeap => Box::new(CMSHeapWrapper::from_snapshot(bytes)?), + SketchType::Unspecified => { + return Err(PrecomputeError::Other( + "sketch.envelope requires a sketch.type".into(), + )) + } + SketchType::Exact => { + return Err(PrecomputeError::Other( + "an exact window is not a sketch".into(), + )) + } + }) +} diff --git a/deploy/collector.Dockerfile b/deploy/collector.Dockerfile new file mode 100644 index 0000000..e01b4e4 --- /dev/null +++ b/deploy/collector.Dockerfile @@ -0,0 +1,4 @@ +FROM ubuntu:24.04 +RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates && rm -rf /var/lib/apt/lists/* +COPY asap-collector /usr/local/bin/asap-collector +ENTRYPOINT ["/usr/local/bin/asap-collector"] diff --git a/deploy/telemetry-functions/README.md b/deploy/telemetry-functions/README.md new file mode 100644 index 0000000..7c84c72 --- /dev/null +++ b/deploy/telemetry-functions/README.md @@ -0,0 +1,57 @@ +# Telemetry functions on real collector containers + +Every stage is the official OTAP dataflow engine (`asap-collector`: the +upstream controller with the ASAP nodes linked in), one container per stage, +each pinned to its own CPU, connected by OTLP/HTTP over the compose network. + +Three pipelines run side by side over the same input and share no processor: + +```text +sketch: generator-a/b -> branch-a/b (sketch per window) -> merge -> detector -> Prometheus +exact: exact-generator-a/b -> exact-branch-a/b (exact window) -> exact-merge -> exact-detector -> Prometheus +raw: raw-generator-a/b -> raw-branch-a/b (forward) -> raw-merge (forward) -> raw-sink -> loki-raw +``` + +The two generators of each pipeline (sources a and b) run the same configs +with the same scenario start (`SCENARIO_START`), and every point is a function +of scenario time, so all three pipelines receive identical points. + +| Stage | Config | What it runs | +| --- | --- | --- | +| generator-a/b, exact-generator-a/b, raw-generator-a/b | `pipelines/generator-*.yaml` | `scenario_generator`: seeded latency points over 20 routes (temporary stand-in for the official traffic generator) | +| branch-a/b | `pipelines/branch.yaml` | `asap_sketches`: one KLL per route per 1-minute window | +| merge | `pipelines/merge.yaml` | `window_merge`: merges both sources' sketches per window, 5 s after the window ends | +| detector | `pipelines/detector.yaml` | `sketch_detector`: p50/p99 per route with lower/upper bounds (`method="sketch"`) | +| exact-branch-a/b | `pipelines/exact-branch.yaml` | `exact_branch`: every value per route per 1-minute window | +| exact-merge | `common/exact-merge.yaml` | `exact_merge`: adds both sources' exact windows, 5 s after the window ends | +| exact-detector | `pipelines/exact-detector.yaml` | `exact_detector`: exact p50/p99 per route (`method="exact"`): the ground truth | +| raw-branch-a/b, raw-merge | `common/raw-*.yaml` | forward raw points (OTLP receiver -> exporter) | +| raw-sink | `common/raw-sink.yaml` | `raw_sink`: writes every raw point as one log record to `loki-raw` and reports when `loki-raw` acknowledged each window's last point | + +Prometheus receives the answers over OTLP (`asap_quantile`, `asap_points`, +labelled `method="sketch"` or `"exact"`) and scrapes each collector's engine +telemetry (`/api/v1/metrics?format=prometheus`): CPU time per pipeline +(`cpu_time_total`), resident memory (`memory_rss`) and payload bytes per +receiver, labelled with the stage. + +Every pipeline reports its window completion delay as `asap_answer_delay` +(seconds from a window's end until its result is complete at the end of the +pipeline): sketch and exact when their detector emits the window's answers, +raw when `loki-raw` has acknowledged the window's last point. raw-sink also +reports each point's time from its event time to that acknowledgement +(`asap_raw_transit`, p50 and p99) and points `loki-raw` rejected +(`asap_raw_failed`). + +## Run + +Linux, Docker with Compose, 18 free CPUs (0-17 by default; `CPU_BRANCH_A=20` +etc. move a stage) and memory for `loki-raw` (capped at `LOKI_RAW_MEMORY`, +default 16g). From the repository root: + +```sh +deploy/telemetry-functions/run.sh # DOCKER_COMMAND="sudo docker" if needed +``` + +Open . The first answers appear about +65 seconds after the first full minute. Stop with +`docker compose -f deploy/telemetry-functions/compose.yaml down`. diff --git a/deploy/telemetry-functions/common/exact-merge.yaml b/deploy/telemetry-functions/common/exact-merge.yaml new file mode 100644 index 0000000..e8f44de --- /dev/null +++ b/deploy/telemetry-functions/common/exact-merge.yaml @@ -0,0 +1,34 @@ +# Exact pipeline, stage 2: add up both sources' exact windows, 5 s after +# each window ends. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false, max_request_body_size: 1GiB} + merge: + type: urn:asap:processor:exact_merge + config: {grace: 5s} + to_detector: + type: exporter:otlp_http + config: + endpoint: http://exact-detector:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: merge} + - {from: merge, to: to_detector} diff --git a/deploy/telemetry-functions/common/raw-branch.yaml b/deploy/telemetry-functions/common/raw-branch.yaml new file mode 100644 index 0000000..ea1b866 --- /dev/null +++ b/deploy/telemetry-functions/common/raw-branch.yaml @@ -0,0 +1,29 @@ +# Raw pipeline, stage 1 at each source: forward every raw point. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + to_merge: + type: exporter:otlp_http + config: + endpoint: http://raw-merge:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: to_merge} diff --git a/deploy/telemetry-functions/common/raw-merge.yaml b/deploy/telemetry-functions/common/raw-merge.yaml new file mode 100644 index 0000000..e9d3664 --- /dev/null +++ b/deploy/telemetry-functions/common/raw-merge.yaml @@ -0,0 +1,29 @@ +# Raw pipeline, stage 2: forward both sources' raw points to one place. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + to_sink: + type: exporter:otlp_http + config: + endpoint: http://raw-sink:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: to_sink} diff --git a/deploy/telemetry-functions/common/raw-sink.yaml b/deploy/telemetry-functions/common/raw-sink.yaml new file mode 100644 index 0000000..fa34ce2 --- /dev/null +++ b/deploy/telemetry-functions/common/raw-sink.yaml @@ -0,0 +1,46 @@ +# Raw pipeline, end: write every raw point as a log record to loki-raw, and +# report per window when loki-raw acknowledged its last point +# (asap.answer.delay, method="raw") and each point's time to acknowledgement. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + sink: + type: urn:asap:processor:raw_sink + outputs: [logs, latency] + config: {window: 60s, grace: 5s} + to_loki_raw: + type: exporter:otlp_http + config: + endpoint: http://loki-raw:3100 + logs_endpoint: http://loki-raw:3100/otlp/v1/logs + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + to_prometheus: + type: exporter:otlp_http + config: + endpoint: http://prometheus:9090 + metrics_endpoint: http://prometheus:9090/api/v1/otlp/v1/metrics + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: sink} + - {from: 'sink["logs"]', to: to_loki_raw} + - {from: 'sink["latency"]', to: to_prometheus} diff --git a/deploy/telemetry-functions/compose.yaml b/deploy/telemetry-functions/compose.yaml new file mode 100644 index 0000000..8931593 --- /dev/null +++ b/deploy/telemetry-functions/compose.yaml @@ -0,0 +1,166 @@ +# Three pipelines over the same generated input, sharing no processor: +# sketch: generator-a/b -> branch-a/b (sketch) -> merge -> detector +# exact: exact-generator-a/b -> exact-branch-a/b (exact window) -> +# exact-merge -> exact-detector +# raw: raw-generator-a/b -> raw-branch-a/b -> raw-merge -> raw-sink +# (every raw point into loki-raw) +# Each stage is the official OTAP engine (asap-collector) in its own container, +# pinned to one CPU, talking OTLP/HTTP over the compose network. The generators +# of a source share one config (`downstream` is their pipeline's first stage) +# and SCENARIO_START, so all three pipelines get identical points. +# ./pipelines holds the use case's stage configs, ./common the stages that do +# not depend on it; CPU_* variables move a stage to another core. Prometheus, +# Grafana and Loki are the observers. +x-collector: &collector + image: asap-collector:local + restart: "no" +x-generator: &generator + <<: *collector + environment: + ASAP_SCENARIO_START: "${SCENARIO_START:-}" + +services: + # sketch pipeline + generator-a: + <<: *generator + command: ["--config", "/pipelines/generator-a.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_GENERATOR_A:-0}" + links: ["branch-a:downstream"] + depends_on: [branch-a] + generator-b: + <<: *generator + command: ["--config", "/pipelines/generator-b.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_GENERATOR_B:-1}" + links: ["branch-b:downstream"] + depends_on: [branch-b] + branch-a: + <<: *collector + command: ["--config", "/pipelines/branch.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_BRANCH_A:-2}" + depends_on: [merge] + branch-b: + <<: *collector + command: ["--config", "/pipelines/branch.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_BRANCH_B:-3}" + depends_on: [merge] + merge: + <<: *collector + command: ["--config", "/pipelines/merge.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_MERGE:-4}" + depends_on: [detector] + detector: + <<: *collector + command: ["--config", "/pipelines/detector.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_DETECTOR:-5}" + depends_on: [prometheus] + # exact pipeline + exact-generator-a: + <<: *generator + command: ["--config", "/pipelines/generator-a.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_EXACT_GENERATOR_A:-6}" + links: ["exact-branch-a:downstream"] + depends_on: [exact-branch-a] + exact-generator-b: + <<: *generator + command: ["--config", "/pipelines/generator-b.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_EXACT_GENERATOR_B:-7}" + links: ["exact-branch-b:downstream"] + depends_on: [exact-branch-b] + exact-branch-a: + <<: *collector + command: ["--config", "/pipelines/exact-branch.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_EXACT_BRANCH_A:-8}" + depends_on: [exact-merge] + exact-branch-b: + <<: *collector + command: ["--config", "/pipelines/exact-branch.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_EXACT_BRANCH_B:-9}" + depends_on: [exact-merge] + exact-merge: + <<: *collector + command: ["--config", "/common/exact-merge.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_EXACT_MERGE:-10}" + depends_on: [exact-detector] + exact-detector: + <<: *collector + command: ["--config", "/pipelines/exact-detector.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_EXACT_DETECTOR:-11}" + depends_on: [prometheus] + # raw pipeline + raw-generator-a: + <<: *generator + command: ["--config", "/pipelines/generator-a.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_RAW_GENERATOR_A:-12}" + links: ["raw-branch-a:downstream"] + depends_on: [raw-branch-a] + raw-generator-b: + <<: *generator + command: ["--config", "/pipelines/generator-b.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_RAW_GENERATOR_B:-13}" + links: ["raw-branch-b:downstream"] + depends_on: [raw-branch-b] + raw-branch-a: + <<: *collector + command: ["--config", "/common/raw-branch.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_RAW_BRANCH_A:-14}" + depends_on: [raw-merge] + raw-branch-b: + <<: *collector + command: ["--config", "/common/raw-branch.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_RAW_BRANCH_B:-15}" + depends_on: [raw-merge] + raw-merge: + <<: *collector + command: ["--config", "/common/raw-merge.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_RAW_MERGE:-16}" + depends_on: [raw-sink] + raw-sink: + <<: *collector + command: ["--config", "/common/raw-sink.yaml"] + volumes: ["./pipelines:/pipelines:ro", "./common:/common:ro"] + cpuset: "${CPU_RAW_SINK:-17}" + depends_on: [prometheus, loki-raw] + + prometheus: + image: prom/prometheus:v3.5.0 + command: + - --config.file=/etc/prometheus/prometheus.yml + - --web.enable-otlp-receiver + ports: ["127.0.0.1:${PROMETHEUS_PORT:-9090}:9090"] + volumes: ["./prometheus.yml:/etc/prometheus/prometheus.yml:ro"] + # The raw pipeline's store, apart from the Loki with the events. + loki-raw: + image: grafana/loki:3.5.0 + command: ["-config.file=/etc/loki/loki-raw.yaml"] + ports: ["127.0.0.1:${LOKI_RAW_PORT:-3101}:3100"] + volumes: ["./loki-raw.yaml:/etc/loki/loki-raw.yaml:ro"] + mem_limit: "${LOKI_RAW_MEMORY:-16g}" + grafana: + image: grafana/grafana:12.1.0 + ports: ["127.0.0.1:${GRAFANA_PORT:-3000}:3000"] + environment: + GF_AUTH_ANONYMOUS_ENABLED: "true" + GF_AUTH_ANONYMOUS_ORG_ROLE: Viewer + volumes: + - ./grafana/provisioning:/etc/grafana/provisioning:ro + - ./grafana/dashboards:/var/lib/grafana/dashboards:ro + loki: + image: grafana/loki:3.5.0 + command: ["-config.file=/etc/loki/local-config.yaml"] diff --git a/deploy/telemetry-functions/grafana/dashboards/latency.json b/deploy/telemetry-functions/grafana/dashboards/latency.json new file mode 100644 index 0000000..ddbd86e --- /dev/null +++ b/deploy/telemetry-functions/grafana/dashboards/latency.json @@ -0,0 +1,660 @@ +{ + "uid": "asap-latency", + "title": "Latency quantiles: sketch vs exact (real collector containers)", + "schemaVersion": 39, + "version": 1, + "refresh": "10s", + "time": { + "from": "now-30m", + "to": "now" + }, + "templating": { + "list": [ + { + "name": "route", + "type": "query", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "query": { + "query": "label_values(asap_quantile, http_route)", + "refId": "route" + }, + "refresh": 2, + "current": { + "text": "http.route-0", + "value": "http.route-0" + } + } + ] + }, + "panels": [ + { + "type": "text", + "title": "Pipeline", + "description": "Three pipelines that share no processor, fed identical points by their own generators (same seed and scenario start). Each box is one collector container pinned to its own CPU; arrows are OTLP/HTTP between containers.", + "gridPos": { + "h": 10, + "w": 24, + "x": 0, + "y": 0 + }, + "options": { + "mode": "html", + "content": "
generator-a/b
sources A and B
→
branch-a/b
Stage 1 per source: KLL (k = 200) per route per 1-min window
→
merge
Stage 2: merges both sources' sketches per window
→
detector
p50/p99 with bounds
→
Prometheus + Loki
method="sketch"
exact-generator-a/b
same points
→
exact-branch-a/b
Stage 1 per source: every value per route per 1-min window
→
exact-merge
Stage 2: adds both sources' exact windows
→
exact-detector
exact p50/p99
→
Prometheus + Loki
method="exact" (ground truth)
raw-generator-a/b
same points
→
raw-branch-a/b
Stage 1 per source: forwards every raw point
→
raw-merge
Stage 2: forwards both sources' raw points
→
raw-sink
writes every raw point; measures when the window's last point is stored
→
loki-raw
one log record per raw point
" + }, + "id": 1 + }, + { + "id": 2, + "type": "timeseries", + "title": "p99 latency for $route: sketch with bound vs exact", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 10 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_quantile{quantile=\"0.99\",http_route=\"$route\",method=\"sketch\",bound=\"estimate\"}", + "legendFormat": "sketch" + }, + { + "refId": "B", + "expr": "asap_quantile{quantile=\"0.99\",http_route=\"$route\",method=\"sketch\",bound=\"lower\"}", + "legendFormat": "sketch lower" + }, + { + "refId": "C", + "expr": "asap_quantile{quantile=\"0.99\",http_route=\"$route\",method=\"sketch\",bound=\"upper\"}", + "legendFormat": "sketch upper" + }, + { + "refId": "D", + "expr": "asap_quantile{quantile=\"0.99\",http_route=\"$route\",method=\"exact\"}", + "legendFormat": "exact" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "ms" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch upper" + }, + "properties": [ + { + "id": "custom.fillBelowTo", + "value": "sketch lower" + }, + { + "id": "custom.lineWidth", + "value": 0 + }, + { + "id": "custom.fillOpacity", + "value": 20 + }, + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sketch lower" + }, + "properties": [ + { + "id": "custom.lineWidth", + "value": 0 + }, + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "orange" + } + }, + { + "id": "custom.lineStyle", + "value": { + "fill": "dash", + "dash": [ + 6, + 4 + ] + } + } + ] + } + ] + } + }, + { + "id": 3, + "type": "timeseries", + "title": "p50 latency for $route: sketch with bound vs exact", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 10 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_quantile{quantile=\"0.5\",http_route=\"$route\",method=\"sketch\",bound=\"estimate\"}", + "legendFormat": "sketch" + }, + { + "refId": "B", + "expr": "asap_quantile{quantile=\"0.5\",http_route=\"$route\",method=\"sketch\",bound=\"lower\"}", + "legendFormat": "sketch lower" + }, + { + "refId": "C", + "expr": "asap_quantile{quantile=\"0.5\",http_route=\"$route\",method=\"sketch\",bound=\"upper\"}", + "legendFormat": "sketch upper" + }, + { + "refId": "D", + "expr": "asap_quantile{quantile=\"0.5\",http_route=\"$route\",method=\"exact\"}", + "legendFormat": "exact" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "ms" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch upper" + }, + "properties": [ + { + "id": "custom.fillBelowTo", + "value": "sketch lower" + }, + { + "id": "custom.lineWidth", + "value": 0 + }, + { + "id": "custom.fillOpacity", + "value": 20 + }, + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sketch lower" + }, + "properties": [ + { + "id": "custom.lineWidth", + "value": 0 + }, + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "blue" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "orange" + } + }, + { + "id": "custom.lineStyle", + "value": { + "fill": "dash", + "dash": [ + 6, + 4 + ] + } + } + ] + } + ] + } + }, + { + "id": 4, + "type": "timeseries", + "title": "Relative error of the sketch p99 vs exact (routes with > 20,000 points per window)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 18 + }, + "targets": [ + { + "refId": "A", + "expr": "abs(asap_quantile{quantile=\"0.99\",method=\"sketch\",bound=\"estimate\"} - on(http_route) asap_quantile{quantile=\"0.99\",method=\"exact\"}) / on(http_route) asap_quantile{quantile=\"0.99\",method=\"exact\"} and on(http_route) asap_points{method=\"exact\"} > 20000", + "legendFormat": "{{http_route}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "percentunit" + }, + "overrides": [] + } + }, + { + "type": "timeseries", + "title": "Window completion delay: seconds from a window's end until its result is complete", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 26 + }, + "targets": [ + { + "refId": "A", + "expr": "avg by (method) (asap_answer_delay)", + "legendFormat": "{{method}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "s" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#5794f2" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#ff9830" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "raw" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#8e8e8e" + } + } + ] + } + ] + }, + "description": "sketch and exact: the detector has emitted the window's answers. raw: loki-raw has acknowledged the window's last raw point.", + "id": 5 + }, + { + "type": "timeseries", + "title": "CPU cores per pipeline (all stages but the generators)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 26 + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (pipeline) (label_replace(label_replace(rate(cpu_time_total{job=\"collectors\",stage!~\".*generator.*\",otel_scope_pipeline_id=\"main\"}[1m]), \"pipeline\", \"sketch\", \"stage\", \".*\"), \"pipeline\", \"$1\", \"stage\", \"(exact|raw)-.*\"))", + "legendFormat": "{{pipeline}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "short" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#5794f2" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#ff9830" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "raw" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#8e8e8e" + } + } + ] + } + ] + }, + "id": 6 + }, + { + "type": "timeseries", + "title": "Bytes received per pipeline (all stages but the generators), log scale", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 34 + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (pipeline) (label_replace(label_replace(rate(payload_size_bytes_total{job=\"collectors\",stage!~\".*generator.*\",otel_scope_node_type=\"receiver\"}[1m]), \"pipeline\", \"sketch\", \"stage\", \".*\"), \"pipeline\", \"$1\", \"stage\", \"(exact|raw)-.*\")) * 8", + "legendFormat": "{{pipeline}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true, + "scaleDistribution": { + "type": "log", + "log": 10 + } + }, + "unit": "bps" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#5794f2" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#ff9830" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "raw" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#8e8e8e" + } + } + ] + } + ] + }, + "id": 7 + }, + { + "type": "timeseries", + "title": "Resident memory per pipeline (all stages but the generators)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 34 + }, + "targets": [ + { + "refId": "A", + "expr": "sum by (pipeline) (label_replace(label_replace(max by (stage) (memory_rss{job=\"collectors\",stage!~\".*generator.*\"}), \"pipeline\", \"sketch\", \"stage\", \".*\"), \"pipeline\", \"$1\", \"stage\", \"(exact|raw)-.*\"))", + "legendFormat": "{{pipeline}}" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "spanNulls": true + }, + "unit": "bytes" + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "sketch" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#5794f2" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "exact" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#ff9830" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "raw" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "#8e8e8e" + } + } + ] + } + ] + }, + "id": 8 + } + ] +} diff --git a/deploy/telemetry-functions/grafana/provisioning/dashboards/asap.yaml b/deploy/telemetry-functions/grafana/provisioning/dashboards/asap.yaml new file mode 100644 index 0000000..bf894a9 --- /dev/null +++ b/deploy/telemetry-functions/grafana/provisioning/dashboards/asap.yaml @@ -0,0 +1,3 @@ +apiVersion: 1 +providers: + - {name: ASAP telemetry functions, folder: ASAP, type: file, options: {path: /var/lib/grafana/dashboards}} diff --git a/deploy/telemetry-functions/grafana/provisioning/datasources/datasources.yaml b/deploy/telemetry-functions/grafana/provisioning/datasources/datasources.yaml new file mode 100644 index 0000000..06fe817 --- /dev/null +++ b/deploy/telemetry-functions/grafana/provisioning/datasources/datasources.yaml @@ -0,0 +1,4 @@ +apiVersion: 1 +datasources: + - {name: Prometheus, uid: asap-prometheus, type: prometheus, access: proxy, url: http://prometheus:9090, isDefault: true, editable: false} + - {name: Loki, uid: asap-loki, type: loki, access: proxy, url: http://loki:3100, editable: false} diff --git a/deploy/telemetry-functions/loki-raw.yaml b/deploy/telemetry-functions/loki-raw.yaml new file mode 100644 index 0000000..04ff483 --- /dev/null +++ b/deploy/telemetry-functions/loki-raw.yaml @@ -0,0 +1,35 @@ +# Store of the raw pipeline: one log record per raw point. Limits are raised +# so that ingestion speed, not a rate limit, decides what the pipeline sees. +auth_enabled: false +server: + http_listen_port: 3100 + grpc_server_max_recv_msg_size: 1073741824 +common: + instance_addr: 127.0.0.1 + path_prefix: /loki + storage: + filesystem: + chunks_directory: /loki/chunks + rules_directory: /loki/rules + replication_factor: 1 + ring: + kvstore: + store: inmemory +schema_config: + configs: + - from: 2020-10-24 + store: tsdb + object_store: filesystem + schema: v13 + index: + prefix: index_ + period: 24h +limits_config: + ingestion_rate_mb: 1024 + ingestion_burst_size_mb: 2048 + per_stream_rate_limit: 1GB + per_stream_rate_limit_burst: 2GB + reject_old_samples: false + max_structured_metadata_entries_count: 512 +analytics: + reporting_enabled: false diff --git a/deploy/telemetry-functions/pipelines/branch.yaml b/deploy/telemetry-functions/pipelines/branch.yaml new file mode 100644 index 0000000..8d2d148 --- /dev/null +++ b/deploy/telemetry-functions/pipelines/branch.yaml @@ -0,0 +1,40 @@ +# Sketch pipeline, stage 1 at each source: one KLL per route per 1-minute +# window goes to merge. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + sketch: + type: urn:asap:processor:asap_sketches + config: + sketch_type: kll + window_size: 60s + output_metric_name: http.server.request.duration + aggregate_by: [http.route] + omit_resource_attrs: true + sketch_params: {k: 200} + to_merge: + type: exporter:otlp_http + config: + endpoint: http://merge:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: sketch} + - {from: sketch, to: to_merge} diff --git a/deploy/telemetry-functions/pipelines/detector.yaml b/deploy/telemetry-functions/pipelines/detector.yaml new file mode 100644 index 0000000..8c70015 --- /dev/null +++ b/deploy/telemetry-functions/pipelines/detector.yaml @@ -0,0 +1,34 @@ +# Answers (with bounds) from each merged window, written to Prometheus. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + detector: + type: urn:asap:processor:sketch_detector + config: {quantiles: [0.5, 0.99]} + to_prometheus: + type: exporter:otlp_http + config: + endpoint: http://prometheus:9090 + metrics_endpoint: http://prometheus:9090/api/v1/otlp/v1/metrics + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: detector} + - {from: detector, to: to_prometheus} diff --git a/deploy/telemetry-functions/pipelines/exact-branch.yaml b/deploy/telemetry-functions/pipelines/exact-branch.yaml new file mode 100644 index 0000000..a061d73 --- /dev/null +++ b/deploy/telemetry-functions/pipelines/exact-branch.yaml @@ -0,0 +1,34 @@ +# Exact pipeline, stage 1 at each source: every value per route per 1-minute window +# goes to exact-merge. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + branch: + type: urn:asap:processor:exact_branch + config: {window: 60s, group_by: [http.route]} + to_merge: + type: exporter:otlp_http + config: + endpoint: http://exact-merge:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: branch} + - {from: branch, to: to_merge} diff --git a/deploy/telemetry-functions/pipelines/exact-detector.yaml b/deploy/telemetry-functions/pipelines/exact-detector.yaml new file mode 100644 index 0000000..740c6b2 --- /dev/null +++ b/deploy/telemetry-functions/pipelines/exact-detector.yaml @@ -0,0 +1,35 @@ +# Exact pipeline, end: exact answers of each merged window, the +# ground truth (method="exact"). +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false, max_request_body_size: 1GiB} + detector: + type: urn:asap:processor:exact_detector + config: {quantiles: [0.5, 0.99]} + to_prometheus: + type: exporter:otlp_http + config: + endpoint: http://prometheus:9090 + metrics_endpoint: http://prometheus:9090/api/v1/otlp/v1/metrics + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: detector} + - {from: detector, to: to_prometheus} diff --git a/deploy/telemetry-functions/pipelines/generator-a.yaml b/deploy/telemetry-functions/pipelines/generator-a.yaml new file mode 100644 index 0000000..d63b7bd --- /dev/null +++ b/deploy/telemetry-functions/pipelines/generator-a.yaml @@ -0,0 +1,34 @@ +# Source a: seeded latency traffic over 20 routes. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + generator: + type: urn:asap:receiver:scenario_generator + config: + seed: 1 + signals_per_second: 20000 + metric_name: http.server.request.duration + resource_attributes: {service.name: checkout, source: "a"} + attributes: + - {key: http.route, cardinality: 20, zipf: 1.1} + value: {lognormal: {median: 50.0, sigma: 0.6}} + export: + type: exporter:otlp_http + config: + endpoint: http://downstream:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: generator, to: export} diff --git a/deploy/telemetry-functions/pipelines/generator-b.yaml b/deploy/telemetry-functions/pipelines/generator-b.yaml new file mode 100644 index 0000000..081b113 --- /dev/null +++ b/deploy/telemetry-functions/pipelines/generator-b.yaml @@ -0,0 +1,34 @@ +# Source b: seeded latency traffic over 20 routes. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + generator: + type: urn:asap:receiver:scenario_generator + config: + seed: 2 + signals_per_second: 20000 + metric_name: http.server.request.duration + resource_attributes: {service.name: checkout, source: "b"} + attributes: + - {key: http.route, cardinality: 20, zipf: 1.1} + value: {lognormal: {median: 50.0, sigma: 0.6}} + export: + type: exporter:otlp_http + config: + endpoint: http://downstream:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: generator, to: export} diff --git a/deploy/telemetry-functions/pipelines/merge.yaml b/deploy/telemetry-functions/pipelines/merge.yaml new file mode 100644 index 0000000..cb392f2 --- /dev/null +++ b/deploy/telemetry-functions/pipelines/merge.yaml @@ -0,0 +1,33 @@ +# Stage 2: merge every source's sketch for a window, 5 s after it ends. +version: otel_dataflow/v1 +engine: + http_admin: + bind_address: 0.0.0.0:8085 +policies: + resources: + core_allocation: + type: core_count + count: 1 +groups: + default: + pipelines: + main: + nodes: + receiver: + type: receiver:otlp + config: + protocols: + http: {listening_addr: "0.0.0.0:4318", wait_for_result: false} + merge: + type: urn:asap:processor:window_merge + config: {grace: 5s} + to_detector: + type: exporter:otlp_http + config: + endpoint: http://detector:4318 + client_pool_size: 1 + max_in_flight: 1 + http: {timeout: 30s, compression: none} + connections: + - {from: receiver, to: merge} + - {from: merge, to: to_detector} diff --git a/deploy/telemetry-functions/prometheus.yml b/deploy/telemetry-functions/prometheus.yml new file mode 100644 index 0000000..b2eff75 --- /dev/null +++ b/deploy/telemetry-functions/prometheus.yml @@ -0,0 +1,20 @@ +global: + scrape_interval: 5s +scrape_configs: + # Each collector's own engine telemetry: CPU time per pipeline, RSS, and + # per-node message and payload counters. stage = the compose service. + - job_name: collectors + metrics_path: /api/v1/metrics + params: {format: [prometheus]} + static_configs: + - targets: [generator-a:8085, generator-b:8085, branch-a:8085, branch-b:8085, merge:8085, + detector:8085, exact-generator-a:8085, exact-generator-b:8085, + exact-branch-a:8085, exact-branch-b:8085, exact-merge:8085, + exact-detector:8085, raw-generator-a:8085, raw-generator-b:8085, + raw-branch-a:8085, raw-branch-b:8085, raw-merge:8085, raw-sink:8085] + relabel_configs: + - source_labels: [__address__] + regex: (.+):8085 + target_label: stage +# Answers (sketch, exact) and the raw pipeline's latency arrive through the +# OTLP receiver. diff --git a/deploy/telemetry-functions/run.sh b/deploy/telemetry-functions/run.sh new file mode 100755 index 0000000..5f27c09 --- /dev/null +++ b/deploy/telemetry-functions/run.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# Build asap-collector, package it, and start every stage plus the observers. +# DOCKER_COMMAND="sudo -n docker" when Docker needs sudo. Stop with: +# $DOCKER_COMMAND compose -f deploy/telemetry-functions/compose.yaml down +set -euo pipefail +repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd) +cd "$repo_root" +target=${CARGO_TARGET_DIR:-"$repo_root/asap-precompute-rs/target"} +read -r -a docker <<< "${DOCKER_COMMAND:-docker}" +cargo build --manifest-path asap-precompute-rs/Cargo.toml --release --features collector --bin asap-collector +context=$(mktemp -d) +trap 'rm -rf "$context"' EXIT +cp "$target/release/asap-collector" "$context/" +strip --strip-debug "$context/asap-collector" +"${docker[@]}" build -q -f deploy/collector.Dockerfile -t asap-collector:local "$context" +# Every generator counts scenario time from this second, so the three +# pipelines get identical points. +export SCENARIO_START=${SCENARIO_START:-$(($(date +%s) + 10))} +"${docker[@]}" compose -f deploy/telemetry-functions/compose.yaml up -d +echo "Grafana: http://localhost:${GRAFANA_PORT:-3000}/d/asap-latency" diff --git a/docs/images/latency-grafana.png b/docs/images/latency-grafana.png new file mode 100644 index 0000000..78e2026 Binary files /dev/null and b/docs/images/latency-grafana.png differ