Add miner-telemetry crate and integrate telemetry support
- Introduce miner-telemetry crate for WebSocket-based telemetry reporting - Add telemetry configuration options to miner-cli CLI - Pass telemetry CLI args as environment variables for miner-service - Enable miner-service to bootstrap telemetry from env vars and emit events - Emit periodic system.interval telemetry with mining stats and metadata - Add miner-telemetry as dependency to miner-service and workspace member
This commit is contained in:
rob thijssen
Generated
+380
-14
@@ -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"
|
||||
|
||||
@@ -7,6 +7,7 @@ members = [
|
||||
"crates/metrics",
|
||||
"crates/miner-cli",
|
||||
"crates/miner-service",
|
||||
"crates/miner-telemetry",
|
||||
"crates/pow-core",
|
||||
]
|
||||
default-members = ["crates/miner-cli"]
|
||||
|
||||
@@ -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<Vec<String>>,
|
||||
|
||||
/// Enable or disable telemetry explicitly
|
||||
#[arg(long = "telemetry-enabled", env = "MINER_TELEMETRY_ENABLED")]
|
||||
telemetry_enabled: Option<bool>,
|
||||
|
||||
/// Telemetry verbosity level (0..=4 typical)
|
||||
#[arg(long = "telemetry-verbosity", env = "MINER_TELEMETRY_VERBOSITY")]
|
||||
telemetry_verbosity: Option<u8>,
|
||||
|
||||
/// Interval seconds for system.interval messages
|
||||
#[arg(
|
||||
long = "telemetry-interval-secs",
|
||||
env = "MINER_TELEMETRY_INTERVAL_SECS"
|
||||
)]
|
||||
telemetry_interval_secs: Option<u64>,
|
||||
|
||||
/// Default association: chain name
|
||||
#[arg(long = "telemetry-chain", env = "MINER_TELEMETRY_CHAIN")]
|
||||
telemetry_chain: Option<String>,
|
||||
|
||||
/// Default association: genesis hash (hex)
|
||||
#[arg(long = "telemetry-genesis", env = "MINER_TELEMETRY_GENESIS")]
|
||||
telemetry_genesis: Option<String>,
|
||||
|
||||
/// Default association: node telemetry id
|
||||
#[arg(long = "telemetry-node-id", env = "MINER_TELEMETRY_NODE_ID")]
|
||||
telemetry_node_id: Option<String>,
|
||||
|
||||
/// Default association: node libp2p peer id
|
||||
#[arg(long = "telemetry-node-peer-id", env = "MINER_TELEMETRY_NODE_PEER_ID")]
|
||||
telemetry_node_peer_id: Option<String>,
|
||||
|
||||
/// Default association: node name
|
||||
#[arg(long = "telemetry-node-name", env = "MINER_TELEMETRY_NODE_NAME")]
|
||||
telemetry_node_name: Option<String>,
|
||||
|
||||
/// Default association: node version
|
||||
#[arg(long = "telemetry-node-version", env = "MINER_TELEMETRY_NODE_VERSION")]
|
||||
telemetry_node_version: Option<String>,
|
||||
}
|
||||
|
||||
#[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...");
|
||||
|
||||
|
||||
@@ -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" }
|
||||
|
||||
@@ -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<String> = 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::<u8>().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::<u64>().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::<u64>().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")]
|
||||
|
||||
@@ -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"] }
|
||||
@@ -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<String>,
|
||||
/// Default verbosity (u8), applied if not overridden per message.
|
||||
pub verbosity: u8,
|
||||
/// Optional static identity to put in system.connected.
|
||||
pub name: Option<String>,
|
||||
pub implementation: Option<String>,
|
||||
pub version: Option<String>,
|
||||
pub chain: Option<String>,
|
||||
pub genesis_hash: Option<String>,
|
||||
/// Interval (seconds) recommendation for system.interval from the service (not enforced here).
|
||||
pub interval_secs: Option<u64>,
|
||||
/// Optional default association with a node (used when job-specific context doesn't provide one).
|
||||
pub default_link: Option<TelemetryNodeLink>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub node_peer_id: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub node_name: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub node_version: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub chain: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub genesis_hash: Option<String>,
|
||||
}
|
||||
|
||||
/// Snapshot used to populate `system.interval` payloads.
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct SystemInterval {
|
||||
pub uptime_ms: u64,
|
||||
pub engine: Option<String>,
|
||||
pub workers: Option<u32>,
|
||||
pub hash_rate: Option<f64>,
|
||||
pub active_jobs: Option<i64>,
|
||||
/// If you want to hint which node is primarily linked when multiple jobs exist.
|
||||
pub linked_node_hint: Option<String>,
|
||||
}
|
||||
|
||||
/// Cloneable handle to send telemetry events.
|
||||
#[derive(Clone)]
|
||||
pub struct TelemetryHandle {
|
||||
session_id: String,
|
||||
verbosity: u8,
|
||||
// One sender per endpoint task.
|
||||
senders: Arc<Vec<mpsc::Sender<EnvelopeMsg>>>,
|
||||
// Keep config for convenience payload builders.
|
||||
config: Arc<TelemetryConfig>,
|
||||
}
|
||||
|
||||
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<u8>) {
|
||||
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<u8>,
|
||||
}
|
||||
|
||||
/// 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<mpsc::Sender<EnvelopeMsg>> = Vec::new();
|
||||
|
||||
if enabled {
|
||||
for ep in &cfg.endpoints {
|
||||
let (tx, rx) = mpsc::channel::<EnvelopeMsg>(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<EnvelopeMsg>,
|
||||
session_id: String,
|
||||
default_verbosity: u8,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
// Writer half of the websocket.
|
||||
type WsWrite =
|
||||
futures::stream::SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>;
|
||||
|
||||
let mut writer: Option<WsWrite> = 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<futures::stream::SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, 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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user