improvements

refactor config
start unpicking influxdb
2026-02-04 18:53:05 -08:00 · 2026-02-04 14:25:17 -08:00 · 2026-02-04 13:50:24 -08:00 · 2026-02-04 11:50:27 -08:00 · 2026-02-04 11:49:56 -08:00 · 2026-02-04 11:49:24 -08:00
14 changed files with 1408 additions and 1158 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -20,15 +20,6 @@ dependencies = [
 "libc",
 ]

-[[package]]
-name = "ansi_term"
-version = "0.12.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d52a9bb7ec0cf484c551830a7ce27bd20d67eac647e1befb56b0be4ee39a55d2"
-dependencies = [
- "winapi",
-]
-
 [[package]]
 name = "anstream"
 version = "0.6.21"
@@ -159,6 +150,12 @@ version = "0.21.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567"

+[[package]]
+name = "base64"
+version = "0.22.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+
 [[package]]
 name = "bitflags"
 version = "1.3.2"
@@ -171,6 +168,15 @@ version = "2.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3"

+[[package]]
+name = "block2"
+version = "0.6.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cdeb9d870516001442e364c5220d3574d2da8dc765554b4a617230d33fa58ef5"
+dependencies = [
+ "objc2",
+]
+
 [[package]]
 name = "bumpalo"
 version = "3.19.1"
@@ -230,6 +236,7 @@ dependencies = [
 "chrono",
 "chrony-candm",
 "clap",
+ "ctrlc",
 "figment",
 "futures",
 "gethostname",
@@ -241,13 +248,14 @@ dependencies = [
 "serde_derive",
 "serde_json",
 "serde_repr",
+ "serde_with",
 "serialport",
 "tokio",
 "tokio-serial",
 "tokio-stream",
 "tokio-util",
 "tracing",
- "tracing-subscriber 0.3.22",
+ "tracing-subscriber",
 ]

 [[package]]
@@ -279,7 +287,7 @@ dependencies = [
 "num_enum",
 "rand",
 "siphasher",
- "thiserror 2.0.18",
+ "thiserror",
 "tokio",
 ]

@@ -387,13 +395,49 @@ dependencies = [
 ]

 [[package]]
-name = "dashmap"
-version = "4.0.2"
+name = "ctrlc"
+version = "3.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e77a43b28d0668df09411cb0bc9a8c2adc40f9a048afe863e05fd43251e8e39c"
+checksum = "73736a89c4aff73035ba2ed2e565061954da00d4970fc9ac25dcc85a2a20d790"
 dependencies = [
- "cfg-if",
- "num_cpus",
+ "dispatch2",
+ "nix 0.30.1",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "darling"
+version = "0.21.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9cdf337090841a411e2a7f3deb9187445851f91b309c0c0a29e05f74a00a48c0"
+dependencies = [
+ "darling_core",
+ "darling_macro",
+]
+
+[[package]]
+name = "darling_core"
+version = "0.21.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1247195ecd7e3c85f83c8d2a366e4210d588e802133e1e355180a9870b517ea4"
+dependencies = [
+ "fnv",
+ "ident_case",
+ "proc-macro2",
+ "quote",
+ "strsim",
+ "syn 2.0.114",
+]
+
+[[package]]
+name = "darling_macro"
+version = "0.21.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d38308df82d1080de0afee5d069fa14b0326a88c14f15c5ccda35b4a6c414c81"
+dependencies = [
+ "darling_core",
+ "quote",
+ "syn 2.0.114",
 ]

 [[package]]
@@ -403,6 +447,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ececcb659e7ba858fb4f10388c250a7252eb0a27373f1a72b8748afdd248e587"
 dependencies = [
 "powerfmt",
+ "serde_core",
+]
+
+[[package]]
+name = "dispatch2"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "89a09f22a6c6069a18470eb92d2298acf25463f14256d24778e1230d789a2aec"
+dependencies = [
+ "bitflags 2.10.0",
+ "block2",
+ "libc",
+ "objc2",
 ]

 [[package]]
@@ -423,10 +480,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "780955b8b195a21ab8e4ac6b60dd1dbdcec1dc6c51c0617964b08c81785e12c9"

 [[package]]
-name = "dotenv"
-version = "0.15.0"
+name = "dyn-clone"
+version = "1.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "77c90badedccf4105eca100756a0b1289e191f6fcbdadd3cee1d2f614f97da8f"
+checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"

 [[package]]
 name = "either"
@@ -672,13 +729,19 @@ dependencies = [
 "futures-sink",
 "futures-util",
 "http",
- "indexmap",
+ "indexmap 2.13.0",
 "slab",
 "tokio",
 "tokio-util",
 "tracing",
 ]

+[[package]]
+name = "hashbrown"
+version = "0.12.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"
+
 [[package]]
 name = "hashbrown"
 version = "0.16.1"
@@ -691,12 +754,6 @@ version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"

-[[package]]
-name = "hermit-abi"
-version = "0.5.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
-
 [[package]]
 name = "hex"
 version = "0.4.3"
@@ -879,6 +936,12 @@ dependencies = [
 "zerovec",
 ]

+[[package]]
+name = "ident_case"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
+
 [[package]]
 name = "idna"
 version = "1.1.0"
@@ -900,6 +963,17 @@ dependencies = [
 "icu_properties",
 ]

+[[package]]
+name = "indexmap"
+version = "1.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99"
+dependencies = [
+ "autocfg",
+ "hashbrown 0.12.3",
+ "serde",
+]
+
 [[package]]
 name = "indexmap"
 version = "2.13.0"
@@ -907,38 +981,33 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017"
 dependencies = [
 "equivalent",
- "hashbrown",
+ "hashbrown 0.16.1",
+ "serde",
+ "serde_core",
 ]

 [[package]]
 name = "influxdb2"
-version = "0.3.9"
+version = "0.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a152a72a5d8d387580a1c6df5dc078891ba914d8462da51f7286321680f998b4"
+checksum = "24cc9f9d9fee9ebda1a77b61769cde513e03ad09607b347602f4ea887657689e"
 dependencies = [
 "base64 0.13.1",
 "bytes",
 "chrono",
 "csv",
- "dotenv",
 "fallible-iterator",
 "futures",
 "go-parse-duration",
 "influxdb2-derive",
 "influxdb2-structmap",
- "nom",
- "opentelemetry",
 "ordered-float",
 "parking_lot",
 "reqwest",
+ "secrecy",
 "serde",
 "serde_json",
- "serde_qs",
- "smallvec",
 "snafu",
- "tempfile",
- "tracing",
- "tracing-subscriber 0.2.25",
 "url",
 ]

@@ -1099,22 +1168,13 @@ dependencies = [
 "libc",
 ]

-[[package]]
-name = "matchers"
-version = "0.0.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f099785f7595cc4b4553a174ce30dd7589ef93391ff414dbb67f62392b9e0ce1"
-dependencies = [
- "regex-automata 0.1.10",
-]
-
 [[package]]
 name = "matchers"
 version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"
 dependencies = [
- "regex-automata 0.4.13",
+ "regex-automata",
 ]

 [[package]]
@@ -1129,12 +1189,6 @@ version = "0.3.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"

-[[package]]
-name = "minimal-lexical"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
-
 [[package]]
 name = "mio"
 version = "1.1.1"
@@ -1201,13 +1255,15 @@ dependencies = [
 ]

 [[package]]
-name = "nom"
-version = "7.1.3"
+name = "nix"
+version = "0.30.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"
+checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6"
 dependencies = [
- "memchr",
- "minimal-lexical",
+ "bitflags 2.10.0",
+ "cfg-if",
+ "cfg_aliases",
+ "libc",
 ]

 [[package]]
@@ -1234,16 +1290,6 @@ dependencies = [
 "autocfg",
 ]

-[[package]]
-name = "num_cpus"
-version = "1.17.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b"
-dependencies = [
- "hermit-abi",
- "libc",
-]
-
 [[package]]
 name = "num_enum"
 version = "0.5.11"
@@ -1265,6 +1311,21 @@ dependencies = [
 "syn 1.0.109",
 ]

+[[package]]
+name = "objc2"
+version = "0.6.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7c2599ce0ec54857b29ce62166b0ed9b4f6f1a70ccc9a71165b6154caca8c05"
+dependencies = [
+ "objc2-encode",
+]
+
+[[package]]
+name = "objc2-encode"
+version = "4.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33"
+
 [[package]]
 name = "once_cell"
 version = "1.21.3"
@@ -1321,26 +1382,6 @@ dependencies = [
 "vcpkg",
 ]

-[[package]]
-name = "opentelemetry"
-version = "0.13.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b91cea1dfd50064e52db033179952d18c770cbc5dfefc8eba45d619357ba3914"
-dependencies = [
- "async-trait",
- "dashmap",
- "fnv",
- "futures",
- "js-sys",
- "lazy_static",
- "percent-encoding",
- "pin-project",
- "rand",
- "thiserror 1.0.69",
- "tokio",
- "tokio-stream",
-]
-
 [[package]]
 name = "ordered-float"
 version = "3.9.2"
@@ -1381,26 +1422,6 @@ version = "2.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"

-[[package]]
-name = "pin-project"
-version = "1.1.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "677f1add503faace112b9f1373e43e9e054bfdd22ff1a63c1bc485eaec6a6a8a"
-dependencies = [
- "pin-project-internal",
-]
-
-[[package]]
-name = "pin-project-internal"
-version = "1.1.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e918e4ff8c4549eb882f14b3a4bc8c8bc93de829416eacf579f1207a8fbf861"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.114",
-]
-
 [[package]]
 name = "pin-project-lite"
 version = "0.2.16"
@@ -1516,6 +1537,26 @@ dependencies = [
 "bitflags 1.3.2",
 ]

+[[package]]
+name = "ref-cast"
+version = "1.0.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f354300ae66f76f1c85c5f84693f0ce81d747e2c3f21a45fef496d89c960bf7d"
+dependencies = [
+ "ref-cast-impl",
+]
+
+[[package]]
+name = "ref-cast-impl"
+version = "1.0.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7186006dcb21920990093f30e3dea63b7d6e977bf1256be20c3563a5db070da"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.114",
+]
+
 [[package]]
 name = "regex"
 version = "1.12.2"
@@ -1524,17 +1565,8 @@ checksum = "843bc0191f75f3e22651ae5f1e72939ab2f72a4bc30fa80a066bd66edefc24d4"
 dependencies = [
 "aho-corasick",
 "memchr",
- "regex-automata 0.4.13",
- "regex-syntax 0.8.8",
-]
-
-[[package]]
-name = "regex-automata"
-version = "0.1.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132"
-dependencies = [
- "regex-syntax 0.6.29",
+ "regex-automata",
+ "regex-syntax",
 ]

 [[package]]
@@ -1545,15 +1577,9 @@ checksum = "5276caf25ac86c8d810222b3dbb938e512c55c6831a10f3e6ed1c93b84041f1c"
 dependencies = [
 "aho-corasick",
 "memchr",
- "regex-syntax 0.8.8",
+ "regex-syntax",
 ]

-[[package]]
-name = "regex-syntax"
-version = "0.6.29"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1"
-
 [[package]]
 name = "regex-syntax"
 version = "0.8.8"
@@ -1645,12 +1671,45 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

