diff --git a/Cargo.lock b/Cargo.lock index e618f9d..1ac0833 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -157,6 +157,12 @@ dependencies = [ "generic-array", ] +[[package]] +name = "bumpalo" +version = "3.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43" + [[package]] name = "byte-slice-cast" version = "1.2.3" @@ -181,6 +187,16 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" +[[package]] +name = "cc" +version = "1.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65193589c6404eb80b450d618eaf9a2cafaaafd57ecce47370519ef674a7bd44" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.1" @@ -339,7 +355,7 @@ version = "0.5.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76" dependencies = [ - "rand_core", + "rand_core 0.6.4", "subtle", ] @@ -538,6 +554,12 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "find-msvc-tools" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fd99930f64d146689264c637b5af2f0233a933bef0d8570e2526bf9e083192d" + [[package]] name = "find_cuda_helper" version = "0.2.0" @@ -554,7 +576,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "835c052cb0c08c1acf6ffd71c022172e18723949c8282f2b9f27efbc51e64534" dependencies = [ "byteorder", - "rand", + "rand 0.8.5", "rustc-hex", "static_assertions", ] @@ -586,6 +608,21 @@ version = "0.1.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a471a38ef8ed83cd6e40aa59c1ffe17db6855c18e3604d9c4ed8c08ebc28678" +[[package]] +name = "futures" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.31" @@ -602,6 +639,34 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" +[[package]] +name = "futures-executor" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" + +[[package]] +name = "futures-macro" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.103", +] + [[package]] name = "futures-sink" version = "0.3.31" @@ -620,9 +685,13 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", "futures-sink", "futures-task", + "memchr", "pin-project-lite", "pin-utils", "slab", @@ -646,7 +715,19 @@ checksum = "335ff9f135e4384c8150d6f27c6daed433577f86b4750418338c01a1a2528592" dependencies = [ "cfg-if", "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", +] + +[[package]] +name = "getrandom" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26145e563e54f2cadc477553f1ec5ee650b00862f0a58bcd12cbdc5f0ea2d2f4" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasi 0.14.7+wasi-0.2.4", ] [[package]] @@ -972,6 +1053,16 @@ version = "1.0.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c" +[[package]] +name = "js-sys" +version = "0.3.79" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6247da8b8658ad4e73a186e747fcc5fc2a29f979d6fe6269127fdb5fd08298d0" +dependencies = [ + "once_cell", + "wasm-bindgen", +] + [[package]] name = "keccak" version = "0.1.5" @@ -1081,6 +1172,7 @@ dependencies = [ "hex", "log", "metrics", + "miner-telemetry", "num_cpus", "pow-core", "primitive-types", @@ -1092,6 +1184,21 @@ dependencies = [ "warp", ] +[[package]] +name = "miner-telemetry" +version = "0.1.0" +dependencies = [ + "anyhow", + "futures 0.3.31", + "log", + "serde", + "serde_json", + "tokio", + "tokio-tungstenite 0.23.1", + "url", + "uuid", +] + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -1114,7 +1221,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78bed444cc8a2160f01cbcf811ef18cac863ad68ae8ca62092e8db51d51c761c" dependencies = [ "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "windows-sys 0.59.0", ] @@ -1150,7 +1257,7 @@ checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9" dependencies = [ "num-integer", "num-traits 0.2.19", - "rand", + "rand 0.8.5", ] [[package]] @@ -1215,7 +1322,7 @@ version = "0.19.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4297afb442d411793e4e24ee5a2977d15b6c95c743418f1c0ce0a2397d7ec8a3" dependencies = [ - "futures", + "futures 0.1.31", "nodrop", "num-traits 0.2.19", "ocl-core", @@ -1441,9 +1548,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "11778238e7d8b0e3ca62033fdc69e01ef5cdb08809cdc2398b2ce5ec873a1757" dependencies = [ "crossbeam", - "futures", + "futures 0.1.31", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "radium" version = "0.7.0" @@ -1457,8 +1570,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" dependencies = [ "libc", - "rand_chacha", - "rand_core", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.3", ] [[package]] @@ -1468,7 +1591,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.3", ] [[package]] @@ -1477,7 +1610,16 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ - "getrandom", + "getrandom 0.2.16", +] + +[[package]] +name = "rand_core" +version = "0.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "99d9a13982dcf210057a8a78572b2217b667c3beacbf3a0d8b454f6f82837d38" +dependencies = [ + "getrandom 0.3.3", ] [[package]] @@ -1526,6 +1668,20 @@ dependencies = [ "serde", ] +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.16", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rustacuda" version = "0.1.3" @@ -1576,6 +1732,39 @@ dependencies = [ "semver", ] +[[package]] +name = "rustls" +version = "0.23.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0ebcbd2f03de0fc1122ad9bb24b127a5a6cd51d72604a3f3c50ac459762b6cc" +dependencies = [ + "once_cell", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "229a4a4c221013e7e1f1a043678c5cc39fe5171437c88fb47151a21e6f5b5c79" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8572f3c2cb9934231157b45499fc41e1f58c589fdfb81a844ba873265e80f8eb" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.21" @@ -1682,6 +1871,12 @@ dependencies = [ "keccak", ] +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + [[package]] name = "signal-hook-registry" version = "1.4.5" @@ -1850,6 +2045,16 @@ dependencies = [ "syn 2.0.103", ] +[[package]] +name = "tokio-rustls" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e727b36a1a0e8b74c376ac2211e40c2c8af09fb4013c60d910495810f008e9b" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.21.0" @@ -1859,7 +2064,23 @@ dependencies = [ "futures-util", "log", "tokio", - "tungstenite", + "tungstenite 0.21.0", +] + +[[package]] +name = "tokio-tungstenite" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6989540ced10490aaf14e6bad2e3d33728a2813310a0c71d1574304c49631cd" +dependencies = [ + "futures-util", + "log", + "rustls", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite 0.23.0", + "webpki-roots 0.26.11", ] [[package]] @@ -1936,13 +2157,33 @@ dependencies = [ "http 1.3.1", "httparse", "log", - "rand", + "rand 0.8.5", "sha1", "thiserror", "url", "utf-8", ] +[[package]] +name = "tungstenite" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2e2ce1e47ed2994fd43b04c8f618008d4cabdd5ee34027cf14f9d918edd9c8" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http 1.3.1", + "httparse", + "log", + "rand 0.8.5", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror", + "utf-8", +] + [[package]] name = "typenum" version = "1.18.0" @@ -1979,6 +2220,12 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + [[package]] name = "url" version = "2.5.4" @@ -2008,6 +2255,18 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "1.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f87b8aa10b915a06587d0dec516c282ff295b475d94abf425d62b57710070a2" +dependencies = [ + "getrandom 0.3.3", + "js-sys", + "rand 0.9.2", + "wasm-bindgen", +] + [[package]] name = "vek" version = "0.15.10" @@ -2058,7 +2317,7 @@ dependencies = [ "serde_json", "serde_urlencoded", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.21.0", "tokio-util", "tower-service", "tracing", @@ -2070,6 +2329,101 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasi" +version = "0.14.7+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "883478de20367e224c0090af9cf5f9fa85bed63a95c1abf3afc5c083ebc06e8c" +dependencies = [ + "wasip2", +] + +[[package]] +name = "wasip2" +version = "1.0.1+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0562428422c63773dad2c345a1882263bbf4d65cf3f42e90921f787ef5ad58e7" +dependencies = [ + "wit-bindgen", +] + +[[package]] +name = "wasm-bindgen" +version = "0.2.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ad224d2776649cfb4f4471124f8176e54c1cca67a88108e30a0cd98b90e7ad3" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-backend" +version = "0.2.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a1364104bdcd3c03f22b16a3b1c9620891469f5e9f09bc38b2db121e593e732" +dependencies = [ + "bumpalo", + "log", + "proc-macro2", + "quote", + "syn 2.0.103", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d7ab4ca3e367bb1ed84ddbd83cc6e41e115f8337ed047239578210214e36c76" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a518014843a19e2dbbd0ed5dfb6b99b23fb886b14e6192a00803a3e14c552b0" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.103", + "wasm-bindgen-backend", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "255eb0aa4cc2eea3662a00c2bbd66e93911b7361d5e0fcd62385acfd7e15dcee" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.2", +] + +[[package]] +name = "webpki-roots" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e8983c3ab33d6fb807cfcdad2491c4ea8cbc8ed839181c7dfd9c67c83e261b2" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "winapi-util" version = "0.1.9" @@ -2170,6 +2524,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "wit-bindgen" +version = "0.46.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f17a85883d4e6d00e8a97c586de764dabcc06133f7f1d55dce5cdc070ad7fe59" + [[package]] name = "writeable" version = "0.6.1" @@ -2250,6 +2610,12 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zeroize" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ced3678a2879b30306d323f4542626697a464a97c0a07c9aebf7ebca65cd4dde" + [[package]] name = "zerotrie" version = "0.2.2" diff --git a/Cargo.toml b/Cargo.toml index c69714f..6bae6f8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,7 @@ members = [ "crates/metrics", "crates/miner-cli", "crates/miner-service", + "crates/miner-telemetry", "crates/pow-core", ] default-members = ["crates/miner-cli"] diff --git a/crates/miner-cli/src/main.rs b/crates/miner-cli/src/main.rs index 8bc9662..3901c82 100644 --- a/crates/miner-cli/src/main.rs +++ b/crates/miner-cli/src/main.rs @@ -44,6 +44,49 @@ struct Args { /// Note: GPU engines are currently unimplemented and will return a clear error at runtime. #[arg(long, env = "MINER_ENGINE", value_enum, default_value_t = EngineCli::CpuFast)] engine: EngineCli, + + /// Telemetry endpoints (repeat --telemetry-endpoint or comma-separated) + #[arg(long = "telemetry-endpoint", env = "MINER_TELEMETRY_ENDPOINTS", value_delimiter = ',', num_args = 0.., value_name = "URL")] + telemetry_endpoints: Option>, + + /// Enable or disable telemetry explicitly + #[arg(long = "telemetry-enabled", env = "MINER_TELEMETRY_ENABLED")] + telemetry_enabled: Option, + + /// Telemetry verbosity level (0..=4 typical) + #[arg(long = "telemetry-verbosity", env = "MINER_TELEMETRY_VERBOSITY")] + telemetry_verbosity: Option, + + /// Interval seconds for system.interval messages + #[arg( + long = "telemetry-interval-secs", + env = "MINER_TELEMETRY_INTERVAL_SECS" + )] + telemetry_interval_secs: Option, + + /// Default association: chain name + #[arg(long = "telemetry-chain", env = "MINER_TELEMETRY_CHAIN")] + telemetry_chain: Option, + + /// Default association: genesis hash (hex) + #[arg(long = "telemetry-genesis", env = "MINER_TELEMETRY_GENESIS")] + telemetry_genesis: Option, + + /// Default association: node telemetry id + #[arg(long = "telemetry-node-id", env = "MINER_TELEMETRY_NODE_ID")] + telemetry_node_id: Option, + + /// Default association: node libp2p peer id + #[arg(long = "telemetry-node-peer-id", env = "MINER_TELEMETRY_NODE_PEER_ID")] + telemetry_node_peer_id: Option, + + /// Default association: node name + #[arg(long = "telemetry-node-name", env = "MINER_TELEMETRY_NODE_NAME")] + telemetry_node_name: Option, + + /// Default association: node version + #[arg(long = "telemetry-node-version", env = "MINER_TELEMETRY_NODE_VERSION")] + telemetry_node_version: Option, } #[allow(clippy::enum_variant_names)] @@ -87,6 +130,40 @@ async fn main() { } env_logger::init(); + // Telemetry CLI passthrough to env for miner-service bootstrap + if let Some(eps) = args.telemetry_endpoints.as_ref() { + if !eps.is_empty() { + std::env::set_var("MINER_TELEMETRY_ENDPOINTS", eps.join(",")); + } + } + if let Some(v) = args.telemetry_enabled { + std::env::set_var("MINER_TELEMETRY_ENABLED", if v { "1" } else { "0" }); + } + if let Some(v) = args.telemetry_verbosity { + std::env::set_var("MINER_TELEMETRY_VERBOSITY", v.to_string()); + } + if let Some(v) = args.telemetry_interval_secs { + std::env::set_var("MINER_TELEMETRY_INTERVAL_SECS", v.to_string()); + } + if let Some(v) = args.telemetry_chain.as_ref() { + std::env::set_var("MINER_TELEMETRY_CHAIN", v); + } + if let Some(v) = args.telemetry_genesis.as_ref() { + std::env::set_var("MINER_TELEMETRY_GENESIS", v); + } + if let Some(v) = args.telemetry_node_id.as_ref() { + std::env::set_var("MINER_TELEMETRY_NODE_ID", v); + } + if let Some(v) = args.telemetry_node_peer_id.as_ref() { + std::env::set_var("MINER_TELEMETRY_NODE_PEER_ID", v); + } + if let Some(v) = args.telemetry_node_name.as_ref() { + std::env::set_var("MINER_TELEMETRY_NODE_NAME", v); + } + if let Some(v) = args.telemetry_node_version.as_ref() { + std::env::set_var("MINER_TELEMETRY_NODE_VERSION", v); + } + // Log effective configuration (concise; see ServiceConfig Display) log::info!("Starting external miner service..."); diff --git a/crates/miner-service/Cargo.toml b/crates/miner-service/Cargo.toml index f2938e9..48432d5 100644 --- a/crates/miner-service/Cargo.toml +++ b/crates/miner-service/Cargo.toml @@ -46,3 +46,4 @@ engine-montgomery = { path = "../engine-montgomery", optional = true } engine-gpu-cuda = { path = "../engine-gpu-cuda", optional = true, features = ["cuda"] } engine-gpu-opencl = { path = "../engine-gpu-opencl", optional = true } metrics = { path = "../metrics", optional = true } +miner-telemetry = { path = "../miner-telemetry" } diff --git a/crates/miner-service/src/lib.rs b/crates/miner-service/src/lib.rs index bbfa992..dd35705 100644 --- a/crates/miner-service/src/lib.rs +++ b/crates/miner-service/src/lib.rs @@ -1254,6 +1254,126 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> { // Start mining loop service.start_mining_loop().await; + // Telemetry bootstrap from environment variables (optional) + let telemetry_handle_opt = { + let endpoints: Vec = std::env::var("MINER_TELEMETRY_ENDPOINTS") + .ok() + .map(|s| { + s.split(',') + .map(|p| p.trim().to_string()) + .filter(|p| !p.is_empty()) + .collect() + }) + .unwrap_or_default(); + + let enabled = std::env::var("MINER_TELEMETRY_ENABLED") + .ok() + .map(|v| v != "0" && !v.eq_ignore_ascii_case("false")) + .unwrap_or(!endpoints.is_empty()); + + let verbosity = std::env::var("MINER_TELEMETRY_VERBOSITY") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(0); + + if enabled && !endpoints.is_empty() { + let interval_secs = std::env::var("MINER_TELEMETRY_INTERVAL_SECS") + .ok() + .and_then(|v| v.parse::().ok()); + let chain = std::env::var("MINER_TELEMETRY_CHAIN").ok(); + let genesis = std::env::var("MINER_TELEMETRY_GENESIS").ok(); + + // Optional linked node info + let link = miner_telemetry::TelemetryNodeLink { + node_telemetry_id: std::env::var("MINER_TELEMETRY_NODE_ID").ok(), + node_peer_id: std::env::var("MINER_TELEMETRY_NODE_PEER_ID").ok(), + node_name: std::env::var("MINER_TELEMETRY_NODE_NAME").ok(), + node_version: std::env::var("MINER_TELEMETRY_NODE_VERSION").ok(), + chain: chain.clone(), + genesis_hash: genesis.clone(), + }; + let default_link = if link.node_telemetry_id.is_some() + || link.node_peer_id.is_some() + || link.node_name.is_some() + || link.node_version.is_some() + || link.chain.is_some() + || link.genesis_hash.is_some() + { + Some(link) + } else { + None + }; + + let cfg = miner_telemetry::TelemetryConfig { + enabled, + endpoints, + verbosity, + name: Some("quantus-miner".to_string()), + implementation: Some("quantus-miner".to_string()), + version: Some( + option_env!("MINER_VERSION") + .unwrap_or(env!("CARGO_PKG_VERSION")) + .to_string(), + ), + chain, + genesis_hash: genesis, + interval_secs, + default_link, + }; + Some(miner_telemetry::start(cfg)) + } else { + None + } + }; + + if let Some(telemetry) = telemetry_handle_opt { + log::info!("Telemetry enabled; session_id={}", telemetry.session_id()); + telemetry.emit_system_connected(None).await; + + let telemetry_handle = telemetry.clone(); + let svc = service.clone(); + let engine_name_str = engine.name().to_string(); + let interval_secs = std::env::var("MINER_TELEMETRY_INTERVAL_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(15); + let start_instant = std::time::Instant::now(); + + tokio::spawn(async move { + loop { + let (active_jobs, total_rate) = { + let jobs = svc.jobs.lock().await; + let mut running = 0i64; + let mut rate = 0.0; + for (_id, job) in jobs.iter() { + if job.status == JobStatus::Running { + running += 1; + rate += job.last_hash_rate; + } + } + (running, rate) + }; + + let uptime_ms = start_instant.elapsed().as_millis() as u64; + + let interval = miner_telemetry::SystemInterval { + uptime_ms, + engine: Some(engine_name_str.clone()), + workers: Some(svc.workers as u32), + hash_rate: Some(total_rate), + active_jobs: Some(active_jobs), + linked_node_hint: None, + }; + + telemetry_handle.emit_system_interval(&interval, None).await; + + tokio::time::sleep(tokio::time::Duration::from_secs(interval_secs)).await; + } + }); + } else { + log::info!("Telemetry disabled (no endpoints configured)"); + } + // Optionally start metrics exporter if enabled via CLI and feature flag. if let Some(port) = config.metrics_port { #[cfg(feature = "metrics")] diff --git a/crates/miner-telemetry/Cargo.toml b/crates/miner-telemetry/Cargo.toml new file mode 100644 index 0000000..354f248 --- /dev/null +++ b/crates/miner-telemetry/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "miner-telemetry" +version = "0.1.0" +edition = "2021" +publish = false +description = "Telemetry client for Quantus External Miner (Polkadot SDK-compatible envelopes over WebSocket)" + +[lib] +name = "miner_telemetry" +path = "src/lib.rs" + +[features] +default = [] + +[dependencies] +# Dependencies (alphabetical) +anyhow = { workspace = true } +futures = "0.3" +log = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tokio-tungstenite = { version = "0.23", features = ["rustls-tls-webpki-roots"] } +url = "2" +uuid = { version = "1", features = ["fast-rng", "v4"] } diff --git a/crates/miner-telemetry/src/lib.rs b/crates/miner-telemetry/src/lib.rs new file mode 100644 index 0000000..1cf2a74 --- /dev/null +++ b/crates/miner-telemetry/src/lib.rs @@ -0,0 +1,471 @@ +#![deny(rust_2018_idioms)] +#![forbid(unsafe_code)] + +//! Minimal telemetry client for the Quantus External Miner. +//! +//! Capabilities: +//! - Connects to one or more telemetry endpoints (wss://...) asynchronously. +//! - Sends upstream-compatible envelopes with a stable per-process session id. +//! - Provides helpers to emit `system.connected` and `system.interval` events. +//! - Non-blocking: messages are buffered via bounded channels; on overflow, they are dropped. +//! - Auto-reconnects on send error; subsequent messages trigger reconnect attempts. +//! +//! Notes: +//! - This is intentionally small and self-contained. It does not embed Substrate's sc-telemetry. +//! - Endpoints must currently be WebSocket URLs (e.g., wss://telemetry.example/submit/). +//! - If no endpoints are configured or telemetry is disabled, the handle becomes a no-op. + +use anyhow::Result; +use futures::{SinkExt, StreamExt}; +use log::{debug, info, warn}; +use serde::Serialize; +use serde_json::{json, Value}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::net::TcpStream; +use tokio::sync::mpsc; +use tokio::time::sleep; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::MaybeTlsStream; +use tokio_tungstenite::WebSocketStream; +use uuid::Uuid; + +/// Bounded channel capacity per endpoint writer task. +const TELEMETRY_CHANNEL_CAPACITY: usize = 1024; + +/// Telemetry configuration. +#[derive(Clone, Debug, Default)] +pub struct TelemetryConfig { + /// Enable/disable telemetry entirely. + pub enabled: bool, + /// WebSocket endpoints to send telemetry to (e.g., "wss://telemetry.example/submit/"). + pub endpoints: Vec, + /// Default verbosity (u8), applied if not overridden per message. + pub verbosity: u8, + /// Optional static identity to put in system.connected. + pub name: Option, + pub implementation: Option, + pub version: Option, + pub chain: Option, + pub genesis_hash: Option, + /// Interval (seconds) recommendation for system.interval from the service (not enforced here). + pub interval_secs: Option, + /// Optional default association with a node (used when job-specific context doesn't provide one). + pub default_link: Option, +} + +/// Optional association with a node that the miner is working for. +#[derive(Clone, Debug, Default, Serialize)] +pub struct TelemetryNodeLink { + #[serde(skip_serializing_if = "Option::is_none")] + pub node_telemetry_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub node_peer_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub node_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub node_version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub chain: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub genesis_hash: Option, +} + +/// Snapshot used to populate `system.interval` payloads. +#[derive(Clone, Debug, Default)] +pub struct SystemInterval { + pub uptime_ms: u64, + pub engine: Option, + pub workers: Option, + pub hash_rate: Option, + pub active_jobs: Option, + /// If you want to hint which node is primarily linked when multiple jobs exist. + pub linked_node_hint: Option, +} + +/// Cloneable handle to send telemetry events. +#[derive(Clone)] +pub struct TelemetryHandle { + session_id: String, + verbosity: u8, + // One sender per endpoint task. + senders: Arc>>, + // Keep config for convenience payload builders. + config: Arc, +} + +impl TelemetryHandle { + /// Returns true if this handle will actually send to any endpoint. + pub fn is_enabled(&self) -> bool { + !self.senders.is_empty() + } + + /// Returns the stable session id used in the envelope `id`. + pub fn session_id(&self) -> &str { + &self.session_id + } + + /// Emit a fully custom payload (must include `"msg"`). + pub async fn emit_payload(&self, payload: Value, verbosity: Option) { + if self.senders.is_empty() { + return; + } + let msg = EnvelopeMsg { payload, verbosity }; + for tx in self.senders.iter() { + // Drop on full; do not block mining paths. This is intentionally lossy under backpressure. + if tx.try_send(msg.clone()).is_err() { + // Best-effort: if try_send failed due to full buffer, attempt bounded send with small timeout. + // Avoid blocking indefinitely. + let tx = tx.clone(); + let msg = msg.clone(); + tokio::spawn(async move { + let _ = tx.send(msg).await; + }); + } + } + } + + /// Emit a `system.connected` message using configuration defaults and an optional link. + pub async fn emit_system_connected(&self, link: Option<&TelemetryNodeLink>) { + let payload = build_system_connected_payload(&self.config, link); + self.emit_payload(payload, Some(self.verbosity)).await; + } + + /// Emit a `system.interval` message with lightweight health stats and an optional link hint. + pub async fn emit_system_interval(&self, interval: &SystemInterval, link_hint: Option<&str>) { + let mut payload = json!({ + "msg": "system.interval", + "uptime_ms": interval.uptime_ms, + }); + + if let Some(engine) = &interval.engine { + payload + .as_object_mut() + .unwrap() + .insert("engine".to_string(), json!(engine)); + } + if let Some(workers) = interval.workers { + payload + .as_object_mut() + .unwrap() + .insert("workers".to_string(), json!(workers)); + } + if let Some(hash_rate) = interval.hash_rate { + payload + .as_object_mut() + .unwrap() + .insert("hash_rate".to_string(), json!(hash_rate)); + } + if let Some(active_jobs) = interval.active_jobs { + payload + .as_object_mut() + .unwrap() + .insert("active_jobs".to_string(), json!(active_jobs)); + } + if let Some(hint) = link_hint { + payload + .as_object_mut() + .unwrap() + .insert("linked_node_hint".to_string(), json!(hint)); + } + + self.emit_payload(payload, Some(self.verbosity)).await; + } +} + +/// Build a `system.connected` payload with upstream-compatible keys and optional `linked_node`. +fn build_system_connected_payload( + config: &TelemetryConfig, + link: Option<&TelemetryNodeLink>, +) -> Value { + let name = config + .name + .as_deref() + .unwrap_or("quantus-miner") + .to_string(); + let implementation = config + .implementation + .as_deref() + .unwrap_or("quantus-miner") + .to_string(); + // Prefer build-time env var MINER_VERSION or fallback to crate version if the caller sets it. + let version = config.version.clone().unwrap_or_else(|| { + option_env!("MINER_VERSION") + .map(|s| s.to_string()) + .unwrap_or_else(|| env!("CARGO_PKG_VERSION").to_string()) + }); + + let mut payload = json!({ + "msg": "system.connected", + "name": name, + "implementation": implementation, + "version": version, + "authority": false, + "platform": { + "os": std::env::consts::OS, + "arch": std::env::consts::ARCH + } + }); + + if let Some(chain) = &config.chain { + payload + .as_object_mut() + .unwrap() + .insert("chain".to_string(), json!(chain)); + } + if let Some(gh) = &config.genesis_hash { + payload + .as_object_mut() + .unwrap() + .insert("genesis_hash".to_string(), json!(gh)); + } + + let linked = link.or(config.default_link.as_ref()); + if let Some(linked_node) = linked { + let v = serde_json::to_value(linked_node).unwrap_or(json!({})); + if let Some(obj) = v.as_object() { + // Only include non-empty association + if !obj.is_empty() { + payload + .as_object_mut() + .unwrap() + .insert("linked_node".to_string(), Value::Object(obj.clone())); + } + } + } + + payload +} + +/// Telemetry envelope to send to the transport. +#[derive(Clone, Debug)] +struct EnvelopeMsg { + payload: Value, + verbosity: Option, +} + +/// Start the telemetry subsystem with the given configuration. +/// Returns a `TelemetryHandle` for emitting events. If `config.enabled` is false +/// or no endpoints are specified, the handle is a no-op. +pub fn start(config: TelemetryConfig) -> TelemetryHandle { + let enabled = config.enabled && !config.endpoints.is_empty(); + let session_id = Uuid::new_v4().to_string(); + let verbosity = config.verbosity; + let cfg = Arc::new(config); + + let mut senders: Vec> = Vec::new(); + + if enabled { + for ep in &cfg.endpoints { + let (tx, rx) = mpsc::channel::(TELEMETRY_CHANNEL_CAPACITY); + senders.push(tx); + spawn_endpoint_writer_task(ep.clone(), rx, session_id.clone(), verbosity); + } + info!( + "miner-telemetry: started with {} endpoint(s), session_id={}", + cfg.endpoints.len(), + session_id + ); + } else { + info!("miner-telemetry: disabled (no endpoints or disabled in config)"); + } + + TelemetryHandle { + session_id, + verbosity, + senders: Arc::new(senders), + config: cfg, + } +} + +/// Spawn a writer task for a single endpoint. It owns a receiver and attempts to keep +/// a write connection alive, reconnecting on failure. Messages received while disconnected +/// are dropped (best-effort model). +fn spawn_endpoint_writer_task( + endpoint: String, + mut rx: mpsc::Receiver, + session_id: String, + default_verbosity: u8, +) { + tokio::spawn(async move { + // Writer half of the websocket. + type WsWrite = + futures::stream::SplitSink>, Message>; + + let mut writer: Option = None; + let mut backoff_secs: u64 = 1; + + loop { + // Receive next message to send. If all senders are gone, exit task gracefully. + let Some(msg) = rx.recv().await else { + debug!( + "miner-telemetry: endpoint task for {} shutting down (channel closed)", + endpoint + ); + break; + }; + + // Ensure connectivity prior to sending. + if writer.is_none() { + match connect_ws(&endpoint).await { + Ok(w) => { + writer = Some(w); + backoff_secs = 1; // reset backoff after success + debug!("miner-telemetry: connected to {}", endpoint); + } + Err(e) => { + warn!( + "miner-telemetry: connect failed to {}: {e}, backing off {}s", + endpoint, backoff_secs + ); + sleep(Duration::from_secs(backoff_secs)).await; + backoff_secs = (backoff_secs.saturating_mul(2)).min(30); + // Drop this message and continue; newer messages will trigger subsequent attempts. + continue; + } + } + } + + // Build envelope with timestamp and (optional) verbosity. + let ts = now_ms(); + let verbosity = msg.verbosity.unwrap_or(default_verbosity); + let envelope = json!({ + "id": session_id, + "ts": ts, + "verbosity": verbosity, + "payload": msg.payload + }); + let line = match serde_json::to_string(&envelope) { + Ok(s) => s, + Err(e) => { + warn!("miner-telemetry: failed to serialize envelope: {e}"); + continue; + } + }; + + // Send over WebSocket + let mut send_ok = false; + if let Some(w) = writer.as_mut() { + match w.send(Message::Text(line)).await { + Ok(()) => { + send_ok = true; + } + Err(e) => { + warn!( + "miner-telemetry: send error to {}: {e}. Will drop connection and reconnect.", + endpoint + ); + } + } + } + + if !send_ok { + writer = None; // drop and reconnect on next message + // Small backoff to avoid hot-loop if immediate failures keep happening. + sleep(Duration::from_millis(250)).await; + } + } + }); +} + +/// Establish a websocket connection and return the writer half. +/// A reader task is also spawned to continuously drain incoming frames (pings, acks, etc.). +async fn connect_ws( + endpoint: &str, +) -> Result>, Message>> { + let url = endpoint.to_string(); + let (ws, _resp) = tokio_tungstenite::connect_async(url).await?; + let (write, mut read) = ws.split(); + + // Spawn a task to drain incoming frames to keep the connection healthy. + tokio::spawn(async move { + while let Some(msg) = read.next().await { + match msg { + Ok(Message::Ping(data)) => { + debug!("miner-telemetry: received ping ({} bytes)", data.len()); + } + Ok(Message::Pong(_)) => { + // no-op + } + Ok(Message::Text(_)) | Ok(Message::Binary(_)) => { + // Most telemetry servers don't push data; ignore. + } + Ok(Message::Frame(_)) => { + // Ignore low-level frames; keep connection alive + } + Ok(Message::Close(frame)) => { + debug!("miner-telemetry: server closed connection: {:?}", frame); + break; + } + Err(e) => { + debug!("miner-telemetry: read error: {e}"); + break; + } + } + } + debug!("miner-telemetry: reader task ended"); + }); + + Ok(write) +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_else(|_| Duration::from_secs(0)) + .as_millis() as u64 +} + +// ------------------------------------------------------------------------------------- +// Tests (basic serialization and handle behavior) +// ------------------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_build_connected_payload() { + let cfg = TelemetryConfig { + enabled: true, + endpoints: vec![], + verbosity: 0, + name: Some("qm".into()), + implementation: Some("quantus-miner".into()), + version: Some("0.0.1".into()), + chain: Some("Resonance".into()), + genesis_hash: Some("0xabc".into()), + interval_secs: Some(15), + default_link: Some(TelemetryNodeLink { + node_telemetry_id: Some("node-uuid".into()), + node_peer_id: Some("12D3KooW..".into()), + node_name: Some("node01".into()), + node_version: Some("1.2.3".into()), + chain: Some("Resonance".into()), + genesis_hash: Some("0xabc".into()), + }), + }; + let payload = build_system_connected_payload(&cfg, None); + assert_eq!( + payload.get("msg").and_then(|v| v.as_str()), + Some("system.connected") + ); + assert_eq!(payload.get("name").and_then(|v| v.as_str()), Some("qm")); + assert_eq!( + payload.get("chain").and_then(|v| v.as_str()), + Some("Resonance") + ); + assert!(payload.get("linked_node").is_some()); + } + + #[tokio::test] + async fn test_handle_noop_when_disabled() { + let cfg = TelemetryConfig::default(); + let handle = start(cfg); + assert!(!handle.is_enabled()); + handle + .emit_system_interval(&SystemInterval::default(), None) + .await; + handle.emit_system_connected(None).await; + } +}