From 3ae555565aeb83cca93b355b5dbd4895fde77bcb Mon Sep 17 00:00:00 2001 From: Dominik Werder Date: Sun, 26 Nov 2023 17:22:28 +0100 Subject: [PATCH] Support omitting the initial channels list --- .cargo/cargo-lock | 208 +++++++++++++++++---------------- daqingest/src/bin/daqingest.rs | 2 +- daqingest/src/daemon.rs | 74 ++++++------ daqingest/src/opts.rs | 4 +- netfetch/Cargo.toml | 1 - netfetch/src/ca/conn.rs | 30 +++-- netfetch/src/ca/connset.rs | 136 +++++++++++---------- netfetch/src/conf.rs | 58 ++++----- netfetch/src/daemon_common.rs | 4 +- netfetch/src/metrics.rs | 87 ++++++++++---- scywr/src/insertworker.rs | 44 ++++--- stats/src/stats.rs | 6 +- 12 files changed, 367 insertions(+), 287 deletions(-) diff --git a/.cargo/cargo-lock b/.cargo/cargo-lock index f6ab8bb..df08071 100644 --- a/.cargo/cargo-lock +++ b/.cargo/cargo-lock @@ -137,12 +137,12 @@ dependencies = [ [[package]] name = "async-channel" -version = "2.0.0" +version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "336d835910fab747186c56586562cb46f42809c2843ef3a84f47509009522838" +checksum = "d37875bd9915b7d67c2f117ea2c30a0989874d0b2cb694fe25403c85763c0c9e" dependencies = [ "concurrent-queue", - "event-listener 3.0.1", + "event-listener 3.1.0", "event-listener-strategy", "futures-core", "pin-project-lite", @@ -156,7 +156,7 @@ checksum = "a66537f1bb974b254c98ed142ff995236e81b9d0fe4db0575f46612cb15eb0f9" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -238,12 +238,6 @@ dependencies = [ "rustc-demangle", ] -[[package]] -name = "base64" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8" - [[package]] name = "base64" version = "0.21.5" @@ -254,7 +248,7 @@ checksum = "35636a1494ede3b646cc98f74f8e62c773a38a659ebc777a2cf26b9b74171df9" name = "batchtools" version = "0.0.1" dependencies = [ - "async-channel 2.0.0", + "async-channel 2.1.0", "err", "log 0.0.1", "taskrun", @@ -393,9 +387,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.4.7" +version = "4.4.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac495e00dcec98c83465d5ad66c5c4fabd652fd6686e7c6269b117e729a6f17b" +checksum = "2275f18819641850fa26c89acc84d465c1bf91ce57bc2748b28c420473352f64" dependencies = [ "clap_builder", "clap_derive", @@ -403,9 +397,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.4.7" +version = "4.4.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c77ed9a32a62e6ca27175d00d29d05ca32e396ea1eb5fb01d8256b669cec7663" +checksum = "07cdf1b148b25c1e1f7a42225e30a0d99a615cd4637eae7365548dd4529b95bc" dependencies = [ "anstream", "anstyle", @@ -422,7 +416,7 @@ dependencies = [ "heck 0.4.1", "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -691,7 +685,7 @@ dependencies = [ name = "daqingest" version = "0.2.0-alpha.1" dependencies = [ - "async-channel 2.0.0", + "async-channel 2.1.0", "bytes", "chrono", "clap", @@ -733,7 +727,7 @@ dependencies = [ "ident_case", "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -744,7 +738,7 @@ checksum = "836a9bbc7ad63342d6d6e7b815ccab164bc77a2d95d84bc3117a8c0d5c98e2d5" dependencies = [ "darling_core", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -764,7 +758,7 @@ dependencies = [ name = "dbpg" version = "0.0.1" dependencies = [ - "async-channel 2.0.0", + "async-channel 2.1.0", "batchtools", "err", "futures-util", @@ -858,7 +852,7 @@ dependencies = [ "darling", "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -882,7 +876,7 @@ version = "0.0.5" dependencies = [ "anyhow", "async-channel 1.9.0", - "async-channel 2.0.0", + "async-channel 2.1.0", "backtrace", "chrono", "http", @@ -898,9 +892,9 @@ dependencies = [ [[package]] name = "errno" -version = "0.3.5" +version = "0.3.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac3e13f66a2f95e32a39eaa81f6b95d42878ca0e1db0c7543723dfe12557e860" +checksum = "f258a7194e7f7c2a7837a8913aeab7fd8c383457034fa20ce4dd3dcb813e8eb8" dependencies = [ "libc", "windows-sys 0.48.0", @@ -914,9 +908,9 @@ checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" [[package]] name = "event-listener" -version = "3.0.1" +version = "3.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "01cec0252c2afff729ee6f00e903d479fba81784c8e2bd77447673471fdfaea1" +checksum = "d93877bcde0eb80ca09131a08d23f0a5c18a620b01db137dba666d18cd9b30c2" dependencies = [ "concurrent-queue", "parking", @@ -929,7 +923,7 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d96b852f1345da36d551b9473fa1e2b1eb5c5195585c6c018118bc92a8d91160" dependencies = [ - "event-listener 3.0.1", + "event-listener 3.1.0", "pin-project-lite", ] @@ -1053,7 +1047,7 @@ checksum = "53b153fd91e4b0147f4aced87be237c98248656bb01050b96bf3ee89220a8ddb" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -1107,9 +1101,9 @@ dependencies = [ [[package]] name = "getrandom" -version = "0.2.10" +version = "0.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "be4136b2a15dd319360be1c07d9933517ccf0be8f16bf62a3bee4f0d618df427" +checksum = "fe9006bed769170c11f845cf00c7c1e9092aeb3f268e007c3e760ac68008070f" dependencies = [ "cfg-if", "libc", @@ -1135,9 +1129,9 @@ checksum = "6fb8d784f27acf97159b40fc4db5ecd8aa23b9ad5ef69cdd136d3bc80665f0c0" [[package]] name = "h2" -version = "0.3.21" +version = "0.3.22" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91fc23aa11be92976ef4729127f1a74adf36d8436f7816b185d18df956790833" +checksum = "4d6250322ef6e60f93f9a2162799302cd6f68f79f6e5d85c8c16f14d1d958178" dependencies = [ "bytes", "fnv", @@ -1145,7 +1139,7 @@ dependencies = [ "futures-sink", "futures-util", "http", - "indexmap 1.9.3", + "indexmap 2.1.0", "slab", "tokio", "tokio-util", @@ -1184,11 +1178,11 @@ checksum = "f93e7192158dbcda357bdec5fb5788eebf8bbac027f3f33e719d29135ae84156" [[package]] name = "hdrhistogram" -version = "7.5.2" +version = "7.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f19b9f54f7c7f55e31401bb647626ce0cf0f67b0004982ce815b3ee72a02aa8" +checksum = "765c9198f173dd59ce26ff9f95ef0aafd0a0fe01fb9d72841bc5066a4c06511d" dependencies = [ - "base64 0.13.1", + "base64", "byteorder", "flate2", "nom", @@ -1253,9 +1247,9 @@ dependencies = [ [[package]] name = "http" -version = "0.2.9" +version = "0.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd6effc99afb63425aff9b05836f029929e345a6148a14b7ecd5ab67af944482" +checksum = "8947b1a6fad4393052c7ba1f4cd97bed3e953a95c79c92ad9b051a04611d9fbb" dependencies = [ "bytes", "fnv", @@ -1433,7 +1427,7 @@ dependencies = [ name = "ingest-bsread" version = "0.0.1" dependencies = [ - "async-channel 2.0.0", + "async-channel 2.1.0", "batchtools", "bitshuffle", "bytes", @@ -1513,7 +1507,7 @@ dependencies = [ name = "items_proc" version = "0.0.2" dependencies = [ - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -1542,9 +1536,9 @@ checksum = "af150ab688ff2122fcef229be89cb50dd66af9e01a4ff320cc137eecc9bacc38" [[package]] name = "js-sys" -version = "0.3.64" +version = "0.3.65" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c5f195fe497f702db0f318b07fdd68edb16955aed830df8363d837542f8f935a" +checksum = "54c0c35952f67de54bb584e9fd912b3023117cbafc0a77d8f3dee1fb5f572fe8" dependencies = [ "wasm-bindgen", ] @@ -1563,15 +1557,15 @@ checksum = "884e2677b40cc8c339eaefcb701c32ef1fd2493d71118dc0ca4b6a736c93bd67" [[package]] name = "libc" -version = "0.2.149" +version = "0.2.150" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a08173bc88b7955d1b3145aa561539096c421ac8debde8cbc3612ec635fee29b" +checksum = "89d92a4743f9a61002fae18374ed11e7973f530cb3a3255fb354818118b2203c" [[package]] name = "linux-raw-sys" -version = "0.4.10" +version = "0.4.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da2479e8c062e40bf0066ffa0bc823de0a9368974af99c9f6df941d2c231e03f" +checksum = "969488b55f8ac402214f3f5fd243ebb7206cf82de60d3172994707a4bcc2b829" [[package]] name = "lock_api" @@ -1742,7 +1736,7 @@ name = "netfetch" version = "0.0.3" dependencies = [ "arrayref", - "async-channel 2.0.0", + "async-channel 2.1.0", "axum", "batchtools", "bitshuffle", @@ -1878,7 +1872,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -1898,9 +1892,9 @@ checksum = "dd8b5dd2ae5ed71462c540258bedcb51965123ad7e7ccf4b9a8cafaa4a63576d" [[package]] name = "openssl" -version = "0.10.57" +version = "0.10.59" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bac25ee399abb46215765b1cb35bc0212377e58a061560d8b29b024fd0430e7c" +checksum = "7a257ad03cd8fb16ad4172fedf8094451e1af1c4b70097636ef2eac9a5f0cc33" dependencies = [ "bitflags 2.4.1", "cfg-if", @@ -1919,7 +1913,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -1930,9 +1924,9 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" [[package]] name = "openssl-sys" -version = "0.9.93" +version = "0.9.95" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "db4d56a4c0478783083cfafcc42493dd4a981d41669da64b4572a2a089b51b1d" +checksum = "40a4130519a360279579c2053038317e40eff64d13fd3f004f9e1b72b8a6aaf9" dependencies = [ "cc", "libc", @@ -2039,7 +2033,7 @@ checksum = "4359fd9c9171ec6e8c62926d6faaf553a8dc3f64e1507e76da7911b4f6a04405" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -2078,7 +2072,7 @@ version = "0.6.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "49b6c5ef183cd3ab4ba005f1ca64c21e8bd97ce4699cfea9e8d9a2c4958ca520" dependencies = [ - "base64 0.21.5", + "base64", "byteorder", "bytes", "fallible-iterator", @@ -2279,6 +2273,15 @@ dependencies = [ "rand_core", ] +[[package]] +name = "rand_xoshiro" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f97cdb2a36ed4183de61b2f824cc45c9f1037f28afe0a322e9fff4c108b5aaa" +dependencies = [ + "rand_core", +] + [[package]] name = "rayon" version = "1.8.0" @@ -2453,9 +2456,9 @@ dependencies = [ [[package]] name = "rustix" -version = "0.38.21" +version = "0.38.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b426b0506e5d50a7d8dafcf2e81471400deb602392c7dd110815afb4eaf02a3" +checksum = "dc99bc2d4f1fed22595588a013687477aedf3cdcfb26558c559edb67b4d9b22e" dependencies = [ "bitflags 2.4.1", "errno", @@ -2554,14 +2557,14 @@ checksum = "5757ded3dfb10967ca7d1ff1084d072d565b5e10b2b21c286d5335c245425a7e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] name = "scywr" version = "0.0.1" dependencies = [ - "async-channel 2.0.0", + "async-channel 2.1.0", "err", "futures-util", "log 0.0.1", @@ -2606,9 +2609,9 @@ dependencies = [ [[package]] name = "self_cell" -version = "1.0.1" +version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c309e515543e67811222dbc9e3dd7e1056279b782e1dacffe4242b718734fb6" +checksum = "e388332cd64eb80cd595a00941baf513caffae8dce9cfd0467fc9c66397dade6" [[package]] name = "semver" @@ -2618,9 +2621,9 @@ checksum = "836fa6a3e1e547f9a2c4040802ec865b5d85f4014efe00555d7090a3dcaa1090" [[package]] name = "serde" -version = "1.0.190" +version = "1.0.192" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91d3c334ca1ee894a2c6f6ad698fe8c435b76d504b13d436f0685d648d6d96f7" +checksum = "bca2a08484b285dcb282d0f67b26cadc0df8b19f8c12502c13d966bf9482f001" dependencies = [ "serde_derive", ] @@ -2648,13 +2651,13 @@ dependencies = [ [[package]] name = "serde_derive" -version = "1.0.190" +version = "1.0.192" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67c5609f394e5c2bd7fc51efda478004ea80ef42fee983d5c67a65e34f32c0e3" +checksum = "d6c7207fbec9faa48073f3e3074cbe553af6ea512d7c21ba46e434e70ea9fbc1" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -2788,9 +2791,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.11.1" +version = "1.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "942b4a808e05215192e39f4ab80813e599068285906cc91aa64f923db842bd5a" +checksum = "4dccd0940a2dcdf68d092b8cbab7dc0ad8fa938bf95787e1b916b0e3d0e8e970" [[package]] name = "snap" @@ -2850,6 +2853,7 @@ name = "stats" version = "0.0.1-a.0" dependencies = [ "log 0.0.1", + "rand_xoshiro", "stats_proc", "stats_types", ] @@ -2860,7 +2864,7 @@ version = "0.0.1" dependencies = [ "quote", "stats_types", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -2949,9 +2953,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.38" +version = "2.0.39" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e96b79aaa137db8f61e26363a0c9b47d8b4ec75da28b7d1d614c2303e232408b" +checksum = "23e78b90f2fcf45d3e842032ce32e3f2d1545ba6636271dcbf24fa306d87be7a" dependencies = [ "proc-macro2", "quote", @@ -3029,7 +3033,7 @@ source = "git+https://github.com/dominikwerder/thiserror.git#052df05c18b5f26b462 dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -3040,7 +3044,7 @@ checksum = "266b2e40bc00e5a6c09c3584011e08b06f123c00362c92b975ba9843aaaa14b8" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -3099,9 +3103,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.33.0" +version = "1.34.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f38200e3ef7995e5ef13baec2f432a6da0aa9ac495b2c0e8f3b7eec2c92d653" +checksum = "d0c014766411e834f7af5b8f4cf46257aab4036ca95e9d2c144a10f59ad6f5b9" dependencies = [ "backtrace", "bytes", @@ -3129,13 +3133,13 @@ dependencies = [ [[package]] name = "tokio-macros" -version = "2.1.0" +version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e" +checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -3224,7 +3228,7 @@ checksum = "3082666a3a6433f7f511c7192923fa1fe07c69332d3c6a2e6bb040b569199d5a" dependencies = [ "async-trait", "axum", - "base64 0.21.5", + "base64", "bytes", "futures-core", "futures-util", @@ -3296,7 +3300,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", ] [[package]] @@ -3311,9 +3315,9 @@ dependencies = [ [[package]] name = "tracing-log" -version = "0.1.4" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f751112709b4e791d8ce53e32c4ed2d353565a795ce84da2285393f41557bdf2" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" dependencies = [ "log 0.4.20", "once_cell", @@ -3322,9 +3326,9 @@ dependencies = [ [[package]] name = "tracing-subscriber" -version = "0.3.17" +version = "0.3.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30a651bc37f915e81f087d86e62a18eec5f79550c7faff886f7090b4ea757c77" +checksum = "ad0f048c97dbd9faa9b7df56362b8ebcaa52adb06b498c050d2f4e32f90a7a8b" dependencies = [ "matchers", "nu-ansi-term", @@ -3413,9 +3417,9 @@ checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a" [[package]] name = "uuid" -version = "1.5.0" +version = "1.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88ad59a7560b41a70d191093a945f0b87bc1deeda46fb237479708a1d6b6cdfc" +checksum = "5e395fcf16a7a3d8127ec99782007af141946b4795001f876d54fb0d55978560" dependencies = [ "getrandom", ] @@ -3455,9 +3459,9 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" [[package]] name = "wasm-bindgen" -version = "0.2.87" +version = "0.2.88" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7706a72ab36d8cb1f80ffbf0e071533974a60d0a308d01a5d0375bf60499a342" +checksum = "7daec296f25a1bae309c0cd5c29c4b260e510e6d813c286b19eaadf409d40fce" dependencies = [ "cfg-if", "wasm-bindgen-macro", @@ -3465,16 +3469,16 @@ dependencies = [ [[package]] name = "wasm-bindgen-backend" -version = "0.2.87" +version = "0.2.88" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ef2b6d3c510e9625e5fe6f509ab07d66a760f0885d858736483c32ed7809abd" +checksum = "e397f4664c0e4e428e8313a469aaa58310d302159845980fd23b0f22a847f217" dependencies = [ "bumpalo", "log 0.4.20", "once_cell", "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", "wasm-bindgen-shared", ] @@ -3503,9 +3507,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.87" +version = "0.2.88" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dee495e55982a3bd48105a7b947fd2a9b4a8ae3010041b9e0faab3f9cd028f1d" +checksum = "5961017b3b08ad5f3fe39f1e79877f8ee7c23c5e5fd5eb80de95abc41f1f16b2" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -3513,22 +3517,22 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.87" +version = "0.2.88" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "54681b18a46765f095758388f2d0cf16eb8d4169b639ab575a8f5693af210c7b" +checksum = "c5353b8dab669f5e10f5bd76df26a9360c748f054f862ff5f3f8aae0c7fb3907" dependencies = [ "proc-macro2", "quote", - "syn 2.0.38", + "syn 2.0.39", "wasm-bindgen-backend", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-shared" -version = "0.2.87" +version = "0.2.88" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca6ad05a4870b2bf5fe995117d3728437bd27d7cd5f06f13c17443ef369775a1" +checksum = "0d046c5d029ba91a1ed14da14dca44b68bf2f124cfbaf741c54151fdb3e0750b" [[package]] name = "wasmer" @@ -3672,9 +3676,9 @@ dependencies = [ [[package]] name = "web-sys" -version = "0.3.64" +version = "0.3.65" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b85cbef8c220a6abc02aefd892dfc0fc23afb1c6a426316ec33253a3877249b" +checksum = "5db499c5f66323272151db0e666cd34f78617522fb0c1604d31a27c50c206a85" dependencies = [ "js-sys", "wasm-bindgen", @@ -3832,9 +3836,9 @@ checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" [[package]] name = "winnow" -version = "0.5.18" +version = "0.5.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "176b6138793677221d420fd2f0aeeced263f197688b36484660da767bca2fa32" +checksum = "829846f3e3db426d4cee4510841b71a8e58aa2a76b1132579487ae430ccd9c7b" dependencies = [ "memchr", ] diff --git a/daqingest/src/bin/daqingest.rs b/daqingest/src/bin/daqingest.rs index 3df9096..fa9fcd9 100644 --- a/daqingest/src/bin/daqingest.rs +++ b/daqingest/src/bin/daqingest.rs @@ -7,7 +7,7 @@ use netfetch::conf::parse_config; pub fn main() -> Result<(), Error> { let opts = DaqIngestOpts::parse(); // TODO offer again function to get runtime and configure tracing in one call - let runtime = taskrun::get_runtime_opts(opts.nworkers.unwrap_or(12), 32); + let runtime = taskrun::get_runtime_opts(opts.worker_threads.unwrap_or(8), opts.blocking_threads.unwrap_or(256)); match taskrun::tracing_init() { Ok(()) => {} Err(()) => return Err(Error::with_msg_no_trace("tracing init failed")), diff --git a/daqingest/src/daemon.rs b/daqingest/src/daemon.rs index 21efeb7..4e0df3e 100644 --- a/daqingest/src/daemon.rs +++ b/daqingest/src/daemon.rs @@ -5,6 +5,7 @@ use async_channel::Sender; use async_channel::WeakSender; use err::Error; use log::*; +use netfetch::ca::conn; use netfetch::ca::connset::CaConnSet; use netfetch::ca::connset::CaConnSetCtrl; use netfetch::ca::connset::CaConnSetItem; @@ -33,13 +34,16 @@ use std::time::SystemTime; use taskrun::tokio; use tokio::task::JoinHandle; +const CHECK_HEALTH_IVL: Duration = Duration::from_millis(2000); +const CHECK_HEALTH_TIMEOUT: Duration = Duration::from_millis(1500); const PRINT_ACTIVE_INTERVAL: Duration = Duration::from_millis(60000); const PRINT_STATUS_INTERVAL: Duration = Duration::from_millis(20000); +const CHECK_CHANNEL_SLOW_WARN: Duration = Duration::from_millis(500); #[derive(Debug)] enum CheckPeriodic { Waiting(Instant), - Ongoing(Instant), + Ongoing(u32, Instant), } pub struct DaemonOpts { @@ -130,7 +134,7 @@ impl Daemon { } }); - // #[cfg(DISABLED)] + #[cfg(DISABLED)] let query_item_rx = { // TODO only testing, remove tokio::spawn({ @@ -241,16 +245,16 @@ impl Daemon { async fn check_caconn_chans(&mut self, ts1: Instant) -> Result<(), Error> { match &self.connset_status_last { CheckPeriodic::Waiting(since) => { - if *since + Duration::from_millis(2000) < ts1 { - self.connset_ctrl.check_health().await?; - self.connset_status_last = CheckPeriodic::Ongoing(ts1); + if *since + CHECK_HEALTH_IVL <= ts1 { + let id = self.connset_ctrl.check_health().await?; + self.connset_status_last = CheckPeriodic::Ongoing(id, ts1); } } - CheckPeriodic::Ongoing(since) => { + CheckPeriodic::Ongoing(idexp, since) => { let dt = ts1.saturating_duration_since(*since); - if dt > Duration::from_millis(2000) { + if dt > CHECK_HEALTH_TIMEOUT { error!( - "CaConnSet has not reported health status since {:.0}", + "CaConnSet has not reported health status since {:.0} idexp {idexp:08x}", dt.as_secs_f32() * 1e3 ); } @@ -285,8 +289,8 @@ impl Daemon { let ts1 = Instant::now(); self.check_caconn_chans(ts1).await?; let dt = ts1.elapsed(); - if dt > Duration::from_millis(500) { - info!("slow check_chans {}ms", dt.as_secs_f32() * 1e3); + if dt > CHECK_CHANNEL_SLOW_WARN { + info!("slow check_chans {:.0} ms", dt.as_secs_f32() * 1e3); } if false && tsnow.duration_since(self.last_status_print).unwrap_or(Duration::ZERO) >= PRINT_STATUS_INTERVAL { self.last_status_print = tsnow; @@ -304,14 +308,10 @@ impl Daemon { Ok(()) } - async fn handle_channel_add(&mut self, ch: Channel) -> Result<(), Error> { + async fn handle_channel_add(&mut self, ch: Channel, restx: netfetch::ca::conn::CmdResTx) -> Result<(), Error> { // debug!("handle_channel_add {ch:?}"); self.connset_ctrl - .add_channel( - self.ingest_opts.backend().into(), - ch.id().into(), - self.ingest_opts.local_epics_hostname(), - ) + .add_channel(self.ingest_opts.backend().into(), ch.id().into(), restx) .await?; Ok(()) } @@ -375,7 +375,7 @@ impl Daemon { async fn handle_ca_conn_set_item(&mut self, item: CaConnSetItem) -> Result<(), Error> { use CaConnSetItem::*; match item { - Healthy(ts1, ts2) => { + Healthy(id, ts1, ts2) => { let tsnow = Instant::now(); let dt1 = tsnow.duration_since(ts1).as_secs_f32() * 1e3; let dt2 = tsnow.duration_since(ts2).as_secs_f32() * 1e3; @@ -383,7 +383,10 @@ impl Daemon { CheckPeriodic::Waiting(_since) => { error!("received CaConnSet health report without having asked {dt1:.0} ms {dt2:.0} ms"); } - CheckPeriodic::Ongoing(since) => { + CheckPeriodic::Ongoing(idexp, since) => { + if id != *idexp { + warn!("unexpected check health answer id {id:08x} idexp {idexp:08x}"); + } // TODO insert response time as series to scylla. let dtsince = tsnow.duration_since(*since).as_secs_f32() * 1e3; { @@ -391,7 +394,7 @@ impl Daemon { *v += (dtsince - *v) * 0.2; self.stats.connset_health_lat_ema().set(*v as _); } - debug!("received CaConnSet Healthy dtsince {dtsince:.0} ms {dt1:.0} ms {dt2:.0} ms"); + trace!("received CaConnSet Healthy dtsince {dtsince:.0} ms {dt1:.0} ms {dt2:.0} ms"); self.connset_status_last = CheckPeriodic::Waiting(tsnow); self.stats.caconnset_health_response().inc(); } @@ -457,7 +460,7 @@ impl Daemon { let _ = ts1.elapsed(); ret } - ChannelAdd(ch) => self.handle_channel_add(ch).await, + ChannelAdd(ch, tx) => self.handle_channel_add(ch, tx).await, ChannelRemove(ch) => self.handle_channel_remove(ch).await, CaConnSetItem(item) => self.handle_ca_conn_set_item(item).await, Shutdown => self.handle_shutdown().await, @@ -563,7 +566,7 @@ fn handler_sigterm(_a: libc::c_int, _b: *const libc::siginfo_t, _c: *const libc: let _ = ingest_linux::signal::unset_signal_handler(libc::SIGTERM); } -pub async fn run(opts: CaIngestOpts, channels: Vec) -> Result<(), Error> { +pub async fn run(opts: CaIngestOpts, channels: Option>) -> Result<(), Error> { info!("start up {opts:?}"); ingest_linux::signal::set_signal_handler(libc::SIGINT, handler_sigint).map_err(Error::from_string)?; ingest_linux::signal::set_signal_handler(libc::SIGTERM, handler_sigterm).map_err(Error::from_string)?; @@ -587,7 +590,7 @@ pub async fn run(opts: CaIngestOpts, channels: Vec) -> Result<(), Error> let mut channels = channels; if opts.test_bsread_addr.is_some() { - channels.clear(); + channels = None; } let insert_frac = Arc::new(AtomicU64::new(opts.insert_frac())); @@ -634,19 +637,24 @@ pub async fn run(opts: CaIngestOpts, channels: Vec) -> Result<(), Error> let daemon_jh = taskrun::spawn(daemon.daemon()); - debug!("will configure {} channels", channels.len()); - let mut thr_msg = ThrottleTrace::new(Duration::from_millis(1000)); - let mut i = 0; - for s in &channels { - let ch = Channel::new(s.into()); - match tx.send(DaemonEvent::ChannelAdd(ch)).await { - Ok(()) => {} - Err(_) => break, + if let Some(channels) = channels { + debug!("will configure {} channels", channels.len()); + let mut thr_msg = ThrottleTrace::new(Duration::from_millis(1000)); + let mut i = 0; + for s in &channels { + let ch = Channel::new(s.into()); + match tx.send(DaemonEvent::ChannelAdd(ch, async_channel::bounded(1).0)).await { + Ok(()) => {} + Err(e) => { + error!("{e}"); + break; + } + } + thr_msg.trigger("daemon sent ChannelAdd", &[&i as &_]); + i += 1; } - thr_msg.trigger("daemon sent ChannelAdd", &[&i as &_]); - i += 1; + debug!("{} configured channels applied", channels.len()); } - debug!("{} configured channels applied", channels.len()); daemon_jh.await.map_err(|e| Error::with_msg_no_trace(e.to_string()))??; info!("Daemon joined."); metrics_shutdown_tx.send(1).await?; diff --git a/daqingest/src/opts.rs b/daqingest/src/opts.rs index 5497ad4..11ef057 100644 --- a/daqingest/src/opts.rs +++ b/daqingest/src/opts.rs @@ -12,7 +12,9 @@ pub struct DaqIngestOpts { #[command(subcommand)] pub subcmd: SubCmd, #[arg(long)] - pub nworkers: Option, + pub worker_threads: Option, + #[arg(long)] + pub blocking_threads: Option, } #[derive(Debug, clap::Parser)] diff --git a/netfetch/Cargo.toml b/netfetch/Cargo.toml index 804b7aa..707293d 100644 --- a/netfetch/Cargo.toml +++ b/netfetch/Cargo.toml @@ -23,7 +23,6 @@ md-5 = "0.10.5" hex = "0.4.3" regex = "1.8.4" axum = "0.6.18" -http = "0.2" url = "2.2" hyper = "0.14" chrono = "0.4" diff --git a/netfetch/src/ca/conn.rs b/netfetch/src/ca/conn.rs index 0cb4032..dd9d92b 100644 --- a/netfetch/src/ca/conn.rs +++ b/netfetch/src/ca/conn.rs @@ -43,17 +43,15 @@ use series::ChannelStatusSeriesId; use series::SeriesId; use stats::rand_xoshiro::rand_core::RngCore; use stats::rand_xoshiro::rand_core::SeedableRng; -use stats::rand_xoshiro::Xoshiro128StarStar; +use stats::rand_xoshiro::Xoshiro128PlusPlus; use stats::CaConnStats; use stats::CaProtoStats; use stats::IntervalEma; -use stats::XorShift32; use std::collections::BTreeMap; use std::collections::HashMap; use std::collections::VecDeque; use std::net::SocketAddrV4; use std::ops::ControlFlow; -use std::pin::pin; use std::pin::Pin; use std::sync::atomic; use std::sync::atomic::AtomicUsize; @@ -322,13 +320,13 @@ fn wait_fut(dt: u64) -> Pin + Send>> { } struct CidStore { - rng: XorShift32, + rng: Xoshiro128PlusPlus, } impl CidStore { fn new(seed: u32) -> Self { Self { - rng: XorShift32::new(seed), + rng: Xoshiro128PlusPlus::seed_from_u64(seed as _), } } @@ -342,18 +340,18 @@ impl CidStore { } fn next(&mut self) -> Cid { - Cid(self.rng.next()) + Cid(self.rng.next_u32()) } } struct SubidStore { - rng: XorShift32, + rng: Xoshiro128PlusPlus, } impl SubidStore { fn new(seed: u32) -> Self { Self { - rng: XorShift32::new(seed), + rng: Xoshiro128PlusPlus::seed_from_u64(seed as _), } } @@ -367,7 +365,7 @@ impl SubidStore { } fn next(&mut self) -> Subid { - Subid(self.rng.next()) + Subid(self.rng.next_u32()) } } @@ -376,6 +374,8 @@ fn info_store_msp_from_time(ts: SystemTime) -> u32 { (dt.as_secs() / 60 * 60) as u32 } +pub type CmdResTx = Sender>; + #[derive(Debug)] pub enum ConnCommandKind { SeriesLookupResult(Result), @@ -552,7 +552,7 @@ pub struct CaConn { thr_msg_poll: ThrottleTrace, ca_proto_stats: Arc, weird_count: usize, - rng: Xoshiro128StarStar, + rng: Xoshiro128PlusPlus, } #[cfg(DISABLED)] @@ -614,7 +614,7 @@ impl CaConn { } } - fn ioc_ping_ivl_rng(rng: &mut Xoshiro128StarStar) -> Duration { + fn ioc_ping_ivl_rng(rng: &mut Xoshiro128PlusPlus) -> Duration { IOC_PING_IVL * 100 / (70 + (rng.next_u32() % 60)) } @@ -668,7 +668,7 @@ impl CaConn { } fn cmd_check_health(&mut self) { - debug!("cmd_check_health"); + trace!("cmd_check_health"); match self.check_channels_alive() { Ok(_) => {} Err(e) => { @@ -743,8 +743,6 @@ impl CaConn { fn cmd_channel_add(&mut self, name: String, cssid: ChannelStatusSeriesId) { self.channel_add(name, cssid); - // TODO return the result - //self.stats.caconn_command_can_not_reply.inc(); } fn cmd_channel_remove(&mut self, name: String) { @@ -2168,9 +2166,9 @@ impl Stream for CaConn { Pending } else { // TODO error - error!("logic error"); + error!("shutting down, queues not flushed, no progress, no pending"); self.stats.logic_error().inc(); - let e = Error::with_msg_no_trace("shutdown, not done, no progress, no pending"); + let e = Error::with_msg_no_trace("shutting down, queues not flushed, no progress, no pending"); Ready(Some(Err(e))) } } diff --git a/netfetch/src/ca/connset.rs b/netfetch/src/ca/connset.rs index 2b872af..0c05068 100644 --- a/netfetch/src/ca/connset.rs +++ b/netfetch/src/ca/connset.rs @@ -46,6 +46,8 @@ use statemap::ChannelStateValue; use statemap::WithStatusSeriesIdState; use statemap::WithStatusSeriesIdStateInner; use statemap::CHANNEL_STATUS_DUMMY_SCALAR_TYPE; +use stats::rand_xoshiro::rand_core::RngCore; +use stats::rand_xoshiro::Xoshiro128PlusPlus; use stats::CaConnSetStats; use stats::CaConnStats; use stats::CaProtoStats; @@ -134,7 +136,6 @@ impl CaConnRes { pub struct ChannelAddWithAddr { backend: String, name: String, - local_epics_hostname: String, cssid: ChannelStatusSeriesId, addr: SocketAddr, } @@ -143,7 +144,6 @@ pub struct ChannelAddWithAddr { pub struct ChannelAddWithStatusId { backend: String, name: String, - local_epics_hostname: String, cssid: ChannelStatusSeriesId, } @@ -151,7 +151,7 @@ pub struct ChannelAddWithStatusId { pub struct ChannelAdd { backend: String, name: String, - local_epics_hostname: String, + restx: crate::ca::conn::CmdResTx, } #[derive(Debug, Clone)] @@ -196,7 +196,7 @@ impl fmt::Debug for ChannelStatusesRequest { pub enum ConnSetCmd { ChannelAdd(ChannelAdd), ChannelRemove(ChannelRemove), - CheckHealth(Instant), + CheckHealth(u32, Instant), Shutdown, ChannelStatuses(ChannelStatusesRequest), } @@ -213,7 +213,7 @@ impl CaConnSetEvent { #[derive(Debug, Clone)] pub enum CaConnSetItem { Error(Error), - Healthy(Instant, Instant), + Healthy(u32, Instant, Instant), } pub struct CaConnSetCtrl { @@ -224,9 +224,15 @@ pub struct CaConnSetCtrl { ca_proto_stats: Arc, ioc_finder_stats: Arc, jh: JoinHandle>, + rng: Xoshiro128PlusPlus, + idcnt: u32, } impl CaConnSetCtrl { + pub fn new() -> Self { + todo!() + } + pub fn sender(&self) -> Sender { self.tx.clone() } @@ -235,12 +241,13 @@ impl CaConnSetCtrl { self.rx.clone() } - pub async fn add_channel(&self, backend: String, name: String, local_epics_hostname: String) -> Result<(), Error> { - let cmd = ChannelAdd { - backend, - name, - local_epics_hostname, - }; + pub async fn add_channel( + &self, + backend: String, + name: String, + restx: crate::ca::conn::CmdResTx, + ) -> Result<(), Error> { + let cmd = ChannelAdd { backend, name, restx }; let cmd = ConnSetCmd::ChannelAdd(cmd); self.tx.send(CaConnSetEvent::ConnSetCmd(cmd)).await?; Ok(()) @@ -259,16 +266,17 @@ impl CaConnSetCtrl { Ok(()) } - pub async fn check_health(&self) -> Result<(), Error> { - let cmd = ConnSetCmd::CheckHealth(Instant::now()); + pub async fn check_health(&mut self) -> Result { + let id = self.make_id(); + let cmd = ConnSetCmd::CheckHealth(id, Instant::now()); let n = self.tx.len(); if n > 0 { debug!("check_health self.tx.len() {:?}", n); } let s = format!("{:?}", cmd); self.tx.send(CaConnSetEvent::ConnSetCmd(cmd)).await?; - debug!("check_health enqueued {s}"); - Ok(()) + trace!("check_health enqueued {s}"); + Ok(id) } pub async fn join(self) -> Result<(), Error> { @@ -291,6 +299,12 @@ impl CaConnSetCtrl { pub fn ioc_finder_stats(&self) -> &Arc { &self.ioc_finder_stats } + + fn make_id(&mut self) -> u32 { + let id = self.idcnt; + self.idcnt += 1; + self.rng.next_u32() & 0xffff | (id << 16) + } } #[derive(Debug)] @@ -441,6 +455,8 @@ impl CaConnSet { ca_proto_stats, ioc_finder_stats, jh, + idcnt: 0, + rng: stats::xoshiro_from_time(), } } @@ -482,51 +498,13 @@ impl CaConnSet { ConnSetCmd::ChannelRemove(x) => self.handle_remove_channel(x), // ConnSetCmd::IocAddrQueryResult(x) => self.handle_ioc_query_result(x).await, // ConnSetCmd::SeriesLookupResult(x) => self.handle_series_lookup_result(x).await, - ConnSetCmd::CheckHealth(ts1) => self.handle_check_health(ts1), + ConnSetCmd::CheckHealth(id, ts1) => self.handle_check_health(id, ts1), ConnSetCmd::Shutdown => self.handle_shutdown(), ConnSetCmd::ChannelStatuses(x) => self.handle_channel_statuses_req(x), }, } } - fn handle_ca_conn_event(&mut self, addr: SocketAddr, ev: CaConnEvent) -> Result<(), Error> { - match ev.value { - CaConnEventValue::None => Ok(()), - CaConnEventValue::EchoTimeout => Ok(()), - CaConnEventValue::ConnCommandResult(x) => self.handle_conn_command_result(addr, x), - CaConnEventValue::QueryItem(item) => { - todo!("remove this insert case"); - // self.storage_insert_queue.push_back(item); - Ok(()) - } - CaConnEventValue::ChannelCreateFail(x) => self.handle_channel_create_fail(addr, x), - CaConnEventValue::EndOfStream => self.handle_ca_conn_eos(addr), - CaConnEventValue::ConnectFail => self.handle_connect_fail(addr), - } - } - - fn handle_series_lookup_result(&mut self, res: Result) -> Result<(), Error> { - if self.shutdown_stopping { - return Ok(()); - } - trace3!("handle_series_lookup_result {res:?}"); - match res { - Ok(res) => { - let add = ChannelAddWithStatusId { - backend: res.backend, - name: res.channel, - local_epics_hostname: self.local_epics_hostname.clone(), - cssid: ChannelStatusSeriesId::new(res.series.into_inner().id()), - }; - self.handle_add_channel_with_status_id(add)?; - } - Err(e) => { - warn!("TODO handle error {e}"); - } - } - Ok(()) - } - fn handle_add_channel(&mut self, cmd: ChannelAdd) -> Result<(), Error> { if self.shutdown_stopping { trace3!("handle_add_channel but shutdown_stopping"); @@ -555,6 +533,46 @@ impl CaConnSet { tx: Box::pin(SeriesLookupSender { tx }), }; self.channel_info_query_queue.push_back(item); + if let Err(_) = cmd.restx.try_send(Ok(())) { + self.stats.command_reply_fail().inc(); + } + Ok(()) + } + + fn handle_ca_conn_event(&mut self, addr: SocketAddr, ev: CaConnEvent) -> Result<(), Error> { + match ev.value { + CaConnEventValue::None => Ok(()), + CaConnEventValue::EchoTimeout => Ok(()), + CaConnEventValue::ConnCommandResult(x) => self.handle_conn_command_result(addr, x), + CaConnEventValue::QueryItem(item) => { + todo!("remove this insert case"); + // self.storage_insert_queue.push_back(item); + Ok(()) + } + CaConnEventValue::ChannelCreateFail(x) => self.handle_channel_create_fail(addr, x), + CaConnEventValue::EndOfStream => self.handle_ca_conn_eos(addr), + CaConnEventValue::ConnectFail => self.handle_connect_fail(addr), + } + } + + fn handle_series_lookup_result(&mut self, res: Result) -> Result<(), Error> { + if self.shutdown_stopping { + return Ok(()); + } + trace3!("handle_series_lookup_result {res:?}"); + match res { + Ok(res) => { + let add = ChannelAddWithStatusId { + backend: res.backend, + name: res.channel, + cssid: ChannelStatusSeriesId::new(res.series.into_inner().id()), + }; + self.handle_add_channel_with_status_id(add)?; + } + Err(e) => { + warn!("TODO handle error {e}"); + } + } Ok(()) } @@ -721,7 +739,6 @@ impl CaConnSet { name: res.channel, addr: SocketAddr::V4(addr), cssid: status_series_id.clone(), - local_epics_hostname: self.local_epics_hostname.clone(), }; self.handle_add_channel_with_addr(cmd)?; } @@ -747,8 +764,8 @@ impl CaConnSet { Ok(()) } - fn handle_check_health(&mut self, ts1: Instant) -> Result<(), Error> { - trace2!("handle_check_health"); + fn handle_check_health(&mut self, id: u32, ts1: Instant) -> Result<(), Error> { + trace2!("handle_check_health {id:08x}"); if self.shutdown_stopping { return Ok(()); } @@ -771,7 +788,7 @@ impl CaConnSet { } let ts2 = Instant::now(); - let item = CaConnSetItem::Healthy(ts1, ts2); + let item = CaConnSetItem::Healthy(id, ts1, ts2); self.connset_out_queue.push_back(item); Ok(()) } @@ -968,7 +985,7 @@ impl CaConnSet { opts, add.backend.clone(), addr_v4, - add.local_epics_hostname, + self.local_epics_hostname.clone(), self.storage_insert_tx.as_ref().get_ref().clone(), self.channel_info_query_tx .clone() @@ -1281,7 +1298,6 @@ impl CaConnSet { let cmd = ChannelAddWithAddr { backend: self.backend.clone(), name: ch.id().into(), - local_epics_hostname: self.local_epics_hostname.clone(), cssid: status_series_id.clone(), addr: SocketAddr::V4(*addr_v4), }; diff --git a/netfetch/src/conf.rs b/netfetch/src/conf.rs index 2fdeb4e..aa20044 100644 --- a/netfetch/src/conf.rs +++ b/netfetch/src/conf.rs @@ -14,7 +14,7 @@ use tokio::io::AsyncReadExt; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct CaIngestOpts { backend: String, - channels: PathBuf, + channels: Option, api_bind: String, search: Vec, #[serde(default)] @@ -30,7 +30,6 @@ pub struct CaIngestOpts { insert_worker_count: Option, insert_worker_concurrency: Option, insert_scylla_sessions: Option, - insert_queue_max: Option, insert_item_queue_cap: Option, local_epics_hostname: Option, store_workers_rate: Option, @@ -88,10 +87,6 @@ impl CaIngestOpts { self.insert_scylla_sessions.unwrap_or(1) } - pub fn insert_queue_max(&self) -> usize { - self.insert_queue_max.unwrap_or(64) - } - pub fn array_truncate(&self) -> u64 { self.array_truncate.unwrap_or(512) } @@ -163,7 +158,7 @@ scylla: "###; let res: Result = serde_yaml::from_slice(conf.as_bytes()); let conf = res.unwrap(); - assert_eq!(conf.channels, PathBuf::from("/some/path/file.txt")); + assert_eq!(conf.channels, Some(PathBuf::from("/some/path/file.txt"))); assert_eq!(&conf.api_bind, "0.0.0.0:3011"); assert_eq!(conf.search.get(0), Some(&"172.26.0.255".to_string())); assert_eq!(conf.scylla.hosts.get(1), Some(&"sf-nube-12:19042".to_string())); @@ -192,7 +187,7 @@ fn test_duration_parse() { assert_eq!(a.dur, Duration::from_millis(3170)); } -pub async fn parse_config(config: PathBuf) -> Result<(CaIngestOpts, Vec), Error> { +pub async fn parse_config(config: PathBuf) -> Result<(CaIngestOpts, Option>), Error> { let mut file = OpenOptions::new().read(true).open(config).await?; let mut buf = Vec::new(); file.read_to_end(&mut buf).await?; @@ -200,27 +195,32 @@ pub async fn parse_config(config: PathBuf) -> Result<(CaIngestOpts, Vec) drop(file); let re_p = regex::Regex::new(&conf.whitelist.clone().unwrap_or("--nothing-whitelisted--".into()))?; let re_n = regex::Regex::new(&conf.blacklist.clone().unwrap_or("--nothing-blacklisted--".into()))?; - let mut file = OpenOptions::new().read(true).open(&conf.channels).await?; - let mut buf = Vec::new(); - file.read_to_end(&mut buf).await?; - let lines = buf.split(|&x| x == 0x0a); - let mut channels = Vec::new(); - for line in lines { - let line = String::from_utf8_lossy(line); - let line = line.trim(); - let use_line = if line.is_empty() { - false - } else if let Some(_cs) = re_p.captures(&line) { - true - } else if re_n.is_match(&line) { - false - } else { - true - }; - if use_line { - channels.push(line.into()); + let channels = if let Some(fname) = conf.channels.as_ref() { + let mut file = OpenOptions::new().read(true).open(fname).await?; + let mut buf = Vec::new(); + file.read_to_end(&mut buf).await?; + let lines = buf.split(|&x| x == 0x0a); + let mut channels = Vec::new(); + for line in lines { + let line = String::from_utf8_lossy(line); + let line = line.trim(); + let use_line = if line.is_empty() { + false + } else if let Some(_cs) = re_p.captures(&line) { + true + } else if re_n.is_match(&line) { + false + } else { + true + }; + if use_line { + channels.push(line.into()); + } } - } - info!("Parsed {} channels", channels.len()); + info!("Parsed {} channels", channels.len()); + Some(channels) + } else { + None + }; Ok((conf, channels)) } diff --git a/netfetch/src/daemon_common.rs b/netfetch/src/daemon_common.rs index ccd0055..3fd2b47 100644 --- a/netfetch/src/daemon_common.rs +++ b/netfetch/src/daemon_common.rs @@ -20,7 +20,7 @@ impl Channel { #[derive(Debug, Clone)] pub enum DaemonEvent { TimerTick(u32, Sender), - ChannelAdd(Channel), + ChannelAdd(Channel, crate::ca::conn::CmdResTx), ChannelRemove(Channel), CaConnSetItem(CaConnSetItem), Shutdown, @@ -31,7 +31,7 @@ impl DaemonEvent { use DaemonEvent::*; match self { TimerTick(_, _) => format!("TimerTick"), - ChannelAdd(x) => format!("ChannelAdd {x:?}"), + ChannelAdd(x, _) => format!("ChannelAdd {x:?}"), ChannelRemove(x) => format!("ChannelRemove {x:?}"), CaConnSetItem(_) => format!("CaConnSetItem"), Shutdown => format!("Shutdown"), diff --git a/netfetch/src/metrics.rs b/netfetch/src/metrics.rs index 3d900f3..b8e71a4 100644 --- a/netfetch/src/metrics.rs +++ b/netfetch/src/metrics.rs @@ -9,8 +9,12 @@ use async_channel::Receiver; use async_channel::Sender; use async_channel::WeakSender; use axum::extract::Query; +use axum::http; +use axum::response::IntoResponse; +use axum::response::Response; use err::Error; use http::Request; +use http::StatusCode; use log::*; use scywr::iteminsertqueue::QueryItem; use serde::Deserialize; @@ -33,6 +37,41 @@ use std::sync::Arc; use std::time::Duration; use taskrun::tokio; +struct PublicErrorMsg(String); + +trait ToPublicErrorMsg { + fn to_public_err_msg(&self) -> PublicErrorMsg; +} + +impl ToPublicErrorMsg for err::Error { + fn to_public_err_msg(&self) -> PublicErrorMsg { + todo!() + } +} + +impl IntoResponse for PublicErrorMsg { + fn into_response(self) -> axum::response::Response { + todo!() + } +} + +struct CustomErrorResponse(Response); + +impl From for CustomErrorResponse +where + T: ToPublicErrorMsg, +{ + fn from(value: T) -> Self { + todo!() + } +} + +impl IntoResponse for CustomErrorResponse { + fn into_response(self) -> Response { + todo!() + } +} + pub struct StatsSet { daemon: Arc, ca_conn_set: Arc, @@ -79,6 +118,12 @@ impl ExtraInsertsConf { } } +async fn always_error(params: HashMap) -> Result, Response> { + Err(Error::with_public_msg_no_trace("The-public-message") + .to_public_err_msg() + .into_response()) +} + async fn find_channel( params: HashMap, dcom: Arc, @@ -91,20 +136,26 @@ async fn find_channel( } async fn channel_add_inner(params: HashMap, dcom: Arc) -> Result<(), Error> { - if let (Some(backend), Some(name)) = (params.get("backend"), params.get("name")) { - error!("TODO channel_add_inner"); - Err(Error::with_msg_no_trace(format!("TODO channel_add_inner"))) + if let Some(name) = params.get("name") { + let ch = crate::daemon_common::Channel::new(name.into()); + let (tx, rx) = async_channel::bounded(1); + let ev = DaemonEvent::ChannelAdd(ch, tx); + dcom.tx.send(ev).await?; + match rx.recv().await { + Ok(Ok(())) => Ok(()), + Ok(Err(e)) => Err(Error::with_msg_no_trace(format!("{e}"))), + Err(e) => Err(Error::with_msg_no_trace(format!("{e}"))), + } } else { Err(Error::with_msg_no_trace(format!("wrong parameters given"))) } } -async fn channel_add(params: HashMap, dcom: Arc) -> axum::Json { - let ret = match channel_add_inner(params, dcom).await { - Ok(_) => true, - Err(_) => false, - }; - axum::Json(ret) +async fn channel_add(params: HashMap, dcom: Arc) -> Result, Response> { + match channel_add_inner(params, dcom).await { + Ok(_) => Ok(axum::Json::from(true)), + Err(e) => Err(e.to_public_err_msg().into_response()), + } } async fn channel_remove(params: HashMap, dcom: Arc) -> axum::Json { @@ -198,7 +249,8 @@ fn make_routes(dcom: Arc, connset_cmd_tx: Sender, st .route( "/path2", get(|qu: Query| async move { (StatusCode::OK, format!("{qu:?}")) }), - ), + ) + .route("/path3/", get(|| async { (StatusCode::OK, format!("Hello there!")) })), ) .route( "/metrics", @@ -236,6 +288,10 @@ fn make_routes(dcom: Arc, connset_cmd_tx: Sender, st } }), ) + .route( + "/daqingest/always-error/", + get(|Query(params): Query>| always_error(params)), + ) .route( "/daqingest/find/channel", get({ @@ -326,11 +382,7 @@ pub async fn metrics_service( .unwrap() } -pub async fn metrics_agg_task( - query_item_chn: WeakSender, - local_stats: Arc, - store_stats: Arc, -) -> Result<(), Error> { +pub async fn metrics_agg_task(local_stats: Arc, store_stats: Arc) -> Result<(), Error> { let mut agg_last = CaConnStatsAgg::new(); loop { tokio::time::sleep(Duration::from_millis(671)).await; @@ -353,11 +405,6 @@ pub async fn metrics_agg_task( agg.push(g.stats()); } } - { - warn!("TODO provide metrics with a weak ref to the query_item_channel"); - let nitems = query_item_chn.upgrade().map_or(0, |x| x.len()); - agg.store_worker_recv_queue_len.__set(nitems as u64); - } #[cfg(DISABLED)] { let mut m = METRICS.lock().unwrap(); diff --git a/scywr/src/insertworker.rs b/scywr/src/insertworker.rs index caeb72c..f5711e9 100644 --- a/scywr/src/insertworker.rs +++ b/scywr/src/insertworker.rs @@ -22,6 +22,7 @@ use netpod::ScyllaConfig; use smallvec::smallvec; use smallvec::SmallVec; use stats::InsertWorkerStats; +use std::collections::VecDeque; use std::pin::Pin; use std::sync::atomic; use std::sync::atomic::AtomicU64; @@ -99,7 +100,7 @@ pub async fn spawn_scylla_insert_workers( insert_scylla_sessions: usize, insert_worker_count: usize, insert_worker_concurrency: usize, - item_inp: Receiver, + item_inp: Receiver>, insert_worker_opts: Arc, store_stats: Arc, use_rate_limit_queue: bool, @@ -324,7 +325,7 @@ async fn worker( async fn worker_streamed( worker_ix: usize, concurrency: usize, - item_inp: Receiver, + item_inp: Receiver>, ttls: Ttls, insert_worker_opts: Arc, data_store: Arc, @@ -338,29 +339,34 @@ async fn worker_streamed( // TODO possible without box? let item_inp = Box::pin(item_inp); let mut stream = item_inp - .map(|item| { + .map(|batch| { stats.item_recv.inc(); let tsnow_u64 = { let ts = SystemTime::now(); let epoch = ts.duration_since(std::time::UNIX_EPOCH).unwrap(); epoch.as_secs() * SEC + epoch.subsec_nanos() as u64 }; - match item { - QueryItem::Insert(item) => prepare_query_insert_futs(item, &ttls, &data_store, &stats, tsnow_u64), - QueryItem::ConnectionStatus(item) => { - stats.inserted_connection_status().inc(); - let fut = insert_connection_status_fut(item, &ttls, &data_store, stats.clone()); - smallvec![fut] - } - QueryItem::ChannelStatus(item) => { - stats.inserted_channel_status().inc(); - insert_channel_status_fut(item, &ttls, &data_store, stats.clone()) - } - _ => { - // TODO - SmallVec::new() - } + let mut res = Vec::with_capacity(32); + for item in batch { + let futs = match item { + QueryItem::Insert(item) => prepare_query_insert_futs(item, &ttls, &data_store, &stats, tsnow_u64), + QueryItem::ConnectionStatus(item) => { + stats.inserted_connection_status().inc(); + let fut = insert_connection_status_fut(item, &ttls, &data_store, stats.clone()); + smallvec![fut] + } + QueryItem::ChannelStatus(item) => { + stats.inserted_channel_status().inc(); + insert_channel_status_fut(item, &ttls, &data_store, stats.clone()) + } + _ => { + // TODO + SmallVec::new() + } + }; + res.extend(futs.into_iter()); } + res }) .map(|x| futures_util::stream::iter(x)) .flatten_unordered(Some(1)) @@ -416,7 +422,7 @@ fn prepare_query_insert_futs( let mut futs = smallvec![]; // TODO - if true || item_ts_local & 0x3f00000 < 0x0a00000 { + if true || item_ts_local & 0x3f00000 < 0x0600000 { let fut = insert_item_fut(item, &ttls, &data_store, do_insert, stats); futs.push(fut); if msp_bump { diff --git a/stats/src/stats.rs b/stats/src/stats.rs index 82bf792..7760131 100644 --- a/stats/src/stats.rs +++ b/stats/src/stats.rs @@ -286,6 +286,7 @@ stats_proc::stats_struct!(( poll_no_progress_no_pending, handle_add_channel_with_addr, create_ca_conn, + command_reply_fail, ), values( storage_insert_queue_len, @@ -382,7 +383,6 @@ stats_proc::stats_struct!(( insert_item_queue_full, channel_fast_item_drop, logic_error, - store_worker_recv_queue_len, // TODO maybe rename: this is now only the recv of the intermediate queue: store_worker_item_recv, // TODO rename to make clear that this drop is voluntary because of user config choice: @@ -500,7 +500,7 @@ fn test0_diff() { assert_eq!(diff.count0.load(), 3); } -pub fn xoshiro_from_time() -> rand_xoshiro::Xoshiro128StarStar { +pub fn xoshiro_from_time() -> rand_xoshiro::Xoshiro128PlusPlus { use rand_xoshiro::rand_core::SeedableRng; use std::time::SystemTime; let a = SystemTime::now() @@ -511,5 +511,5 @@ pub fn xoshiro_from_time() -> rand_xoshiro::Xoshiro128StarStar { .duration_since(SystemTime::UNIX_EPOCH) .unwrap() .subsec_nanos() as u64; - rand_xoshiro::Xoshiro128StarStar::seed_from_u64(a << 32 ^ b) + rand_xoshiro::Xoshiro128PlusPlus::seed_from_u64(a << 32 ^ b) }