+[[package]]
+name = "schemars"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4cd191f9397d57d581cddd31014772520aa448f65ef991055d7f61582c65165f"
+dependencies = [
+ "dyn-clone",
+ "ref-cast",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "schemars"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a2b42f36aa1cd011945615b92222f6bf73c599a102a300334cd7f8dbeec726cc"
+dependencies = [
+ "dyn-clone",
+ "ref-cast",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "scopeguard"
 version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"

+[[package]]
+name = "secrecy"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9bd1c54ea06cfd2f6b63219704de0b9b4f72dcc2b8fdef820be6cd799780e91e"
+dependencies = [
+ "zeroize",
+]
+
 [[package]]
 name = "security-framework"
 version = "2.11.1"
@@ -1717,17 +1776,6 @@ dependencies = [
 "zmij",
 ]

-[[package]]
-name = "serde_qs"
-version = "0.10.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8cac3f1e2ca2fe333923a1ae72caca910b98ed0630bb35ef6f8c8517d6e81afa"
-dependencies = [
- "percent-encoding",
- "serde",
- "thiserror 1.0.69",
-]
-
 [[package]]
 name = "serde_repr"
 version = "0.1.20"
@@ -1760,6 +1808,37 @@ dependencies = [
 "serde",
 ]

+[[package]]
+name = "serde_with"
+version = "3.16.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4fa237f2807440d238e0364a218270b98f767a00d3dada77b1c53ae88940e2e7"
+dependencies = [
+ "base64 0.22.1",
+ "chrono",
+ "hex",
+ "indexmap 1.9.3",
+ "indexmap 2.13.0",
+ "schemars 0.9.0",
+ "schemars 1.2.1",
+ "serde_core",
+ "serde_json",
+ "serde_with_macros",
+ "time",
+]
+
+[[package]]
+name = "serde_with_macros"
+version = "3.16.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "52a8e3ca0ca629121f70ab50f95249e5a6f925cc0f6ffe8256c45b728875706c"
+dependencies = [
+ "darling",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.114",
+]
+
 [[package]]
 name = "serialport"
 version = "4.8.1"
@@ -1939,33 +2018,13 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

-[[package]]
-name = "thiserror"
-version = "1.0.69"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
-dependencies = [
- "thiserror-impl 1.0.69",
-]
-
 [[package]]
 name = "thiserror"
 version = "2.0.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4"
 dependencies = [
- "thiserror-impl 2.0.18",
-]
-
-[[package]]
-name = "thiserror-impl"
-version = "1.0.69"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.114",
+ "thiserror-impl",
 ]

 [[package]]
@@ -2131,7 +2190,7 @@ version = "0.19.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1b5bb770da30e5cbfde35a2d7b9b8a2c4b8ef89548a7a6aeab5c9a576e3e7421"
 dependencies = [
- "indexmap",
+ "indexmap 2.13.0",
 "toml_datetime",
 "winnow 0.5.40",
 ]
@@ -2142,7 +2201,7 @@ version = "0.22.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a"
 dependencies = [
- "indexmap",
+ "indexmap 2.13.0",
 "serde",
 "serde_spanned",
 "toml_datetime",
@@ -2205,48 +2264,16 @@ dependencies = [
 "tracing-core",
 ]

-[[package]]
-name = "tracing-serde"
-version = "0.1.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc6b213177105856957181934e4920de57730fc69bf42c37ee5bb664d406d9e1"
-dependencies = [
- "serde",
- "tracing-core",
-]
-
-[[package]]
-name = "tracing-subscriber"
-version = "0.2.25"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e0d2eaa99c3c2e41547cfa109e910a68ea03823cccad4a0525dcbc9b01e8c71"
-dependencies = [
- "ansi_term",
- "chrono",
- "lazy_static",
- "matchers 0.0.1",
- "parking_lot",
- "regex",
- "serde",
- "serde_json",
- "sharded-slab",
- "smallvec",
- "thread_local",
- "tracing",
- "tracing-core",
- "tracing-serde",
-]
-
 [[package]]
 name = "tracing-subscriber"
 version = "0.3.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2f30143827ddab0d256fd843b7a66d164e9f271cfa0dde49142c5ca0ca291f1e"
 dependencies = [
- "matchers 0.2.0",
+ "matchers",
 "nu-ansi-term",
 "once_cell",
- "regex-automata 0.4.13",
+ "regex-automata",
 "sharded-slab",
 "smallvec",
 "thread_local",
@@ -2277,7 +2304,7 @@ version = "0.1.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4064ed685c487dbc25bd3f0e9548f2e34bab9d18cefc700f9ec2dba74ba1138e"
 dependencies = [
- "thiserror 2.0.18",
+ "thiserror",
 ]

 [[package]]
@@ -2841,6 +2868,12 @@ dependencies = [
 "synstructure",
 ]

+[[package]]
+name = "zeroize"
+version = "1.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0"
+
 [[package]]
 name = "zerotrie"
 version = "0.2.3"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -10,7 +10,7 @@ release-logs = ["tracing/max_level_info"]
 [dependencies]
 serde = "1.0"
 serde_derive = "1.0"
-tokio = { version = "1", features = ["rt", "io-util"] }
+tokio = { version = "1", features = ["rt-multi-thread", "io-util"] }
 clap = { version = "4.0", features = ["derive"] }
 figment = { version = "0.10", features = ["toml"] }
 futures = "0.3.24"
@@ -32,8 +32,10 @@ tracing-subscriber = { version = "0.3.22", features = ["fmt", "ansi", "time", "e
 serialport = "4.8.1"
 gethostname = "1.1.0"
 bitflags = "2.10.0"
-influxdb2 = "0.3.9"
+influxdb2 = { version = "0.5.2" }
 chrono = "0.4.43"
+serde_with = "3.16.1"
+ctrlc = "3.5.1"

 [dependencies.chrony-candm]
 git = "https://github.com/aws/chrony-candm"
--- a/5
+++ b/5
@@ -1,6 +1,6 @@
-FROM rust:slim as builder
+FROM rust:slim AS builder

-RUN apt-get update && apt-get install -y libssl-dev pkg-config
+RUN apt-get update && apt-get install -y libudev-dev pkg-config
 # build deps only first for build cache
 WORKDIR /usr/src
 RUN USER=root cargo new chimemon
@@ -11,6 +11,7 @@ COPY . .
 RUN cargo build --release

 FROM debian:bullseye-slim
+RUN apt-get update && apt-get install -y libudev
 WORKDIR /app
 COPY --from=builder /usr/src/chimemon/target/release/chimemon chimemon
 CMD ["/app/chimemon"]
--- a/src/config.rs
+++ b/src/config.rs
@@ -0,0 +1,214 @@
+use figment::{Provider, providers::Serialized, util::map, value::Map};
+use gethostname::gethostname;
+use serde_derive::{Deserialize, Serialize};
+use serde_with::{DurationSeconds, serde_as};
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct InfluxConfig {
+    pub enabled: bool,
+    pub timeout: std::time::Duration,
+    pub url: String,
+    pub org: String,
+    pub bucket: String,
+    pub token: String,
+    pub tags: Map<String, String>,
+}
+
+impl Default for InfluxConfig {
+    fn default() -> Self {
+        let host = gethostname().into_string().unwrap();
+        InfluxConfig {
+            enabled: false,
+            timeout: std::time::Duration::from_secs(10),
+            url: "http://localhost:8086".into(),
+            org: "default".into(),
+            bucket: "default".into(),
+            token: "".into(),
+            tags: map! { "host".into() => host },
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct ChronyConfig {
+    pub enabled: bool,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub timeout: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub tracking_interval: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub sources_interval: std::time::Duration,
+    pub measurement_prefix: String,
+    pub tracking_measurement: String,
+    pub sources_measurement: String,
+    pub host: String,
+}
+
+impl Default for ChronyConfig {
+    fn default() -> Self {
+        ChronyConfig {
+            enabled: false,
+            timeout: std::time::Duration::from_secs(5),
+            tracking_interval: std::time::Duration::from_secs(60),
+            sources_interval: std::time::Duration::from_secs(300),
+            measurement_prefix: "chrony.".into(),
+            tracking_measurement: "tracking".into(),
+            sources_measurement: "sources".into(),
+            host: "127.0.0.1:323".into(),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct ChronySockConfig {
+    pub enabled: bool,
+    pub sock: String,
+}
+
+impl Default for ChronySockConfig {
+    fn default() -> Self {
+        ChronySockConfig {
+            enabled: false,
+            sock: "".into(),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct HwmonSensorConfig {
+    pub device: String,
+    pub sensor: String,
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone, Debug)]
+#[serde(default)]
+pub struct HwmonConfig {
+    pub enabled: bool,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub interval: std::time::Duration,
+    pub measurement: String,
+    pub sensors: Map<String, HwmonSensorConfig>,
+}
+
+impl Default for HwmonConfig {
+    fn default() -> Self {
+        HwmonConfig {
+            enabled: false,
+            interval: std::time::Duration::from_secs(60),
+            measurement: "hwmon".into(),
+            sensors: map! {},
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone, Debug)]
+#[serde(default)]
+pub struct GpsdConfig {
+    pub enabled: bool,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub interval: std::time::Duration,
+    pub host: String,
+}
+
+impl Default for GpsdConfig {
+    fn default() -> Self {
+        GpsdConfig {
+            enabled: false,
+            interval: std::time::Duration::from_secs(60),
+            host: "localhost:2947".into(),
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone, Debug)]
+#[serde(default)]
+pub struct Prs10Config {
+    pub enabled: bool,
+    pub port: String,
+    pub baud: u32,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub timeout: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub status_interval: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub stats_interval: std::time::Duration,
+}
+
+impl Default for Prs10Config {
+    fn default() -> Self {
+        Prs10Config {
+            enabled: false,
+            port: "/dev/ttyS0".into(),
+            baud: 9600,
+            timeout: std::time::Duration::from_secs(1),
+            status_interval: std::time::Duration::from_secs(10),
+            stats_interval: std::time::Duration::from_secs(30),
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct UCCMConfig {
+    pub enabled: bool,
+    pub port: String,
+    pub baud: u32,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub status_interval: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub timeout: std::time::Duration,
+    pub measurement: String,
+}
+
+impl Default for UCCMConfig {
+    fn default() -> Self {
+        UCCMConfig {
+            enabled: false,
+            port: "/dev/ttyS0".into(),
+            baud: 57600,
+            status_interval: std::time::Duration::from_secs(10),
+            timeout: std::time::Duration::from_secs(1),
+            measurement: "uccm_gpsdo".into(),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum SourceConfig {
+    Chrony(ChronyConfig),
+    Hwmon(HwmonConfig),
+    Uccm(UCCMConfig),
+    Gpsd(GpsdConfig),
+    Prs10(Prs10Config),
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum TargetConfig {
+    ChronySock(ChronySockConfig),
+    Influxdb(InfluxConfig),
+}
+
+#[derive(Serialize, Deserialize, Clone, Default)]
+pub struct Config {
+    pub sources: Map<String, SourceConfig>,
+    pub targets: Map<String, TargetConfig>,
+}
+
+impl Provider for Config {
+    fn metadata(&self) -> figment::Metadata {
+        figment::Metadata::named("Default config")
+    }
+    fn data(&self) -> Result<Map<figment::Profile, figment::value::Dict>, figment::Error> {
+        Serialized::defaults(Config::default()).data()
+    }
+}
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,151 +1,22 @@
+pub mod config;
 pub mod sources;
 pub mod targets;

+#[macro_export]
+macro_rules! fatal {
+    ($($arg:tt)*) => {{
+        tracing::error!($($arg)*);
+        std::process::exit(-1);
+    }};
+}
+
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
-use figment::{
-    Figment,
-    providers::{Format, Serialized, Toml},
-    util::map,
-    value::Map,
-};
-use gethostname::gethostname;
-use influxdb2::models::DataPoint;
-use serde_derive::{Deserialize, Serialize};
-use tokio::sync::broadcast::*;

-use std::{fmt::Debug, path::Path, sync::Arc};
+use tokio::sync::broadcast;
+use tokio_util::sync::CancellationToken;

-#[derive(Serialize, Deserialize, Clone)]
-pub struct InfluxConfig {
-    pub enabled: bool,
-    pub url: String,
-    pub org: String,
-    pub bucket: String,
-    pub token: String,
-    pub tags: Map<String, String>,
-}
-
-impl Default for InfluxConfig {
-    fn default() -> Self {
-        let host = gethostname().into_string().unwrap();
-        InfluxConfig {
-            enabled: false,
-            url: "http://localhost:8086".into(),
-            org: "default".into(),
-            bucket: "default".into(),
-            token: "".into(),
-            tags: map! { "host".into() => host },
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct ChronyConfig {
-    pub enabled: bool,
-    pub timeout: u64,
-    pub tracking_interval: u64,
-    pub sources_interval: u64,
-    pub measurement_prefix: String,
-    pub tracking_measurement: String,
-    pub sources_measurement: String,
-    pub host: String,
-}
-
-impl Default for ChronyConfig {
-    fn default() -> Self {
-        ChronyConfig {
-            enabled: false,
-            timeout: 5,
-            tracking_interval: 60,
-            sources_interval: 300,
-            measurement_prefix: "chrony.".into(),
-            tracking_measurement: "tracking".into(),
-            sources_measurement: "sources".into(),
-            host: "127.0.0.1:323".into(),
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct ChronySockConfig {
-    pub enabled: bool,
-    pub sock: String,
-}
-
-impl Default for ChronySockConfig {
-    fn default() -> Self {
-        ChronySockConfig {
-            enabled: false,
-            sock: "".into(),
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct HwmonSensorConfig {
-    pub name: String,
-    pub sensor: String,
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct HwmonConfig {
-    pub enabled: bool,
-    pub interval: u64,
-    pub measurement: String,
-    pub sensors: Map<String, HwmonSensorConfig>,
-}
-
-impl Default for HwmonConfig {
-    fn default() -> Self {
-        HwmonConfig {
-            enabled: false,
-            interval: 60,
-            measurement: "hwmon".into(),
-            sensors: map! {},
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct GpsdConfig {
-    pub enabled: bool,
-    pub interval: u64,
-    pub host: String,
-}
-
-impl Default for GpsdConfig {
-    fn default() -> Self {
-        GpsdConfig {
-            enabled: false,
-            interval: 60,
-            host: "localhost:2947".into(),
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-pub struct Prs10Config {
-    pub enabled: bool,
-    pub port: String,
-    pub baud: u32,
-    pub timeout: std::time::Duration,
-    pub status_interval: std::time::Duration,
-    pub stats_interval: std::time::Duration,
-}
-
-impl Default for Prs10Config {
-    fn default() -> Self {
-        Prs10Config {
-            enabled: false,
-            port: "/dev/ttyS0".into(),
-            baud: 9600,
-            timeout: std::time::Duration::from_secs(1),
-            status_interval: std::time::Duration::from_secs(10),
-            stats_interval: std::time::Duration::from_secs(30),
-        }
-    }
-}
+use std::{fmt::Debug, sync::Arc};

 #[derive(Clone, Debug)]
 pub struct TimeReport {
@@ -165,48 +36,53 @@ pub enum SourceStatus {
    Unknown,
 }

-#[derive(Clone, Debug)]
+#[derive(Copy, Clone, Debug)]
 pub enum MetricValue {
    Int(i64),
    Float(f64),
    Bool(bool),
 }

+type MetricTag = (&'static str, String);
+type MetricTags = Vec<MetricTag>;
+
 #[derive(Clone, Debug)]
 pub struct SourceMetric {
-    name: String,
+    name: &'static str,
    value: MetricValue,
-    tags: Arc<Vec<(String, String)>>,
 }

 impl SourceMetric {
-    pub fn new_int(name: &str, value: i64, tags: Arc<Vec<(String, String)>>) -> Self {
+    pub fn new_int(name: &'static str, value: i64) -> Self {
        Self {
-            name: name.to_owned(),
+            name: name,
            value: MetricValue::Int(value),
-            tags,
        }
    }

-    pub fn new_float(name: &str, value: f64, tags: Arc<Vec<(String, String)>>) -> Self {
+    pub fn new_float(name: &'static str, value: f64) -> Self {
        Self {
-            name: name.to_owned(),
+            name: name,
            value: MetricValue::Float(value),
-            tags,
        }
    }

-    pub fn new_bool(name: &str, value: bool, tags: Arc<Vec<(String, String)>>) -> Self {
+    pub fn new_bool(name: &'static str, value: bool) -> Self {
        Self {
-            name: name.to_owned(),
+            name: name,
            value: MetricValue::Bool(value),
-            tags,
        }
    }
 }

+#[derive(Debug)]
+pub struct SourceMetricSet {
+    metrics: Vec<SourceMetric>,
+    tags: Arc<MetricTags>,
+}
+
 pub trait SourceReportDetails: Debug + Send + Sync {
-    fn to_metrics(&self) -> Vec<SourceMetric>;
+    fn to_metrics(&self) -> Vec<SourceMetricSet>;
    fn is_healthy(&self) -> bool;
 }

@@ -214,76 +90,15 @@ pub trait SourceReportDetails: Debug + Send + Sync {
 pub struct SourceReport {
    pub name: String,
    pub status: SourceStatus,
-    pub details: Arc<dyn SourceReportDetails>,
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct UCCMConfig {
-    pub enabled: bool,
-    pub port: String,
-    pub baud: u32,
-    pub status_interval: std::time::Duration,
-    pub timeout: std::time::Duration,
-    pub measurement: String,
-}
-
-impl Default for UCCMConfig {
-    fn default() -> Self {
-        UCCMConfig {
-            enabled: false,
-            port: "/dev/ttyS0".into(),
-            baud: 57600,
-            status_interval: std::time::Duration::from_secs(10),
-            timeout: std::time::Duration::from_secs(1),
-            measurement: "uccm_gpsdo".into(),
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone, Default)]
-pub struct SourcesConfig {
-    pub chrony: ChronyConfig,
-    pub hwmon: HwmonConfig,
-    pub uccm: UCCMConfig,
-    pub gpsd: GpsdConfig,
-    pub prs10: Prs10Config,
-}
-
-#[derive(Serialize, Deserialize, Clone, Default)]
-pub struct TargetsConfig {
-    pub chrony: ChronySockConfig,
-}
-
-#[derive(Serialize, Deserialize, Clone, Default)]
-pub struct Config {
-    pub influxdb: InfluxConfig,
-    pub sources: SourcesConfig,
-    pub targets: TargetsConfig,
-}
-
-pub fn load_config(filename: &Path) -> Figment {
-    Figment::from(Serialized::defaults(Config::default())).merge(Toml::file(filename))
+    pub details: Arc<dyn SourceReportDetails + Send + Sync>,
 }

 #[derive(Debug, Clone)]
 pub enum ChimemonMessage {
-    DataPoint(DataPoint),
-    DataPoints(Vec<DataPoint>),
    TimeReport(TimeReport),
    SourceReport(SourceReport),
 }

-impl From<DataPoint> for ChimemonMessage {
-    fn from(dp: DataPoint) -> Self {
-        ChimemonMessage::DataPoint(dp)
-    }
-}
-impl From<Vec<DataPoint>> for ChimemonMessage {
-    fn from(dps: Vec<DataPoint>) -> Self {
-        ChimemonMessage::DataPoints(dps)
-    }
-}
-
 impl From<TimeReport> for ChimemonMessage {
    fn from(tr: TimeReport) -> Self {
        ChimemonMessage::TimeReport(tr)
@@ -296,15 +111,21 @@ impl From<SourceReport> for ChimemonMessage {
    }
 }

-pub type ChimemonSourceChannel = Sender<ChimemonMessage>;
-pub type ChimemonTargetChannel = Receiver<ChimemonMessage>;
+pub type ChimemonSourceChannel = broadcast::Sender<ChimemonMessage>;
+pub type ChimemonTargetChannel = broadcast::Receiver<ChimemonMessage>;

 #[async_trait]
 pub trait ChimemonSource {
-    async fn run(self, chan: ChimemonSourceChannel);
+    type Config;
+    const TASK_NAME: &'static str;
+    fn new(name: &str, config: Self::Config) -> Self;
+    async fn run(self, chan: ChimemonSourceChannel, cancel: CancellationToken);
 }

 #[async_trait]
 pub trait ChimemonTarget {
-    async fn run(self, chan: ChimemonTargetChannel);
+    type Config;
+    const TASK_NAME: &'static str;
+    fn new(name: &str, config: Self::Config) -> Self;
+    async fn run(self, mut chan: ChimemonTargetChannel, cancel: CancellationToken);
 }
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,15 +1,23 @@
-use clap::{Parser, ValueEnum};
-use futures::future::join_all;
-use std::path::Path;
-use tokio::sync::broadcast;
-use tracing::{Instrument, debug, error, info, info_span, warn};
-use tracing_subscriber::{
-    self, EnvFilter,
-    fmt::format::{self, FmtSpan},
-    prelude::*,
-};
+use std::sync::Arc;

-use chimemon::*;
+use async_trait::async_trait;
+use clap::{Parser, ValueEnum};
+use figment::{
+    Figment,
+    providers::{Format, Toml},
+};
+use futures::future::join_all;
+use tokio::{select, sync::broadcast, task::JoinHandle};
+use tokio_util::sync::CancellationToken;
+use tracing::{Instrument, debug, error, info, info_span, warn};
+use tracing_subscriber::{self, EnvFilter, fmt::format::FmtSpan};
+
+use chimemon::{
+    config::{SourceConfig, TargetConfig},
+    targets::influx::InfluxTarget,
+    *,
+};
+use config::Config;
 use sources::{
    chrony::ChronyClient, gpsd::GpsdSource, hwmon::HwmonSource, prs10::Prs10Monitor,
    uccm::UCCMMonitor,
@@ -27,171 +35,211 @@ enum Level {
    Error,
 }

+impl From<Level> for tracing::Level {
+    fn from(level: Level) -> Self {
+        match level {
+            Level::Debug => tracing::Level::DEBUG,
+            Level::Info => tracing::Level::INFO,
+            Level::Warn => tracing::Level::WARN,
+            Level::Error => tracing::Level::ERROR,
+        }
+    }
+}
+
+impl Level {
+    fn level(self) -> tracing::Level {
+        self.into()
+    }
+}
+
 #[derive(Parser)]
 struct Args {
    /// TOML configuration to load
    #[arg(short, long, default_value_t = String::from("config.toml"))]
    config_file: String,
-    #[arg(value_enum, default_value_t = Level::Warn)]
+    #[arg(value_enum, default_value_t = Level::Info)]
    log_level: Level,
+    #[arg(short, long, default_value_t = false)]
+    echo_task: bool,
 }

-#[tokio::main(flavor = "current_thread")]
+fn run_source(
+    name: &str,
+    source: SourceConfig,
+    chan: ChimemonSourceChannel,
+    cancel: CancellationToken,
+) -> Option<JoinHandle<()>> {
+    match source {
+        SourceConfig::Chrony(cfg) if cfg.enabled => {
+            spawn_source::<ChronyClient>(name, cfg, chan, cancel)
+        }
+        SourceConfig::Gpsd(cfg) if cfg.enabled => {
+            spawn_source::<GpsdSource>(name, cfg, chan, cancel)
+        }
+        SourceConfig::Hwmon(cfg) if cfg.enabled => {
+            spawn_source::<HwmonSource>(name, cfg, chan, cancel)
+        }
+        SourceConfig::Prs10(cfg) if cfg.enabled => {
+            spawn_source::<Prs10Monitor>(name, cfg, chan, cancel)
+        }
+        SourceConfig::Uccm(cfg) if cfg.enabled => {
+            spawn_source::<UCCMMonitor>(name, cfg, chan, cancel)
+        }
+        _ => {
+            debug!("Disabled source {name} skipped");
+            None
+        }
+    }
+}
+
+fn run_target(
+    name: &str,
+    target: TargetConfig,
+    chan: ChimemonTargetChannel,
+    cancel: CancellationToken,
+) -> Option<JoinHandle<()>> {
+    match target {
+        TargetConfig::ChronySock(cfg) if cfg.enabled => {
+            spawn_target::<ChronySockServer>(name, cfg, chan, cancel)
+        }
+        TargetConfig::Influxdb(cfg) if cfg.enabled => {
+            spawn_target::<InfluxTarget>(name, cfg, chan, cancel)
+        }
+        _ => {
+            debug!("Disabled target {name} skipped");
+            None
+        }
+    }
+}
+
+fn spawn_source<T: ChimemonSource + Send + Sync + 'static>(
+    name: &str,
+    config: T::Config,
+    chan: ChimemonSourceChannel,
+    cancel: CancellationToken,
+) -> Option<JoinHandle<()>> {
+    let span = info_span!("source", task = name);
+    let s = T::new(name, config);
+    Some(tokio::spawn(s.run(chan, cancel).instrument(span)))
+}
+
+fn spawn_target<T: ChimemonTarget + Send + Sync + 'static>(
+    name: &str,
+    config: T::Config,
+    chan: ChimemonTargetChannel,
+    cancel: CancellationToken,
+) -> Option<JoinHandle<()>> {
+    let span = info_span!("target", task = name);
+    let t = T::new(name, config);
+    Some(tokio::spawn(async move {
+        t.run(chan, cancel).instrument(span).await
+    }))
+}
+
+struct EchoTarget {}
+struct EchoTargetConfig {}
+#[async_trait]
+impl ChimemonTarget for EchoTarget {
+    type Config = EchoTargetConfig;
+    const TASK_NAME: &'static str = "echo-task";
+
+    fn new(_name: &str, _config: Self::Config) -> Self {
+        EchoTarget {}
+    }
+    async fn run(self, mut chan: ChimemonTargetChannel, cancel: CancellationToken) {
+        info!("Dummy receiver task started");
+        loop {
+            let msg = select! {
+                _ = cancel.cancelled() => {
+                    return
+                },
+                msg = chan.recv() => msg
+            };
+            match msg {
+                Ok(ChimemonMessage::SourceReport(report)) => {
+                    let metrics = report.details.to_metrics();
+                    info!("instance: {} metrics: {metrics:?}", report.name);
+                }
+                Ok(msg) => {
+                    info!("message: {msg:?}");
+                }
+                Err(e) => {
+                    warn!(error = ?e, "Error receiving message");
+                }
+            }
+        }
+    }
+}
+
+#[tokio::main(flavor = "multi_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args = Args::parse();
    tracing_subscriber::fmt()
-        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::default()))
+        .with_env_filter(
+            EnvFilter::try_from_default_env()
+                .unwrap_or(EnvFilter::default().add_directive(args.log_level.level().into())),
+        )
        // .event_format(format::Format::default().pretty())
        .with_span_events(FmtSpan::CLOSE)
        .init();

-    let args = Args::parse();
-
    info!("{PROGRAM_NAME} v{VERSION} starting...");
-    let fig = load_config(Path::new(&args.config_file));
+    let fig = Figment::new()
+        .merge(Config::default())
+        .merge(Toml::file(&args.config_file));
    debug!("{fig:?}");
    let config: Config = fig.extract()?;

    let mut tasks = Vec::new();
-    let (tx, _) = broadcast::channel(16);
-    let sourcechan: ChimemonSourceChannel = tx;
+    let (sourcechan, _) = broadcast::channel(16);

-    if config.influxdb.enabled {
-        info!(
-            "Connecting to influxdb {} org: {} using token",
-            &config.influxdb.url, &config.influxdb.org
-        );
-        let config = config.clone();
-        let influx = influxdb2::Client::new(
-            &config.influxdb.url,
-            &config.influxdb.org,
-            &config.influxdb.token,
-        );
+    let shutdown_token = CancellationToken::new();

-        let mut influxrx = sourcechan.subscribe();
-
-        tasks.push(tokio::spawn(
-            async move {
-                let stream = async_stream::stream! {
-                    while let Ok(msg) = influxrx.recv().await {
-                        match msg { ChimemonMessage::DataPoint(dp) => {
-                            yield dp
-                        }, ChimemonMessage::DataPoints(dps) => {
-                            for p in dps {
-                                yield p
+    for (name, target) in config.targets {
+        if let Some(task) = run_target(
+            &name,
+            target,
+            sourcechan.subscribe(),
+            shutdown_token.clone(),
+        ) {
+            tasks.push(task)
        }
-                        }, _ => {}
-                    }            }
-                };
-                influx.write(&config.influxdb.bucket, stream).await.unwrap();
-            }
-            .instrument(info_span!("influx-task")),
-        ));
    }

-    let chrony = if config.sources.chrony.enabled {
-        Some(ChronyClient::new(config.to_owned()))
-    } else {
-        None
-    };
-    if let Some(c) = chrony {
-        tasks.push(tokio::spawn(
-            c.run(sourcechan.clone())
-                .instrument(info_span!("chrony-task")),
-        ));
-    };
-
-    let hwmon = if config.sources.hwmon.enabled {
-        Some(HwmonSource::new(config.to_owned()))
-    } else {
-        None
-    };
-    if let Some(hwmon) = hwmon {
-        tasks.push(tokio::spawn(
-            hwmon
-                .run(sourcechan.clone())
-                .instrument(info_span!("hwmon-task")),
-        ));
-    };
-
-    let uccm = if config.sources.uccm.enabled {
-        Some(UCCMMonitor::new(config.to_owned()))
-    } else {
-        None
-    };
-    if let Some(uccm) = uccm {
-        tasks.push(tokio::spawn(
-            uccm.run(sourcechan.clone())
-                .instrument(info_span!("uccm-task")),
-        ));
-    };
-
-    let gpsd = if config.sources.gpsd.enabled {
-        Some(GpsdSource::new(config.to_owned()).await.unwrap())
-    } else {
-        None
-    };
-    if let Some(gpsd) = gpsd {
-        tasks.push(tokio::spawn(
-            gpsd.run(sourcechan.clone())
-                .instrument(info_span!("gpsd-task")),
-        ))
+    for (name, source) in config.sources {
+        if let Some(task) = run_source(&name, source, sourcechan.clone(), shutdown_token.clone()) {
+            tasks.push(task)
        }
-
-    let prs10 = if config.sources.prs10.enabled {
-        Some(Prs10Monitor::new(config.sources.prs10))
-    } else {
-        None
-    };
-    if let Some(prs10) = prs10 {
-        tasks.push(tokio::spawn(
-            prs10
-                .run(sourcechan.clone())
-                .instrument(info_span!("prs10-task")),
-        ))
    }

-    let chrony_refclock = if config.targets.chrony.enabled {
-        Some(ChronySockServer::new(config.targets.chrony.to_owned()))
-    } else {
-        None
-    };
-    if let Some(chrony_refclock) = chrony_refclock {
-        tasks.push(tokio::spawn(
-            chrony_refclock
-                .run(sourcechan.subscribe())
-                .instrument(info_span!("chrony-refclock-task")),
-        ));
-    };
-
    if tasks.len() == 0 {
        error!("No tasks configured, exiting.");
        return Ok(()); // not an error, but exit before starting a dummy task
    }
-
+    if sourcechan.strong_count() == 0 {
+        warn!("No sources configured, no events will be generated");
+    }
    if sourcechan.receiver_count() == 0 {
-        warn!("No consumers configured, events will be discarded");
-        let mut chan = sourcechan.subscribe();
-        // spawn a dummy task to reap the channel and keep the process alive
-        tasks.push(tokio::spawn(
-            async move {
-                loop {
-                    while let Ok(m) = chan.recv().await {
-                        info!("received {m:?}");
-                        match m {
-                            ChimemonMessage::SourceReport(report) => {
-                                let metrics = report.details.to_metrics();
-                                info!("metrics: {metrics:?}");
+        warn!("No targets configured, events will be discarded");
    }
-                            _ => {}
-                        }
-                    }
-                }
-            }
-            .instrument(info_span!("dummy-receiver-task")),
-        ))
+    if args.echo_task || sourcechan.receiver_count() == 0 {
+        let c = EchoTargetConfig {};
+        tasks.push(
+            spawn_target::<EchoTarget>("echo", c, sourcechan.subscribe(), shutdown_token.clone())
+                .unwrap(),
+        )
    }

    debug!("Task setup complete, tasks: {}", tasks.len());
+    ctrlc::set_handler(move || {
+        if shutdown_token.is_cancelled() {
+            info!("Forced shutdown");
+            std::process::exit(1);
+        }
+        info!("Shutting down");
+        shutdown_token.cancel()
+    })
+    .unwrap();

    join_all(tasks).await;

--- a/src/sources/chrony.rs
+++ b/src/sources/chrony.rs
@@ -1,26 +1,30 @@
-use crate::{
-    ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport, SourceReportDetails,
-    SourceStatus,
-};
+use std::net::{SocketAddr, ToSocketAddrs};
+use std::sync::Arc;
+
 use async_trait::async_trait;
 use chrony_candm::reply::{self, ReplyBody, SourceMode};
 use chrony_candm::request::{self, RequestBody};
 use chrony_candm::{ClientOptions, blocking_query};
-use std::net::{SocketAddr, ToSocketAddrs};
-use std::sync::Arc;
-use std::time::Duration;
-use tokio::join;
+use tokio::select;
+use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};

+use crate::SourceMetricSet;
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, MetricTags, SourceMetric, SourceReport,
+    SourceReportDetails, SourceStatus, config::ChronyConfig,
+};
+
 pub struct ChronyClient {
    pub server: SocketAddr,
+    pub name: String,
    client_options: ClientOptions,
-    config: Config,
+    config: ChronyConfig,
 }

 #[derive(Debug)]
 pub struct ChronyTrackingReport {
-    tags: Arc<Vec<(String, String)>>,
+    tags: Arc<MetricTags>,
    pub ref_id: i64,
    pub ref_ip_addr: String,
    pub stratum: i64,
@@ -40,26 +44,24 @@ impl SourceReportDetails for ChronyTrackingReport {
    fn is_healthy(&self) -> bool {
        true
    }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
-        let tags = &self.tags;
-        vec![
-            SourceMetric::new_int("ref_id", self.ref_id, tags.clone()),
-            SourceMetric::new_int("stratum", self.stratum, tags.clone()),
-            SourceMetric::new_int("leap_status", self.leap_status, tags.clone()),
-            SourceMetric::new_float("current_correction", self.current_correction, tags.clone()),
-            SourceMetric::new_float("last_offset", self.last_offset, tags.clone()),
-            SourceMetric::new_float("rms_offset", self.rms_offset, tags.clone()),
-            SourceMetric::new_float("freq_ppm", self.freq_ppm, tags.clone()),
-            SourceMetric::new_float("resid_freq_ppm", self.resid_freq_ppm, tags.clone()),
-            SourceMetric::new_float("skew_ppm", self.skew_ppm, tags.clone()),
-            SourceMetric::new_float("root_delay", self.root_delay, tags.clone()),
-            SourceMetric::new_float("root_dispersion", self.root_dispersion, tags.clone()),
-            SourceMetric::new_float(
-                "last_update_interval",
-                self.last_update_interval,
-                tags.clone(),
-            ),
-        ]
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        vec![SourceMetricSet {
+            tags: self.tags.clone(),
+            metrics: vec![
+                SourceMetric::new_int("ref_id", self.ref_id),
+                SourceMetric::new_int("stratum", self.stratum),
+                SourceMetric::new_int("leap_status", self.leap_status),
+                SourceMetric::new_float("current_correction", self.current_correction),
+                SourceMetric::new_float("last_offset", self.last_offset),
+                SourceMetric::new_float("rms_offset", self.rms_offset),
+                SourceMetric::new_float("freq_ppm", self.freq_ppm),
+                SourceMetric::new_float("resid_freq_ppm", self.resid_freq_ppm),
+                SourceMetric::new_float("skew_ppm", self.skew_ppm),
+                SourceMetric::new_float("root_delay", self.root_delay),
+                SourceMetric::new_float("root_dispersion", self.root_dispersion),
+                SourceMetric::new_float("last_update_interval", self.last_update_interval),
+            ],
+        }]
    }
 }

@@ -73,14 +75,14 @@ impl SourceReportDetails for ChronySourcesReport {
        //TODO: think about whether there is an idea of unhealthy sources
        true
    }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
-        let mut metrics = Vec::with_capacity(8 * self.sources.len());
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let mut metrics = Vec::with_capacity(self.sources.len());

        for source in &self.sources {
            let tags = Arc::new(vec![
-                ("ref_id".to_owned(), source.ip_addr.to_string()),
+                ("ref_id", source.ip_addr.to_string()),
                (
-                    "mode".to_owned(),
+                    "mode",
                    match source.mode {
                        SourceMode::Client => String::from("server"),
                        SourceMode::Peer => String::from("peer"),
@@ -88,7 +90,7 @@ impl SourceReportDetails for ChronySourcesReport {
                    },
                ),
                (
-                    "state".to_owned(),
+                    "state",
                    match source.state {
                        reply::SourceState::Selected => String::from("best"),
                        reply::SourceState::NonSelectable => String::from("unusable"),
@@ -99,28 +101,19 @@ impl SourceReportDetails for ChronySourcesReport {
                    },
                ),
            ]);
-            metrics.extend([
-                SourceMetric::new_int("poll", source.poll as i64, tags.clone()),
-                SourceMetric::new_int("stratum", source.stratum as i64, tags.clone()),
-                SourceMetric::new_int("flags", source.flags.bits() as i64, tags.clone()),
-                SourceMetric::new_int(
-                    "reachability",
-                    source.reachability.count_ones() as i64,
-                    tags.clone(),
-                ),
-                SourceMetric::new_int("since_sample", source.since_sample as i64, tags.clone()),
-                SourceMetric::new_float(
-                    "orig_latest_meas",
-                    source.orig_latest_meas.into(),
-                    tags.clone(),
-                ),
-                SourceMetric::new_float("latest_meas", source.latest_meas.into(), tags.clone()),
-                SourceMetric::new_float(
-                    "latest_meas_err",
-                    source.latest_meas_err.into(),
-                    tags.clone(),
-                ),
-            ]);
+            metrics.push(SourceMetricSet {
+                tags: tags,
+                metrics: vec![
+                    SourceMetric::new_int("poll", source.poll as i64),
+                    SourceMetric::new_int("stratum", source.stratum as i64),
+                    SourceMetric::new_int("flags", source.flags.bits() as i64),
+                    SourceMetric::new_int("reachability", source.reachability.count_ones() as i64),
+                    SourceMetric::new_int("since_sample", source.since_sample as i64),
+                    SourceMetric::new_float("orig_latest_meas", source.orig_latest_meas.into()),
+                    SourceMetric::new_float("latest_meas", source.latest_meas.into()),
+                    SourceMetric::new_float("latest_meas_err", source.latest_meas_err.into()),
+                ],
+            });
        }

        metrics
@@ -129,7 +122,7 @@ impl SourceReportDetails for ChronySourcesReport {

 fn report_from_tracking(
    t: &reply::Tracking,
-    config: &Config,
+    config: &ChronyConfig,
 ) -> Result<ChronyTrackingReport, Box<dyn std::error::Error>> {
    let report = ChronyTrackingReport {
        tags: Arc::new(vec![]), //TODO: allow configuring tags in the source
@@ -151,25 +144,6 @@ fn report_from_tracking(
 }

 impl ChronyClient {
-    pub fn new(config: Config) -> Self {
-        let server = config
-            .sources
-            .chrony
-            .host
-            .to_socket_addrs()
-            .unwrap()
-            .next()
-            .expect("Unable to parse host:port:");
-        let client_options = ClientOptions {
-            n_tries: 3,
-            timeout: Duration::from_secs(config.sources.chrony.timeout),
-        };
-        ChronyClient {
-            server,
-            client_options,
-            config,
-        }
-    }
    async fn query(&self, request: RequestBody) -> Result<reply::Reply, std::io::Error> {
        let server = self.server;
        let client_options = self.client_options;
@@ -265,7 +239,7 @@ impl ChronyClient {

        let tracking_data = report_from_tracking(&tracking, &self.config)?;
        let report = SourceReport {
-            name: "chrony-tracking".to_owned(),
+            name: self.name.clone(),
            status: SourceStatus::Unknown,
            details: Arc::new(tracking_data),
        };
@@ -283,7 +257,7 @@ impl ChronyClient {
        let sources = self.get_sources().await?;
        let details = ChronySourcesReport { sources };
        let report = SourceReport {
-            name: "chrony-sources".to_owned(),
+            name: self.name.clone(),
            status: SourceStatus::Unknown,
            details: Arc::new(details),
        };
@@ -295,42 +269,53 @@ impl ChronyClient {

 #[async_trait]
 impl ChimemonSource for ChronyClient {
-    async fn run(self, chan: ChimemonSourceChannel) {
+    type Config = ChronyConfig;
+    const TASK_NAME: &'static str = "chrony-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let server = config
+            .host
+            .to_socket_addrs()
+            .unwrap()
+            .next()
+            .expect("Unable to parse host:port:");
+        let client_options = ClientOptions {
+            n_tries: 3,
+            timeout: config.timeout,
+        };
+        ChronyClient {
+            name: name.to_owned(),
+            server,
+            client_options,
+            config,
+        }
+    }
+    async fn run(self, chan: ChimemonSourceChannel, cancel: CancellationToken) {
        info!("Chrony task started");

-        let mut t_interval = tokio::time::interval(Duration::from_secs(
-            self.config.sources.chrony.tracking_interval,
-        ));
-        let mut s_interval = tokio::time::interval(Duration::from_secs(
-            self.config.sources.chrony.sources_interval,
-        ));
-
-        let t_future = async {
-            let lchan = chan.clone();
+        let mut t_interval = tokio::time::interval(self.config.tracking_interval);
+        let mut s_interval = tokio::time::interval(self.config.sources_interval);
        loop {
-                t_interval.tick().await;
-
-                match self.tracking_poll(&lchan).await {
+            select! {
+                _ = cancel.cancelled() => {
+                    return
+                },
+                _ = t_interval.tick() => {
+                    match self.tracking_poll(&chan).await {
                        Ok(_) => (),
                        Err(e) => {
-                        warn!("Error in chrony task: {}", e.to_string());
+                            warn!(error = ?e, "Error in chrony tracking task");
                        }
                    }
-            }
-        };
-        let s_future = async {
-            let lchan = chan.clone();
-            loop {
-                s_interval.tick().await;
-
-                match self.sources_poll(&lchan).await {
+                },
+                _ = s_interval.tick() => {
+                    match self.sources_poll(&chan).await {
                        Ok(_) => (),
                        Err(e) => {
-                        warn!("Error in chrony task: {}", e.to_string());
+                            warn!(error = ?e, "Error in chrony sources task");
+                        }
+                    }
                }
            }
        }
-        };
-        join!(t_future, s_future);
    }
 }
--- a/src/sources/gpsd.rs
+++ b/src/sources/gpsd.rs
@@ -1,8 +1,3 @@
-use crate::{
-    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport,
-    SourceReportDetails, SourceStatus,
-};
-
 use std::collections::HashMap;
 use std::f64;
 use std::fmt::Debug;
@@ -12,18 +7,25 @@ use std::time::Duration;

 use async_trait::async_trait;
 use backoff::ExponentialBackoff;
-use futures::StreamExt;
-use futures::{SinkExt, Stream};
+use futures::{SinkExt, Stream, StreamExt};
 use gpsd_proto::{Device, Gst, Mode, Pps, Sky, Tpv, UnifiedResponse, Version};
 use serde::Serialize;
 use serde_json;
 use tokio::net::{TcpStream, ToSocketAddrs, lookup_host};
 use tokio::time::{interval, timeout};
 use tokio_util::codec::{Framed, LinesCodec};
-use tracing::{debug, debug_span, info, instrument, warn};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, debug_span, error, info, instrument, warn};
+
+use crate::SourceMetricSet;
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, SourceMetric, SourceReport, SourceReportDetails,
+    SourceStatus, config::GpsdConfig,
+};

 pub struct GpsdSource {
-    pub config: Config,
+    pub name: String,
+    pub config: GpsdConfig,
    conn: GpsdTransport,
    devices: HashMap<String, Device>,
    last_gst: Option<Gst>,
@@ -75,20 +77,24 @@ impl SourceReportDetails for GpsdSourceReport {
    fn is_healthy(&self) -> bool {
        self.fix_type != GpsdFixType::Unknown && self.fix_type != GpsdFixType::NoFix
    }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
-        let no_tags = Arc::new(vec![]);
-        vec![
-            SourceMetric::new_int("sats_visible", self.sats_visible as i64, no_tags.clone()),
-            SourceMetric::new_int("sats_tracked", self.sats_tracked as i64, no_tags.clone()),
-            SourceMetric::new_float("tdop", self.tdop, no_tags.clone()),
-        ]
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let tags = Arc::new(vec![]);
+        vec![SourceMetricSet {
+            tags,
+            metrics: vec![
+                SourceMetric::new_int("sats_visible", self.sats_visible as i64),
+                SourceMetric::new_int("sats_tracked", self.sats_tracked as i64),
+                SourceMetric::new_float("tdop", self.tdop),
+            ],
+        }]
    }
 }

 impl GpsdSource {
-    pub async fn new(config: Config) -> Result<Self, std::io::Error> {
-        let conn = GpsdTransport::new(&config.sources.gpsd.host).await?;
+    async fn inner_new(name: &str, config: GpsdConfig) -> Result<Self, std::io::Error> {
+        let conn = GpsdTransport::new(&config.host).await?;
        Ok(Self {
+            name: name.to_owned(),
            config,
            conn,
            devices: HashMap::new(),
@@ -98,9 +104,6 @@ impl GpsdSource {
            last_sky: None,
        })
    }
-}
-
-impl GpsdSource {
    async fn send_status(&self, chan: &mut ChimemonSourceChannel) {
        let sky = self.last_sky.as_ref();
        let tpv = self.last_tpv.as_ref();
@@ -117,8 +120,9 @@ impl GpsdSource {
            .and_then(|sky| sky.tdop)
            .map_or(f64::INFINITY, |tdop| tdop as f64);

-        chan.send(ChimemonMessage::SourceReport(SourceReport {
-            name: "gpsd".into(),
+        if let Err(e) = chan.send(
+            SourceReport {
+                name: self.name.clone(),
                status: SourceStatus::Unknown,
                details: Arc::new(GpsdSourceReport {
                    fix_type: tpv.map_or(GpsdFixType::Unknown, |tpv| tpv.mode.into()),
@@ -126,8 +130,11 @@ impl GpsdSource {
                    sats_visible,
                    tdop,
                }),
-        }))
-        .unwrap();
+            }
+            .into(),
+        ) {
+            error!(error = ?e, "Unable to send to channel")
+        }
    }

    fn handle_msg(&mut self, msg: String) -> Result<(), Box<dyn std::error::Error>> {
@@ -163,10 +170,20 @@ impl GpsdSource {

 #[async_trait]
 impl ChimemonSource for GpsdSource {
-    async fn run(mut self, mut chan: ChimemonSourceChannel) {
+    type Config = GpsdConfig;
+    const TASK_NAME: &'static str = "gpsd-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        // TODO: refactor so this mess isn't necessary
+        // Should do async setup at the start of run(), not here
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .build()
+            .unwrap();
+        rt.block_on(Self::inner_new(name, config)).unwrap()
+    }
+    async fn run(mut self, mut chan: ChimemonSourceChannel, cancel: CancellationToken) {
        info!("gpsd task started");
        self.conn.conn().await.unwrap();
-        let mut ticker = interval(Duration::from_secs(self.config.sources.gpsd.interval));
+        let mut ticker = interval(self.config.interval);

        let mut params = WatchParams::default();
        params.json = Some(true);
@@ -177,6 +194,9 @@ impl ChimemonSource for GpsdSource {
        loop {
            let framed = self.conn.framed.as_mut().expect("must be connected");
            tokio::select! {
+                _ = cancel.cancelled() => {
+                    return
+                },
                _ = ticker.tick() => {
                    self.send_status(&mut chan).await
                },
--- a/src/sources/hwmon.rs
+++ b/src/sources/hwmon.rs
@@ -1,13 +1,18 @@
-use crate::{
-    ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport, SourceReportDetails,
-    SourceStatus,
-};
+use std::{fs::File, io::Read, path::PathBuf, sync::Arc};
+
 use async_trait::async_trait;
-use std::{fs::File, io::Read, path::PathBuf, sync::Arc, time::Duration};
+use tokio::select;
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};

+use crate::{
+    ChimemonSource, ChimemonSourceChannel, MetricTags, SourceMetric, SourceMetricSet, SourceReport,
+    SourceReportDetails, SourceStatus, config::HwmonConfig,
+};
+
 pub struct HwmonSource {
-    config: Config,
+    name: String,
+    config: HwmonConfig,
    sensors: Vec<Arc<HwmonSensor>>,
 }

@@ -18,7 +23,7 @@ struct HwmonSensor {
    device: String,
    sensor: String,
    label: Option<String>,
-    tags: Arc<Vec<(String, String)>>,
+    tags: Arc<MetricTags>,
 }

 impl HwmonSensor {
@@ -40,12 +45,12 @@ impl HwmonSensor {
            None
        };
        let mut tags_vec = vec![
-            ("name".to_owned(), name.to_owned()),
-            ("device".to_owned(), device.to_owned()),
-            ("sensor".to_owned(), sensor.to_owned()),
+            ("name", name.to_owned()),
+            ("device", device.to_owned()),
+            ("sensor", sensor.to_owned()),
        ];
        if let Some(label) = &label {
-            tags_vec.push(("label".to_owned(), label.clone()))
+            tags_vec.push(("label", label.clone()))
        }
        Self {
            value_path,
@@ -68,14 +73,13 @@ impl SourceReportDetails for HwmonReport {
        //self.alarms.iter().any(|(_sensor, alarm)| *alarm)
        true
    }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
        let mut metrics = Vec::new();
        for (sensor, value) in &self.values {
-            metrics.push(SourceMetric::new_float(
-                "hwmon_value",
-                *value,
-                sensor.tags.clone(),
-            ))
+            metrics.push(SourceMetricSet {
+                tags: sensor.tags.clone(),
+                metrics: vec![SourceMetric::new_float("hwmon_value", *value)],
+            })
        }
        // for (sensor, alarm) in &self.alarms {
        //     metrics.push(SourceMetric::new_bool(
@@ -92,18 +96,6 @@ impl SourceReportDetails for HwmonReport {
 const HWMON_ROOT: &str = "/sys/class/hwmon";

 impl HwmonSource {
-    pub fn new(config: Config) -> Self {
-        let sensors = config
-            .sources
-            .hwmon
-            .sensors
-            .iter()
-            .map(|(k, v)| Arc::new(HwmonSensor::new(k, &v.name, &v.sensor)))
-            .collect();
-
-        HwmonSource { config, sensors }
-    }
-
    async fn get_raw_value(sensor: &HwmonSensor) -> Result<String, std::io::Error> {
        tokio::fs::read_to_string(&sensor.value_path).await
    }
@@ -111,15 +103,33 @@ impl HwmonSource {

 #[async_trait]
 impl ChimemonSource for HwmonSource {
-    async fn run(self, chan: ChimemonSourceChannel) {
+    type Config = HwmonConfig;
+    const TASK_NAME: &'static str = "hwmon-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let sensors = config
+            .sensors
+            .iter()
+            .map(|(k, v)| Arc::new(HwmonSensor::new(k, &v.device, &v.sensor)))
+            .collect();
+        debug!("config: {config:?}");
+        HwmonSource {
+            name: name.to_owned(),
+            config,
+            sensors,
+        }
+    }
+    async fn run(self, chan: ChimemonSourceChannel, cancel: CancellationToken) {
        info!("hwmon task started");
-        let mut interval =
-            tokio::time::interval(Duration::from_secs(self.config.sources.hwmon.interval));
+        let mut interval = tokio::time::interval(self.config.interval);
        loop {
-            interval.tick().await;
+            select! {
+                _ = cancel.cancelled() => { return; },
+                _ = interval.tick() => {
                    let mut values = Vec::new();
                    for s in &self.sensors {
-                if let Ok(sensor_val) = HwmonSource::get_raw_value(s).await {
+                        debug!("Sensor {s:?}");
+                        match HwmonSource::get_raw_value(s).await {
+                            Ok(sensor_val) => {
                            debug!(
                                "hwmon {} raw value {}",
                                s.value_path.to_string_lossy(),
@@ -133,12 +143,15 @@ impl ChimemonSource for HwmonSource {
                                    s.value_path.to_string_lossy()
                                );
                            }
-                } else {
-                    error!("Unable to get hwmon sensor value");
+                        },
+                       Err(e) => {
+                            error!("Unable to get hwmon sensor value ({})", e.to_string());
+                            continue
+                        }
                        }
                    }
                    let report = SourceReport {
-                name: "hwmon".to_owned(),
+                        name: self.name.clone(),
                        status: SourceStatus::Healthy,
                        details: Arc::new(HwmonReport { values }),
                    };
@@ -152,3 +165,5 @@ impl ChimemonSource for HwmonSource {
                }
            }
        }
+    }
+}
--- a/src/sources/prs10.rs
+++ b/src/sources/prs10.rs
@@ -1,21 +1,27 @@
+use std::any::type_name;
 use std::str::FromStr;
 use std::sync::Arc;
+use std::time::Duration;

-use crate::{
-    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Prs10Config, SourceMetric,
-    SourceReport, SourceReportDetails, SourceStatus,
-};
 use async_trait::async_trait;
 use bitflags::bitflags;
 use itertools::Itertools;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf};
+use tokio::io::{
+    AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter, ReadHalf, WriteHalf,
+};
 use tokio::select;
 use tokio::sync::OnceCell;
 use tokio::time::{interval, timeout};
-use tokio_serial;
 use tokio_serial::{SerialPort, SerialStream};
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, debug_span, error, info, instrument, warn};

+use crate::SourceMetricSet;
+use crate::{
+    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, MetricTags, SourceMetric, SourceReport,
+    SourceReportDetails, SourceStatus, config::Prs10Config, fatal,
+};
+
 #[derive(Debug)]
 pub struct Prs10Info {
    pub model: String,
@@ -53,7 +59,7 @@ bitflags! {
 }

 impl Prs10PowerLampFlags {
-    pub fn get_metrics(&self, no_tags: Arc<Vec<(String, String)>>) -> Vec<SourceMetric> {
+    pub fn get_metrics(&self, tags: Arc<MetricTags>) -> Vec<SourceMetric> {
        // Define the mapping statically
        const FLAG_LABELS: [(&Prs10PowerLampFlags, &str); 8] = [
            (&Prs10PowerLampFlags::ELEC_VOLTAGE_LOW, "elec_voltage_low"),
@@ -71,7 +77,7 @@ impl Prs10PowerLampFlags {
            .iter()
            .map(|(flag, label)| {
                // We track whether each flag is set (true) or not (false)
-                SourceMetric::new_bool(*label, self.contains(**flag), no_tags.clone())
+                SourceMetric::new_bool(*label, self.contains(**flag))
            })
            .collect()
    }
@@ -114,6 +120,7 @@ bitflags! {
        const EFC_LOW = (1<<3);
        const CAL_VOLTAGE_HIGH = (1<<4);
        const CAL_VOLTAGE_LOW = (1<<5);
+        const _ = !0;
    }
 }

@@ -178,20 +185,19 @@ impl SourceReportDetails for Prs10Status {
            && self.pps_flags == HEALTHY_PPS
    }

-    fn to_metrics(&self) -> Vec<SourceMetric> {
-        let no_tags = Arc::new(vec![]);
-        vec![
-            SourceMetric::new_int(
-                "volt_lamp_flags",
-                self.volt_lamp_flags.bits() as i64,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_int("rf_flags", self.rf_flags.bits() as i64, no_tags.clone()),
-            SourceMetric::new_int("temp_flags", self.temp_flags.bits() as i64, no_tags.clone()),
-            SourceMetric::new_int("fll_flags", self.fll_flags.bits() as i64, no_tags.clone()),
-            SourceMetric::new_int("pps_flags", self.pps_flags.bits() as i64, no_tags.clone()),
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let tags = Arc::new(vec![]);
+        vec![SourceMetricSet {
+            tags,
+            metrics: vec![
+                SourceMetric::new_int("volt_lamp_flags", self.volt_lamp_flags.bits() as i64),
+                SourceMetric::new_int("rf_flags", self.rf_flags.bits() as i64),
+                SourceMetric::new_int("temp_flags", self.temp_flags.bits() as i64),
+                SourceMetric::new_int("fll_flags", self.fll_flags.bits() as i64),
+                SourceMetric::new_int("pps_flags", self.pps_flags.bits() as i64),
                // system flags are kind of useless because we can't guarantee they get upstreamed and will only appear once since they are 'event flags'
-        ]
+            ],
+        }]
    }
 }

@@ -203,19 +209,28 @@ impl TryFrom<&[u8]> for Prs10Status {
            .map(|s| str::from_utf8(s).unwrap().parse::<u8>())
            .collect_tuple()
            .ok_or("Not enough parts in ST reply")?;
+
+        let volt_lamp_flags = volt_lamp_flags?;
+        let rf_flags = rf_flags?;
+        let temp_flags = temp_flags?;
+        let fll_flags = fll_flags?;
+        let pps_flags = pps_flags?;
+        let system_flags = system_flags?;
+
        Ok(Self {
-            volt_lamp_flags: Prs10PowerLampFlags::from_bits(volt_lamp_flags?)
-                .ok_or("Invalid bits set ({volt_lamp_flags}) for power/lamp flags")?,
-            rf_flags: Prs10RfFlags::from_bits(rf_flags?)
-                .ok_or("Invalid bits set ({rf_flags}) for RF flags")?,
-            temp_flags: Prs10TempFlags::from_bits(temp_flags?)
-                .ok_or("Invalid bits set ({temp_flags}) for temp flags")?,
-            fll_flags: Prs10FllFlags::from_bits(fll_flags?)
-                .ok_or("Invalid bits set ({fll_flags}) for FLL flags")?,
-            pps_flags: Prs10PpsFlags::from_bits(pps_flags?)
-                .ok_or("Invalid bits set ({pps_flags}) for PPS flags")?,
-            system_flags: Prs10SystemFlags::from_bits(system_flags?)
-                .ok_or("Invalid bits set ({system_flags}) for system flags")?,
+            volt_lamp_flags: Prs10PowerLampFlags::from_bits(volt_lamp_flags).ok_or_else(|| {
+                format!("Invalid bits set ({volt_lamp_flags}) for power/lamp flags")
+            })?,
+            rf_flags: Prs10RfFlags::from_bits(rf_flags)
+                .ok_or_else(|| format!("Invalid bits set ({rf_flags}) for RF flags"))?,
+            temp_flags: Prs10TempFlags::from_bits(temp_flags)
+                .ok_or_else(|| format!("Invalid bits set ({temp_flags}) for temp flags"))?,
+            fll_flags: Prs10FllFlags::from_bits(fll_flags)
+                .ok_or_else(|| format!("Invalid bits set ({fll_flags}) for FLL flags"))?,
+            pps_flags: Prs10PpsFlags::from_bits(pps_flags)
+                .ok_or_else(|| format!("Invalid bits set ({pps_flags}) for PPS flags"))?,
+            system_flags: Prs10SystemFlags::from_bits(system_flags)
+                .ok_or_else(|| format!("Invalid bits set ({system_flags}) for system flags"))?,
        })
    }
 }
@@ -225,7 +240,7 @@ pub struct Prs10Stats {
    pub ocxo_efc: u32,
    pub error_signal_volts: f64,
    pub detect_signal_volts: f64,
-    pub freq_offset_ppt: u16,
+    pub freq_offset_ppt: i16,
    pub mag_efc: u16,
    pub heat_volts: f64,
    pub elec_volts: f64,
@@ -252,118 +267,70 @@ impl SourceReportDetails for Prs10Stats {
    fn is_healthy(&self) -> bool {
        true
    }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
-        let no_tags = Arc::new(vec![]);
-        vec![
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let tags = Arc::new(vec![]);
+        vec![SourceMetricSet {
+            tags,
+            metrics: vec![
                // Integer Metrics
-            SourceMetric::new_int("ocxo_efc", self.ocxo_efc as i64, no_tags.clone()),
+                SourceMetric::new_int("ocxo_efc", self.ocxo_efc as i64),
                // Float Metrics
-            SourceMetric::new_float(
-                "error_signal_volts",
-                self.error_signal_volts,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_float(
-                "detect_signal_volts",
-                self.detect_signal_volts,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_float("heat_volts", self.heat_volts, no_tags.clone()),
-            SourceMetric::new_float("elec_volts", self.elec_volts, no_tags.clone()),
-            SourceMetric::new_float(
-                "lamp_fet_drain_volts",
-                self.lamp_fet_drain_volts,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_float(
-                "lamp_fet_gate_volts",
-                self.lamp_fet_gate_volts,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_float("ocxo_heat_volts", self.ocxo_heat_volts, no_tags.clone()),
-            SourceMetric::new_float("cell_heat_volts", self.cell_heat_volts, no_tags.clone()),
-            SourceMetric::new_float("lamp_heat_volts", self.lamp_heat_volts, no_tags.clone()),
-            SourceMetric::new_float("rb_photo", self.rb_photo, no_tags.clone()),
-            SourceMetric::new_float("rb_photo_iv", self.rb_photo_iv, no_tags.clone()),
-            SourceMetric::new_float("case_temp", self.case_temp, no_tags.clone()),
-            SourceMetric::new_float("ocxo_therm", self.ocxo_therm, no_tags.clone()),
-            SourceMetric::new_float("cell_therm", self.cell_therm, no_tags.clone()),
-            SourceMetric::new_float("lamp_therm", self.lamp_therm, no_tags.clone()),
-            SourceMetric::new_float("ext_cal_volts", self.ext_cal_volts, no_tags.clone()),
-            SourceMetric::new_float("analog_gnd_volts", self.analog_gnd_volts, no_tags.clone()),
-            SourceMetric::new_float(
-                "if_vco_varactor_volts",
-                self.if_vco_varactor_volts,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_float(
-                "op_vco_varactor_volts",
-                self.op_vco_varactor_volts,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_float(
-                "mul_amp_gain_volts",
-                self.mul_amp_gain_volts,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_float("rf_lock_volts", self.rf_lock_volts, no_tags.clone()),
+                SourceMetric::new_float("error_signal_volts", self.error_signal_volts),
+                SourceMetric::new_float("detect_signal_volts", self.detect_signal_volts),
+                SourceMetric::new_float("heat_volts", self.heat_volts),
+                SourceMetric::new_float("elec_volts", self.elec_volts),
+                SourceMetric::new_float("lamp_fet_drain_volts", self.lamp_fet_drain_volts),
+                SourceMetric::new_float("lamp_fet_gate_volts", self.lamp_fet_gate_volts),
+                SourceMetric::new_float("ocxo_heat_volts", self.ocxo_heat_volts),
+                SourceMetric::new_float("cell_heat_volts", self.cell_heat_volts),
+                SourceMetric::new_float("lamp_heat_volts", self.lamp_heat_volts),
+                SourceMetric::new_float("rb_photo", self.rb_photo),
+                SourceMetric::new_float("rb_photo_iv", self.rb_photo_iv),
+                SourceMetric::new_float("case_temp", self.case_temp),
+                SourceMetric::new_float("ocxo_therm", self.ocxo_therm),
+                SourceMetric::new_float("cell_therm", self.cell_therm),
+                SourceMetric::new_float("lamp_therm", self.lamp_therm),
+                SourceMetric::new_float("ext_cal_volts", self.ext_cal_volts),
+                SourceMetric::new_float("analog_gnd_volts", self.analog_gnd_volts),
+                SourceMetric::new_float("if_vco_varactor_volts", self.if_vco_varactor_volts),
+                SourceMetric::new_float("op_vco_varactor_volts", self.op_vco_varactor_volts),
+                SourceMetric::new_float("mul_amp_gain_volts", self.mul_amp_gain_volts),
+                SourceMetric::new_float("rf_lock_volts", self.rf_lock_volts),
                // U16 Metrics (optional, but can be treated as integers)
-            SourceMetric::new_int(
-                "freq_offset_ppt",
-                self.freq_offset_ppt as i64,
-                no_tags.clone(),
-            ),
-            SourceMetric::new_int("mag_efc", self.mag_efc as i64, no_tags.clone()),
-        ]
+                SourceMetric::new_int("freq_offset_ppt", self.freq_offset_ppt as i64),
+                SourceMetric::new_int("mag_efc", self.mag_efc as i64),
+            ],
+        }]
    }
 }

 #[derive(Debug)]
 pub struct Prs10Monitor {
-    rx: ReadHalf<SerialStream>,
-    tx: WriteHalf<SerialStream>,
-    info: OnceCell<Prs10Info>,
+    name: String,
    config: Prs10Config,
+    rx: ReadHalf<SerialStream>,
+    tx: BufWriter<WriteHalf<SerialStream>>,
+    info: OnceCell<Prs10Info>,
 }

 impl Prs10Monitor {
-    pub fn new(config: Prs10Config) -> Self {
-        let builder = tokio_serial::new(&config.port, config.baud)
-            .timeout(config.timeout)
-            .data_bits(tokio_serial::DataBits::Eight)
-            .parity(tokio_serial::Parity::None)
-            .stop_bits(tokio_serial::StopBits::One)
-            .flow_control(tokio_serial::FlowControl::None);
-        let mut port = SerialStream::open(&builder).expect("Must be able to open serial port");
-        port.set_exclusive(true).expect("Can't lock serial port");
-        info!(
-            "Opened serial port {}@{}",
-            port.name().unwrap(),
-            port.baud_rate().unwrap()
-        );
-        let (rx, tx) = tokio::io::split(port);
-
-        Self {
-            rx,
-            tx,
-            config,
-            info: OnceCell::new(),
-        }
-    }
-
    pub fn info(&self) -> &Prs10Info {
        self.info.get().expect("info() used before run()")
    }

    #[instrument(level = "debug", skip_all, fields(cmd = String::from_utf8_lossy(cmd).to_string()))]
    pub async fn cmd_response(&mut self, cmd: &[u8]) -> Result<Vec<u8>, std::io::Error> {
-        self.tx.write_all(cmd).await.unwrap();
-        self.tx.write_u8(b'\r').await.unwrap();
+        self.tx.write_all(cmd).await?;
+        self.tx.write_u8(b'\r').await?;
+        self.tx.flush().await?;
        let mut reader = BufReader::new(&mut self.rx);
        let mut buf = Vec::new();
        let read = timeout(self.config.timeout, reader.read_until(b'\r', &mut buf)).await??;
        buf.truncate(buf.len() - 1); // strip "\r"
-        debug!("response: ({read}) `{buf:?}`");
+        debug!(
+            "raw response: ({read}) `{}`",
+            str::from_utf8(&buf).unwrap_or("<garbage>")
+        );
        Ok(buf)
    }

@@ -391,13 +358,11 @@ impl Prs10Monitor {
    }

    pub async fn get_analog(&mut self, id: u16) -> Result<f64, Box<dyn std::error::Error>> {
-        debug!("Getting analog value {id}");
        let mut cmd = b"AD".to_vec();
        cmd.extend_from_slice(id.to_string().as_bytes());
        cmd.push(b'?');
-        let resp = self.cmd_response(&cmd).await?;
-        let value = str::from_utf8(&resp)?.parse::<f64>()?;
-
+        let value = self.get_parsed(&cmd).await?;
+        debug!("Got: {value}");
        Ok(value)
    }

@@ -408,7 +373,11 @@ impl Prs10Monitor {
    where
        T::Err: std::error::Error + 'static,
    {
-        debug!("Getting int value for command {cmd:?}");
+        debug!(
+            "Getting parsed <{}> value for command {}",
+            type_name::<T>(),
+            str::from_utf8(cmd).unwrap_or("<garbage>"),
+        );
        let resp = self.cmd_response(cmd).await?;
        let val = str::from_utf8(&resp)?.parse::<T>()?;
        Ok(val)
@@ -430,38 +399,38 @@ impl Prs10Monitor {
    }

    pub async fn get_detected_signals(&mut self) -> Result<(f64, f64), Box<dyn std::error::Error>> {
-        debug!("Getting detected signals pair");
+        debug!("Getting i16,i16 -> f64,f64 detected signals pair");
        let resp = self.cmd_response(b"DS?").await?;
        let (error, signal) = resp
            .splitn(2, |c| *c == b',')
-            .map(|s| str::from_utf8(s).unwrap().parse::<u16>())
+            .map(|s| str::from_utf8(s).unwrap().parse::<i16>())
            .collect_tuple()
            .ok_or("Not enough values in response to DS?".to_string())?;
        Ok((error? as f64 * 0.15e-6, signal? as f64 * 0.001))
    }

+    #[instrument(skip_all)]
    async fn status_poll(&mut self) -> Result<ChimemonMessage, Box<dyn std::error::Error>> {
-        debug!("polling status");
        let status = self.get_status().await?;
-        Ok(ChimemonMessage::SourceReport(SourceReport {
-            name: "prs10".into(),
+        Ok(SourceReport {
+            name: self.name.clone(),
            status: if status.is_healthy() {
                SourceStatus::Healthy
            } else {
                SourceStatus::Unknown
            },
            details: Arc::new(status),
-        }))
+        }
+        .into())
    }

+    #[instrument(skip_all)]
    async fn stats_poll(&mut self) -> Result<ChimemonMessage, Box<dyn std::error::Error>> {
        const ANALOG_SCALING: [f64; 20] = [
            0.0, 10.0, 10.0, 10.0, 10.0, 1.0, 1.0, 1.0, 1.0, 4.0, 100.0, 1.0, 1.0, 1.0, 1.0, 1.0,
            4.0, 4.0, 4.0, 1.0,
        ];

-        debug!("polling stats");
-
        let stats_span = debug_span!("get_stats_serial");
        let stats_guard = stats_span.enter();
        let ocxo_efc = self.get_ocxo_efc().await?;
@@ -474,8 +443,8 @@ impl Prs10Monitor {
        }
        drop(stats_guard);

-        Ok(ChimemonMessage::SourceReport(SourceReport {
-            name: "prs10".into(),
+        Ok(SourceReport {
+            name: self.name.clone(),
            status: SourceStatus::Unknown,
            details: Arc::new(Prs10Stats {
                ocxo_efc,
@@ -503,16 +472,75 @@ impl Prs10Monitor {
                mul_amp_gain_volts: analog_values[18],
                rf_lock_volts: analog_values[19],
            }),
-        }))
+        }
+        .into())
+    }
+
+    async fn reset_rx_state(&mut self) -> Result<(), Box<dyn std::error::Error>> {
+        // flush any pending input and potential responses from the receiver side
+        self.tx.write_u8(b'\r').await?;
+        self.tx.flush().await?;
+        let mut discard = vec![];
+        loop {
+            match timeout(Duration::from_millis(100), self.rx.read_buf(&mut discard)).await {
+                Ok(_) => discard.clear(),
+                Err(_) => break,
+            }
+        }
+        Ok(())
    }
 }

 #[async_trait]
 impl ChimemonSource for Prs10Monitor {
-    async fn run(mut self, chan: ChimemonSourceChannel) {
-        info!("PRS10 task starting");
+    type Config = Prs10Config;
+    const TASK_NAME: &'static str = "prs10-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let builder = tokio_serial::new(&config.port, config.baud)
+            .timeout(config.timeout)
+            .data_bits(tokio_serial::DataBits::Eight)
+            .parity(tokio_serial::Parity::None)
+            .stop_bits(tokio_serial::StopBits::One)
+            .flow_control(tokio_serial::FlowControl::None);
+        let mut port = SerialStream::open(&builder).unwrap_or_else(|e| {
+            fatal!(
+                "Failed to open serial port `{}` ({})",
+                config.port,
+                e.to_string()
+            )
+        });
+        port.set_exclusive(true).unwrap_or_else(|e| {
+            fatal!(
+                "Can't lock serial port `{}` ({})",
+                config.port,
+                e.to_string()
+            )
+        });
+        info!(
+            "Opened serial port {}@{}",
+            port.name().unwrap(),
+            port.baud_rate().unwrap()
+        );
+
+        let (rx, tx) = tokio::io::split(port);
+        let tx = BufWriter::new(tx);
+        Self {
+            name: name.to_owned(),
+            config,
+            rx,
+            tx,
+            info: OnceCell::new(),
+        }
+    }
+
+    async fn run(mut self, chan: ChimemonSourceChannel, cancel: CancellationToken) {
+        info!("PRS10 task started");
+        if let Err(e) = self.reset_rx_state().await {
+            error!(error = ?e, "Error clearing PRS10 RX state");
+            return;
+        }
        if let Err(e) = self.set_info().await {
-            warn!("Error starting PRS10: {e:?}");
+            error!(error = ?e, "Error starting PRS10");
            return;
        }
        info!(
@@ -527,6 +555,9 @@ impl ChimemonSource for Prs10Monitor {

        loop {
            let msg = select! {
+                _ = cancel.cancelled() => {
+                    return;
+                },
                _ = status_timer.tick() => {
                    self.status_poll().await
                },
--- a/src/sources/uccm.rs
+++ b/src/sources/uccm.rs
@@ -1,25 +1,26 @@
-use crate::{
-    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Config, SourceMetric,
-    SourceReportDetails, TimeReport,
-};
+use std::io::{BufRead, Cursor};
+use std::str;
+use std::sync::Arc;
+
 use async_trait::async_trait;
 use bitflags::bitflags;
 use byteorder::{BigEndian, ReadBytesExt};
 use bytes::{Buf, BytesMut};
 use chrono::{DateTime, Duration, NaiveDateTime, Utc};
-use figment::value::Map;
-use influxdb2::models::DataPoint;
-use influxdb2::models::data_point::DataPointBuilder;
 use itertools::Itertools;
-use std::io::{BufRead, Cursor};
-use std::str;
-use std::sync::Arc;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf};
-use tokio::join;
+use tokio::select;
 use tokio::sync::Mutex;
 use tokio::time::sleep;
 use tokio_serial::{SerialPort, SerialStream};
-use tracing::{debug, info, warn};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, warn};
+
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, SourceMetric, SourceReport, SourceReportDetails,
+    SourceStatus, TimeReport, config::UCCMConfig,
+};
+use crate::{SourceMetricSet, fatal};

 pub const GPS_EPOCH: i64 = 315964800; // Doesn't seem possible to have a const DateTime object
 pub type UccmEndian = BigEndian;
@@ -32,11 +33,11 @@ pub enum UCCMMonitorParseState {
 }

 pub struct UCCMMonitor {
-    // pub port: SerialStream,
+    pub name: String,
+    config: UCCMConfig,
    rx: ReadHalf<SerialStream>,
    tx: WriteHalf<SerialStream>,
    pub info: Option<UCCMInfo>,
-    config: Config,
 }

 #[derive(Debug)]
@@ -59,88 +60,33 @@ impl SourceReportDetails for UCCMTODReport {
            && !self.flags.contains(UCCMFlags::GPS_LOS)
    }

-    fn to_metrics(&self) -> Vec<SourceMetric> {
-        let no_tags = Arc::new(vec![]);
-        vec![
-            SourceMetric::new_int("leaps", self.leaps as i64, no_tags.clone()),
-            SourceMetric::new_bool(
-                "osc_lock",
-                self.flags.contains(UCCMFlags::OSC_LOCK),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "leap_flag",
-                self.flags.contains(UCCMFlags::LEAP_FLAG),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "init_unlock",
-                self.flags.contains(UCCMFlags::INIT_UNLOCK),
-                no_tags.clone(),
-            ),
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let tags = Arc::new(vec![]);
+        vec![SourceMetricSet {
+            tags,
+            metrics: vec![
+                SourceMetric::new_int("leaps", self.leaps as i64),
+                SourceMetric::new_bool("osc_lock", self.flags.contains(UCCMFlags::OSC_LOCK)),
+                SourceMetric::new_bool("leap_flag", self.flags.contains(UCCMFlags::LEAP_FLAG)),
+                SourceMetric::new_bool("init_unlock", self.flags.contains(UCCMFlags::INIT_UNLOCK)),
                SourceMetric::new_bool(
                    "init_no_sats",
                    self.flags.contains(UCCMFlags::INIT_NO_SATS),
-                no_tags.clone(),
                ),
                SourceMetric::new_bool(
                    "have_gps_time",
                    self.flags.contains(UCCMFlags::HAVE_GPS_TIME),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "power_fail",
-                self.flags.contains(UCCMFlags::POWER_FAIL),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "no_gps_sync",
-                self.flags.contains(UCCMFlags::NO_GPS_SYNC),
-                no_tags.clone(),
                ),
+                SourceMetric::new_bool("power_fail", self.flags.contains(UCCMFlags::POWER_FAIL)),
+                SourceMetric::new_bool("no_gps_sync", self.flags.contains(UCCMFlags::NO_GPS_SYNC)),
                SourceMetric::new_bool(
                    "no_gps_sync2",
                    self.flags.contains(UCCMFlags::NO_GPS_SYNC2),
-                no_tags.clone(),
                ),
-            SourceMetric::new_bool(
-                "ant_fault",
-                self.flags.contains(UCCMFlags::NO_ANT),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "gps_los",
-                self.flags.contains(UCCMFlags::GPS_LOS),
-                no_tags.clone(),
-            ),
-        ]
-    }
-}
-
-impl UCCMTODReport {
-    pub fn as_builder(&self, measurement: &String, tags: &Map<String, String>) -> DataPointBuilder {
-        let mut builder =
-            DataPoint::builder(measurement).timestamp(self.time.timestamp_nanos_opt().unwrap());
-        builder = builder.field("leaps", self.leaps as i64);
-        builder = builder.field("osc_lock", self.flags.contains(UCCMFlags::OSC_LOCK));
-        builder = builder.field("leap_flag", self.flags.contains(UCCMFlags::LEAP_FLAG));
-        builder = builder.field("init_unlock", self.flags.contains(UCCMFlags::INIT_UNLOCK));
-        builder = builder.field("init_no_sats", self.flags.contains(UCCMFlags::INIT_NO_SATS));
-        builder = builder.field(
-            "have_gps_time",
-            self.flags.contains(UCCMFlags::HAVE_GPS_TIME),
-        );
-        builder = builder.field("power_fail", self.flags.contains(UCCMFlags::POWER_FAIL));
-        builder = builder.field("no_gps_sync", self.flags.contains(UCCMFlags::NO_GPS_SYNC));
-        builder = builder.field("no_gps_sync2", self.flags.contains(UCCMFlags::NO_GPS_SYNC2));
-        builder = builder.field("ant_fault", self.flags.contains(UCCMFlags::NO_ANT));
-        builder = builder.field("gps_los", self.flags.contains(UCCMFlags::GPS_LOS));
-
-        builder = tags
-            .iter()
-            .fold(builder, |builder, (k, v)| builder.tag(k, v));
-
-        builder
+                SourceMetric::new_bool("ant_fault", self.flags.contains(UCCMFlags::NO_ANT)),
+                SourceMetric::new_bool("gps_los", self.flags.contains(UCCMFlags::GPS_LOS)),
+            ],
+        }]
    }
 }

@@ -149,14 +95,16 @@ pub struct UCCMLoopDiagReport {
    pub ocxo: f32,
 }

-impl UCCMLoopDiagReport {
-    pub fn as_builder(&self, measurement: &String, tags: &Map<String, String>) -> DataPointBuilder {
-        let mut builder = DataPoint::builder(measurement);
-        builder = builder.field("ocxo_offset", self.ocxo as f64);
-        builder = tags
-            .iter()
-            .fold(builder, |builder, (k, v)| builder.tag(k, v));
-        builder
+impl SourceReportDetails for UCCMLoopDiagReport {
+    fn is_healthy(&self) -> bool {
+        true
+    }
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let tags = Arc::new(vec![]);
+        vec![SourceMetricSet {
+            tags,
+            metrics: vec![SourceMetric::new_float("ocxo_offset", self.ocxo as f64)],
+        }]
    }
 }

@@ -166,16 +114,9 @@ pub struct UCCMGpsSvTracking {
    pub cno: u8,
 }

-impl UCCMGpsSvTracking {
-    fn as_builder(&self, measurement: &String, tags: &Map<String, String>) -> DataPointBuilder {
-        let mut builder = DataPoint::builder(measurement)
-            .field("sv_cno", self.cno as i64)
-            .tag("sv_id", self.sv.to_string());
-        builder = tags
-            .iter()
-            .fold(builder, |builder, (k, v)| builder.tag(k, v));
-
-        builder
+impl From<&UCCMGpsSvTracking> for SourceMetric {
+    fn from(value: &UCCMGpsSvTracking) -> Self {
+        SourceMetric::new_int("sv_cno", value.cno as i64)
    }
 }

@@ -184,13 +125,15 @@ pub struct UCCMGPSSatsReport {
    pub tracked_svs: Vec<UCCMGpsSvTracking>,
 }

-impl UCCMGPSSatsReport {
-    pub fn build(&self, measurement: &String, tags: &Map<String, String>) -> Vec<DataPoint> {
-        self.tracked_svs
-            .iter()
-            .map(|sv| sv.as_builder(measurement, tags))
-            .map(|b| b.build().unwrap())
-            .collect()
+impl SourceReportDetails for UCCMGPSSatsReport {
+    fn is_healthy(&self) -> bool {
+        self.tracked_svs.len() >= 4
+    }
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        vec![SourceMetricSet {
+            tags: Arc::new(vec![]),
+            metrics: self.tracked_svs.iter().map(|sv| sv.into()).collect(),
+        }]
    }
 }

@@ -273,20 +216,23 @@ impl SourceReportDetails for UCCMStatusReport {
    fn is_healthy(&self) -> bool {
        self.gps_pps_valid
    }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
-        let no_tags = Arc::new(vec![]);
-        vec![
-            SourceMetric::new_int("tfom", self.tfom as i64, no_tags.clone()),
-            SourceMetric::new_int("ffom", self.ffom as i64, no_tags.clone()),
-            SourceMetric::new_float("gps_phase", self.gps_phase as f64, no_tags.clone()),
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let tags = Arc::new(vec![]);
+        vec![SourceMetricSet {
+            tags,
+            metrics: vec![
+                SourceMetric::new_int("tfom", self.tfom as i64),
+                SourceMetric::new_int("ffom", self.ffom as i64),
+                SourceMetric::new_float("gps_phase", self.gps_phase as f64),
                // TODO: sv info
                // TOOD: timestamp
-            SourceMetric::new_float("ant_voltage", self.ant_voltage as f64, no_tags.clone()),
-            SourceMetric::new_float("ant_current", self.ant_current as f64, no_tags.clone()),
-            SourceMetric::new_float("temp", self.temp as f64, no_tags.clone()),
-            SourceMetric::new_int("efc_dac", self.efc_dac as i64, no_tags.clone()),
-            SourceMetric::new_float("freq_error", self.freq_error as f64, no_tags.clone()),
-        ]
+                SourceMetric::new_float("ant_voltage", self.ant_voltage as f64),
+                SourceMetric::new_float("ant_current", self.ant_current as f64),
+                SourceMetric::new_float("temp", self.temp as f64),
+                SourceMetric::new_int("efc_dac", self.efc_dac as i64),
+                SourceMetric::new_float("freq_error", self.freq_error as f64),
+            ],
+        }]
    }
 }

@@ -399,30 +345,6 @@ impl TryFrom<&str> for UCCMGPSSatsReport {
 }

 impl UCCMMonitor {
-    pub fn new(config: Config) -> Self {
-        let builder = tokio_serial::new(&config.sources.uccm.port, config.sources.uccm.baud)
-            .timeout(config.sources.uccm.timeout)
-            .data_bits(tokio_serial::DataBits::Eight)
-            .parity(tokio_serial::Parity::None)
-            .stop_bits(tokio_serial::StopBits::One)
-            .flow_control(tokio_serial::FlowControl::None);
-        let mut port = SerialStream::open(&builder).expect("Must be able to open serial port");
-        port.set_exclusive(true).expect("Can't lock serial port");
-        info!(
-            "Opened serial port {}@{}",
-            port.name().unwrap(),
-            port.baud_rate().unwrap()
-        );
-        let (rx, tx) = tokio::io::split(port);
-        UCCMMonitor {
-            // port,
-            rx,
-            tx,
-            info: None,
-            config,
-        }
-    }
-
    pub async fn send_cmd(&mut self, cmd: &[u8]) -> Result<String, std::io::Error> {
        debug!("cmd: `{:?}`", String::from_utf8_lossy(cmd));
        self.tx.write_all(cmd).await.unwrap();
@@ -478,10 +400,10 @@ impl UCCMMonitor {
        state: Arc<Mutex<UCCMMonitorParseState>>,
    ) {
        let mut rdbuf = BytesMut::with_capacity(1024);
-        let mut last_loop_diag: Option<UCCMLoopDiagReport> = None;
-        let mut last_gps_sats: Option<UCCMGPSSatsReport> = None;
+        let mut last_loop_diag: Option<Arc<UCCMLoopDiagReport>> = None;
+        let mut last_gps_sats: Option<Arc<UCCMGPSSatsReport>> = None;

-        let mut last_sent_report = Utc::now() - self.config.sources.uccm.status_interval;
+        let mut last_sent_report = Utc::now() - self.config.status_interval;

        loop {
            match tokio::io::AsyncReadExt::read_buf(&mut self.rx, &mut rdbuf).await {
@@ -516,46 +438,45 @@ impl UCCMMonitor {
                                && tod
                                    .flags
                                    .contains(UCCMFlags::OSC_LOCK | UCCMFlags::HAVE_GPS_TIME);
-                            chan.send(ChimemonMessage::TimeReport(TimeReport {
+                            chan.send(
+                                TimeReport {
                                    system_time: sysnow,
                                    offset,
                                    leaps: tod.leaps as isize,
                                    leap_flag: tod.flags.contains(UCCMFlags::LEAP_FLAG),
                                    valid,
-                            }))
+                                }
+                                .into(),
+                            )
                            .expect("Unable to send to channel");
                            if sysnow - last_sent_report
-                                >= Duration::from_std(self.config.sources.uccm.status_interval)
-                                    .unwrap()
+                                >= Duration::from_std(self.config.status_interval).unwrap()
                            {
-                                let mut points = vec![
-                                    tod.as_builder(
-                                        &self.config.sources.uccm.measurement,
-                                        &self.config.influxdb.tags,
-                                    )
-                                    .build()
-                                    .unwrap(),
-                                ];
                                if let Some(loop_diag) = &last_loop_diag {
-                                    points.push(
-                                        loop_diag
-                                            .as_builder(
-                                                &self.config.sources.uccm.measurement,
-                                                &self.config.influxdb.tags,
-                                            )
-                                            .build()
-                                            .unwrap(),
-                                    )
+                                    if let Err(e) = chan.send(
+                                        SourceReport {
+                                            name: "uccm".to_owned(),
+                                            status: SourceStatus::Unknown,
+                                            details: loop_diag.clone(),
+                                        }
+                                        .into(),
+                                    ) {
+                                        error!(error = ?e, "Unable to send message to channel");
+                                    }
                                }
                                if let Some(gps_sats) = &last_gps_sats {
-                                    points.extend(gps_sats.build(
-                                        &self.config.sources.uccm.measurement,
-                                        &self.config.influxdb.tags,
-                                    ));
+                                    if let Err(e) = chan.send(
+                                        SourceReport {
+                                            name: "uccm".to_owned(),
+                                            status: SourceStatus::Unknown,
+                                            details: gps_sats.clone(),
+                                        }
+                                        .into(),
+                                    ) {
+                                        error!(error = ?e, "Unable to send message to channel");
+                                    }
                                }

-                                chan.send(ChimemonMessage::DataPoints(points))
-                                    .expect("Unable to send to channel");
                                last_sent_report = sysnow;
                            }
                        }
@@ -569,7 +490,7 @@ impl UCCMMonitor {
                    let loop_report = UCCMLoopDiagReport::try_from(loop_diag_resp.as_str());
                    let gps_report = UCCMGPSSatsReport::try_from(gps_sats_resp.as_str());
                    if let Ok(loop_report) = loop_report {
-                        last_loop_diag = Some(loop_report)
+                        last_loop_diag = Some(Arc::new(loop_report))
                    } else {
                        warn!(
                            "Unable to parse loop diag report `{}`: {}",
@@ -578,7 +499,7 @@ impl UCCMMonitor {
                        );
                    }
                    if let Ok(gps_report) = gps_report {
-                        last_gps_sats = Some(gps_report)
+                        last_gps_sats = Some(Arc::new(gps_report))
                    } else {
                        warn!(
                            "Unable to parse GPS sats report `{}`: {}",
@@ -597,7 +518,38 @@ impl UCCMMonitor {

 #[async_trait]
 impl ChimemonSource for UCCMMonitor {
-    async fn run(mut self, chan: ChimemonSourceChannel) {
+    type Config = UCCMConfig;
+    const TASK_NAME: &'static str = "uccm-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let builder = tokio_serial::new(&config.port, config.baud)
+            .timeout(config.timeout)
+            .data_bits(tokio_serial::DataBits::Eight)
+            .parity(tokio_serial::Parity::None)
+            .stop_bits(tokio_serial::StopBits::One)
+            .flow_control(tokio_serial::FlowControl::None);
+        let mut port = match SerialStream::open(&builder) {
+            Ok(port) => port,
+            Err(e) => fatal!(error = ?e, "Error opening port {}", &config.port),
+        };
+        if let Err(e) = port.set_exclusive(true) {
+            fatal!(error= ?e, "Can't lock serial port");
+        };
+        info!(
+            "Opened serial port {}@{}",
+            port.name().unwrap(),
+            port.baud_rate().unwrap()
+        );
+        let (rx, tx) = tokio::io::split(port);
+        UCCMMonitor {
+            name: name.to_owned(),
+            config,
+            rx,
+            tx,
+            info: None,
+        }
+    }
+
+    async fn run(mut self, chan: ChimemonSourceChannel, cancel: CancellationToken) {
        info!("UCCM task starting");
        if self.get_info().await.is_err() {
            warn!("Error starting UCCM");
@@ -623,7 +575,10 @@ impl ChimemonSource for UCCMMonitor {
        //         wfut.await;
        //     }
        // });
-
-        join!(rx_handle).0.unwrap();
+        //
+        select! {
+            _ = cancel.cancelled() => { return },
+            _ = rx_handle => { return }
+        };
    }
 }
--- a/src/targets/chrony_refclock.rs
+++ b/src/targets/chrony_refclock.rs
@@ -1,14 +1,19 @@
-use crate::{ChimemonMessage, ChimemonTarget, ChimemonTargetChannel, ChronySockConfig};
-use async_trait::async_trait;
-use libc::{c_double, c_int, timeval};
 use std::mem;
 use std::os::unix::net::UnixDatagram;
 use std::path::PathBuf;
-use tracing::debug;
+
+use async_trait::async_trait;
+use libc::{c_double, c_int, timeval};
+use tokio::select;
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, info, warn};
+
+use crate::{ChimemonMessage, ChimemonTarget, ChimemonTargetChannel, config::ChronySockConfig};

 const CHRONY_MAGIC: c_int = 0x534f434b;

 pub struct ChronySockServer {
+    name: String,
    sock_path: PathBuf,
 }

@@ -23,21 +28,27 @@ pub struct ChronyTimeReport {
    magic: c_int,
 }

-impl ChronySockServer {
-    pub fn new(config: ChronySockConfig) -> Self {
-        ChronySockServer {
-            sock_path: config.sock.into(),
-        }
-    }
-}
+impl ChronySockServer {}

 #[async_trait]
 impl ChimemonTarget for ChronySockServer {
-    async fn run(mut self, mut chan: ChimemonTargetChannel) {
+    type Config = ChronySockConfig;
+    const TASK_NAME: &'static str = "chrony-refclock-task";
+    fn new(name: &str, config: ChronySockConfig) -> Self {
+        ChronySockServer {
+            name: name.to_owned(),
+            sock_path: config.sock.into(),
+        }
+    }
+    async fn run(mut self, mut chan: ChimemonTargetChannel, cancel: CancellationToken) {
+        info!("Chrony refclock task started");
        loop {
-            let msg = chan.recv().await.unwrap();
+            select! {
+                _ = cancel.cancelled() => { return }
+                msg = chan.recv() => {
                    match msg {
-                ChimemonMessage::TimeReport(tr) => {
+                        Ok(ChimemonMessage::TimeReport(tr)) => {
+                            debug!(tr = ?tr, "Got timereport");
                            if tr.valid {
                                {
                                    let frame = ChronyTimeReport {
@@ -66,9 +77,12 @@ impl ChimemonTarget for ChronySockServer {
                                    sock.send_to(bs, &self.sock_path).unwrap();
                                }
                            }
-                }
+                        },
+                        Err(e) => warn!("Error receiving from channel: {}", e.to_string()),
                        _ => continue,
                    }
                }
            }
        }
+    }
+}
--- a/src/targets/influx.rs
+++ b/src/targets/influx.rs
@@ -0,0 +1,110 @@
+use async_trait::async_trait;
+use futures::stream;
+use influxdb2::{
+    Client,
+    models::{DataPoint, FieldValue},
+};
+use tokio::{select, sync::broadcast, time::timeout};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, instrument};
+
+use crate::{
+    ChimemonMessage, ChimemonTarget, ChimemonTargetChannel, MetricValue, SourceReport,
+    config::InfluxConfig, fatal,
+};
+
+pub struct InfluxTarget {
+    name: String,
+    config: InfluxConfig,
+    influx: Client,
+}
+
+impl From<MetricValue> for FieldValue {
+    fn from(value: MetricValue) -> Self {
+        match value {
+            MetricValue::Bool(b) => FieldValue::Bool(b),
+            MetricValue::Float(f) => FieldValue::F64(f),
+            MetricValue::Int(i) => FieldValue::I64(i),
+        }
+    }
+}
+
+#[async_trait]
+impl ChimemonTarget for InfluxTarget {
+    type Config = InfluxConfig;
+    const TASK_NAME: &'static str = "influx-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let influx = Client::new(&config.url, &config.org, &config.token);
+        Self {
+            name: name.to_owned(),
+            config: config,
+            influx,
+        }
+    }
+    async fn run(self, mut chan: ChimemonTargetChannel, cancel: CancellationToken) {
+        info!("Influx task started");
+
+        loop {
+            let msg = select! {
+                _ = cancel.cancelled() => { return },
+                msg = chan.recv() => msg
+            };
+            debug!(msg = ?msg, "Got msg");
+            let msg = match msg {
+                Ok(msg) => msg,
+                Err(broadcast::error::RecvError::Closed) => {
+                    fatal!("Permanent channel closed, terminating")
+                }
+                Err(broadcast::error::RecvError::Lagged(_)) => {
+                    error!("Channel lagged");
+                    continue;
+                }
+            };
+            if let Err(e) = self.handle_msg(&msg).await {
+                error!(error = ?e, msg=?&msg, "Error handling message");
+            }
+        }
+    }
+}
+
+impl InfluxTarget {
+    #[instrument(skip_all)]
+    async fn handle_source_report(
+        &self,
+        sr: &SourceReport,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        debug!("Handling source report {}", sr.name);
+        let mut dps = Vec::new();
+        for metric_set in &sr.details.to_metrics() {
+            let mut builder = DataPoint::builder(&sr.name);
+            builder = self
+                .config
+                .tags
+                .iter()
+                .fold(builder, |builder, (k, v)| builder.tag(k, v));
+            builder = metric_set
+                .tags
+                .iter()
+                .fold(builder, |builder, (k, v)| builder.tag(*k, v));
+            builder = metric_set.metrics.iter().fold(builder, |builder, metric| {
+                builder.field(metric.name, metric.value)
+            });
+            dps.push(builder.build()?);
+        }
+        debug!("Sending {} datapoints to influx", dps.len());
+        timeout(
+            self.config.timeout,
+            self.influx.write(&self.config.bucket, stream::iter(dps)),
+        )
+        .await??;
+        debug!("All datapoints sent");
+        Ok(())
+    }
+    async fn handle_msg(&self, msg: &ChimemonMessage) -> Result<(), Box<dyn std::error::Error>> {
+        debug!(msg = ?msg, "Handling msg");
+        match msg {
+            ChimemonMessage::TimeReport(_tr) => Ok(()),
+            ChimemonMessage::SourceReport(sr) => self.handle_source_report(sr).await,
+        }
+    }
+}
--- a/src/targets/mod.rs
+++ b/src/targets/mod.rs
@@ -1 +1,2 @@
 pub mod chrony_refclock;
+pub mod influx;
Author	SHA1	Message	Date
Keenan Tims	c22ad764e6	improvements	2026-02-04 18:53:05 -08:00
Keenan Tims	50894fd0e1	refactor config	2026-02-04 14:25:17 -08:00
Keenan Tims	d31bf8d39e	start unpicking influxdb	2026-02-04 13:50:24 -08:00
Keenan Tims	d3eab1da12	prs10: improve serial handling	2026-02-04 11:50:27 -08:00
Keenan Tims	46eedec11e	fixup log level handling	2026-02-04 11:49:56 -08:00
Keenan Tims	30b48f686f	improve error handling with fatal!()	2026-02-04 11:49:24 -08:00
Keenan Tims	08871a5782	cancellation	2026-02-04 03:31:17 -08:00
Keenan Tims	156df9ae86	refactor metric tags & sources config	2026-02-04 02:04:51 -08:00
Keenan Tims	d464cf8ee6	improve parsing of durations	2026-02-03 19:38:02 -08:00
Keenan Tims	2e8e731d80	prs10: fix signedness for parsing some values	2026-02-03 19:37:36 -08:00