diff --git a/Cargo.lock b/Cargo.lock index 9ce1a61..e01221b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1740,16 +1740,16 @@ dependencies = [ ] [[package]] -name = "opentelemetry-http" -version = "0.31.0" +name = "opentelemetry-appender-tracing" +version = "0.31.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d" +checksum = "ef6a1ac5ca3accf562b8c306fa8483c85f4390f768185ab775f242f7fe8fdcc2" dependencies = [ - "async-trait", - "bytes", - "http", "opentelemetry", - "reqwest 0.12.28", + "tracing", + "tracing-core", + "tracing-opentelemetry", + "tracing-subscriber", ] [[package]] @@ -1760,15 +1760,12 @@ checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" dependencies = [ "http", "opentelemetry", - "opentelemetry-http", "opentelemetry-proto", "opentelemetry_sdk", "prost", - "reqwest 0.12.28", "thiserror 2.0.18", "tokio", "tonic", - "tracing", ] [[package]] @@ -1797,8 +1794,6 @@ dependencies = [ "percent-encoding", "rand 0.9.4", "thiserror 2.0.18", - "tokio", - "tokio-stream", ] [[package]] @@ -2183,40 +2178,6 @@ version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" -[[package]] -name = "reqwest" -version = "0.12.28" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" -dependencies = [ - "base64", - "bytes", - "futures-channel", - "futures-core", - "futures-util", - "http", - "http-body", - "http-body-util", - "hyper", - "hyper-util", - "js-sys", - "log", - "percent-encoding", - "pin-project-lite", - "serde", - "serde_json", - "serde_urlencoded", - "sync_wrapper", - "tokio", - "tower", - "tower-http", - "tower-service", - "url", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "reqwest" version = "0.13.2" @@ -3258,8 +3219,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fec7c61a0695dc1887c1b53952990f3ad2e3a31453e1f49f10e75424943a93ec" dependencies = [ "async-trait", + "axum", "base64", "bytes", + "h2", "http", "http-body", "http-body-util", @@ -3268,6 +3231,7 @@ dependencies = [ "hyper-util", "percent-encoding", "pin-project", + "socket2", "sync_wrapper", "tokio", "tokio-stream", @@ -3653,15 +3617,13 @@ dependencies = [ "http", "macaddr", "nix 0.31.2", - "reqwest 0.13.2", + "reqwest", "serde", "serde_json", - "time", "tokio", "tokio-tungstenite 0.29.0", "toml", "tracing", - "tracing-subscriber", "url", "wakey", "wakey-core", @@ -3678,20 +3640,14 @@ dependencies = [ "futures-util", "macaddr", "nix 0.31.2", - "opentelemetry", - "opentelemetry-otlp", - "opentelemetry_sdk", - "reqwest 0.13.2", + "reqwest", "serde", "serde_json", "sqlx", - "time", "tokio", "toml", "tower-http", "tracing", - "tracing-opentelemetry", - "tracing-subscriber", "uuid", "wakey-agent", "wakey-core", @@ -3701,12 +3657,25 @@ dependencies = [ name = "wakey-core" version = "0.3.6" dependencies = [ + "anyhow", "macaddr", + "opentelemetry", + "opentelemetry-appender-tracing", + "opentelemetry-otlp", + "opentelemetry-proto", + "opentelemetry_sdk", "serde", "serde_json", "serde_with", "strum", "thiserror 2.0.18", + "time", + "tokio", + "tokio-stream", + "tonic", + "tracing", + "tracing-opentelemetry", + "tracing-subscriber", ] [[package]] diff --git a/deploy/systemd/wakey-cc.service b/deploy/systemd/wakey-cc.service index f338c54..b090b03 100644 --- a/deploy/systemd/wakey-cc.service +++ b/deploy/systemd/wakey-cc.service @@ -12,6 +12,7 @@ ExecStart=/opt/wakey/bin/wakey-control-plane serve --config-file /etc/wakey-cont Restart=on-failure RestartSec=2 Environment=RUST_LOG=info +EnvironmentFile=-/etc/wakey-control-plane/environment [Install] WantedBy=multi-user.target diff --git a/docs/openwrt-uci.md b/docs/openwrt-uci.md new file mode 100644 index 0000000..116bfdd --- /dev/null +++ b/docs/openwrt-uci.md @@ -0,0 +1,169 @@ +# OpenWrt UCI for Wakey + +Wakey's OpenWrt init script reads `/etc/config/wakey` for the log filter and +OTLP authorization header, then passes them to the agent through procd. +The agent's application settings remain in `/etc/wakey-agent/config.toml`. +The previous `/etc/wakey-agent/environment` file is no longer read. + +## The three names + +```sh +uci set wakey.main.otlp_authorization='Basic BASE64_OF_USERNAME_COLON_PASSWORD' +``` + +| Part | Meaning | +| --- | --- | +| `wakey` | Package: `/etc/config/wakey` | +| `main` | Named section inside that package | +| `otlp_authorization` | Option inside the section | + +Quotes keep the complete `Basic ...` value together despite its space. +The credential is the Base64 encoding of `username:password`. +It is not the htpasswd hash, and Base64 is not encryption. + +## Create and save + +Run as root on the router. `touch` does not truncate an existing file: + +```sh +touch /etc/config/wakey +chown root:root /etc/config/wakey +chmod 600 /etc/config/wakey + +uci set wakey.main=service +uci set wakey.main.log_filter='info' +uci set wakey.main.otlp_authorization='Basic BASE64_OF_USERNAME_COLON_PASSWORD' +uci commit wakey +chmod 600 /etc/config/wakey +``` + +`wakey.main=service` creates a named section of type `service`. The type is our +chosen label, not a built-in Wakey feature. Existing options are not cleared. +`set` stages changes; `commit wakey` writes that package to disk. Committing does +not restart a service. + +The resulting file looks like: + +```text +config service 'main' + option log_filter 'info' + option otlp_authorization 'Basic BASE64_OF_USERNAME_COLON_PASSWORD' +``` + +The example above uses a placeholder. Avoid typing real credentials into a +command that your shell will save in history. On OpenWrt's ash, you can instead +read the Basic header without echoing it and send it to UCI over stdin: + +```sh +read -r -s -p 'Complete Basic authorization header: ' authorization +printf '\n' +printf "set wakey.main.otlp_authorization='%s'\ncommit wakey\n" "$authorization" | uci batch +unset authorization +chmod 600 /etc/config/wakey +``` + +Use a normal `Basic ` header here, without quotes/newlines inside the value. +The shell's builtin `printf` avoids passing the secret in `uci`'s command arguments. +UCI is not a secret vault: committed files, staged changes, router backups, and +root access can expose these credentials. Do not paste their output into logs/chat. + +## Inspect, change, delete + +Inspect the non-secret option: + +```sh +uci get wakey.main.log_filter +``` + +`uci show wakey`, `uci export wakey`, `uci changes wakey`, and +`uci get wakey.main.otlp_authorization` **print the credential**. Use them only +when you deliberately want to see it. + +Change an option: + +```sh +uci set wakey.main.log_filter='wakey_agent=debug,wakey=debug,opentelemetry=warn' +uci commit wakey +``` + +Delete just the authorization option: + +```sh +uci -q delete wakey.main.otlp_authorization +uci commit wakey +``` + +If TOML still requires `authorization_env`, removing the variable makes startup +fail. Remove that TOML field too only when the receiver no longer requires auth. + +Discard all uncommitted changes to this package: + +```sh +uci revert wakey +``` + +## How the service reads it + +`config_load` and `config_get` are shell helpers from `/lib/functions.sh`, available +in the normal `/etc/rc.common` init-script environment. They are not standalone +commands to run directly in your SSH shell. + +```sh +config_load wakey +config_get authorization main otlp_authorization +``` + +`config_get` arguments are **destination variable, section, option, optional default**. +The line above puts the option into the shell variable `authorization`. It does +not export it to the agent automatically; procd must receive it explicitly. + +The checked-in [init script](../scripts/init/openwrt/wakey) uses these helpers. +Its configuration handling is: + +```sh +start_service() { + local authorization='' log_filter='info' + if [ -f /etc/config/wakey ]; then + config_load wakey || return 1 + config_get authorization main otlp_authorization + config_get log_filter main log_filter 'info' + fi + + procd_open_instance + procd_set_param command "$BIN" serve --config "$CONFIG" + procd_set_param file "$CONFIG" + [ ! -f /etc/config/wakey ] || procd_append_param file /etc/config/wakey + procd_set_param env "RUST_LOG=$log_filter" + [ -z "$authorization" ] || procd_append_param env "WAKEY_OTLP_AUTHORIZATION=$authorization" + procd_set_param respawn 5 1 0 + procd_set_param stdout 1 + procd_set_param stderr 1 + procd_close_instance +} +``` + +Keep these fields in the agent's TOML: + +```toml +[telemetry] +otlp_endpoint = "http://10.5.0.1:4317" +authorization_env = "WAKEY_OTLP_AUTHORIZATION" +``` + +UCI stores the header; procd passes the environment variable; Rust reads it using +the name in TOML. The application does not parse UCI itself. + +After installing the updated init script and committing settings: + +```sh +/etc/init.d/wakey restart +``` + +Use **restart**, not the existing HUP-based reload: a running process cannot receive +changed environment variables that way. UCI is only for the router; the CC's +systemd environment file remains unchanged. See [telemetry.md](telemetry.md). + +## References + +- [OpenWrt UCI CLI source and command usage](https://github.com/openwrt/uci/blob/master/cli.c) +- [OpenWrt config_load/config_get helpers](https://github.com/openwrt/openwrt/blob/main/package/base-files/files/lib/functions.sh) diff --git a/docs/telemetry.md b/docs/telemetry.md new file mode 100644 index 0000000..558e0dd --- /dev/null +++ b/docs/telemetry.md @@ -0,0 +1,127 @@ +# Telemetry + +CC and the agent daemon share the optional `wakey-core/telemetry` module. +Local logs remain enabled. Configure an OTLP **gRPC** receiver to additionally +export traces, structured logs, and metrics; HTTP/4318 is not supported in this slice. + +CC configuration: + +```toml +[telemetry] +otlp_endpoint = "http://127.0.0.1:4317" +authorization_env = "WAKEY_OTLP_AUTHORIZATION" +service_name = "wakey-control-plane" +json_logs = true +``` + +Agent configuration (replace the example with the collector's WireGuard address): + +```toml +[telemetry] +otlp_endpoint = "http://10.5.0.1:4317" +authorization_env = "WAKEY_OTLP_AUTHORIZATION" +service_name = "wakey-agent" +json_logs = false +``` + +Omit `otlp_endpoint` to disable exporting. Omit `authorization_env` for a receiver +that requires no authentication. Existing configs need no changes. +An empty agent service name uses `wakey-agent`; its enrolled agent ID becomes +the resource's `service.instance.id`, not a metric label. Telemetry config changes +require a daemon restart, not SIGHUP. Short-lived enroll/reload/observe commands +only log locally and do not initialize exporters. + +## Receiver and filtering + +The collector must enable OTLP receivers in **traces, logs, and metrics** pipelines. +Loopback is sufficient for CC on the same host/network namespace. The router +needs a collector listener reachable over WireGuard; do not expose an unauthenticated +receiver on the public Internet. This initial setup assumes plaintext inside that +tunnel. `json_logs` changes local output, +not whether OTLP logs are exported. + +## Credentials and services + +The OpenWrt agent uses `/etc/config/wakey`; see the +[UCI setup guide](openwrt-uci.md) for commands and permissions. +The CC uses a systemd environment file as described below. + +`authorization_env` names an environment variable containing the complete header: +`Basic ` or `Bearer `. With exporting enabled, +missing/empty/invalid values fail startup; credentials are never written to TOML +or included in error messages. Metadata is marked sensitive for Debug output. +Base64 is not encryption: use the WireGuard/SSH tunnel or another protected transport. + +The systemd CC reads the optional `/etc/wakey-control-plane/environment` file. + +Use this portable literal format, **without quotes or `export`**: + +```text +# Replace this placeholder with the actual encoded credential locally. +WAKEY_OTLP_AUTHORIZATION=Basic BASE64_OF_USERNAME_COLON_PASSWORD +RUST_LOG=info +``` + +Protect the file with owner `root:root` and mode `0600`. systemd's manager reads it +before switching to the service user. Use the simple format above, with LF endings +and no inline comments or multiline values. +An absent file is allowed; configuring auth without providing its variable is not. + +After changing the file, **restart** the service. On systemd, run `daemon-reload` +once after installing the updated unit. OpenWrt's HUP reload does not reload the +process environment. For a foreground WSL test, export the variable in that shell +and use `http://127.0.0.1:4317` with the SSH tunnel; do not put credentials in CLI args. + +`RUST_LOG` overrides the verbosity defaults for both local output and exports. +For example, `RUST_LOG=warn` suppresses informational command spans as well as logs. +Use `wakey_agent=info,wakey=info,wakey_core=info,opentelemetry=warn` for the agent and +`wakey_control_plane=info,wakey_core=info,opentelemetry=warn` for CC. Exporter warnings +remain local to prevent export-failure feedback loops. Increasing verbosity exports the +existing debug events too; inspect them before enabling it broadly. + +## Signals + +- `relay_command` on CC propagates W3C context in an optional command field. + `execute_command` on the agent adopts it; older messages/agents remain compatible. +- `collect_device_snapshot` and `store_device_snapshot` measure collection and + persistence separately. Snapshot traces are not yet linked across the wire. +- Terminal create/attach/close/resume spans describe lifecycle operations, not + output chunks. Trace-linked OTLP logs use the SDK's tracing-context bridge. + +Metrics are exported every 30 seconds and at orderly shutdown: + +| Metric | Meaning | +| --- | --- | +| `wakey.operations` | Completed command/snapshot attempts, labeled by operation and outcome | +| `wakey.operation.duration` | Corresponding duration histogram in seconds | +| `wakey.snapshot.devices` | Cumulative devices in snapshots successfully sent by the agent | +| `wakey.terminal.active` | Agent PTY workers currently alive | +| `wakey.terminal.input.bytes` | Input bytes received by the terminal relay | +| `wakey.terminal.output.bytes` | Output bytes read from the PTY, including while detached | +| `wakey.terminal.backpressure` | Relay queue-full encounters, not dropped output | + +Metric labels do not include request/session IDs, MACs, or IPs. New spans use +`skip_all` and explicitly select safe fields; terminal bytes and credentials are +not attached. Profiles, host CPU/RSS metrics, connected-agent gauges, and export-loss +dashboards are not part of this initial slice. + +## Failure policy + +Trace and log queues are capped at 256 records each, with export batches of 64 +and a 3-second request timeout. Export happens off the application path. On an +outage, telemetry can be dropped rather than accumulating unbounded router RAM; +terminal output follows its existing lossless backpressure policy independently. +The bound is a record count, not an absolute byte limit on arbitrary log messages. +Providers flush on orderly daemon exit (Ctrl-C/SIGTERM), on a blocking worker while +Tokio remains alive. Only completed spans are exported; this is not graceful draining +of every in-flight request. SIGKILL/power loss cannot flush. This is not a persistent +telemetry spool. + +## Validation + +```sh +cargo test -p wakey-core --features telemetry --test telemetry +``` + +The test starts a temporary gRPC collector and verifies all three signals, +command parent/child IDs, log correlation, and shutdown delivery without sleeps. diff --git a/scripts/init/openwrt/wakey b/scripts/init/openwrt/wakey index 1b15af9..60c6361 100644 --- a/scripts/init/openwrt/wakey +++ b/scripts/init/openwrt/wakey @@ -7,12 +7,26 @@ USE_PROCD=1 NAME=wakey BIN=/root/.bin/wakey-agent CONFIG=/etc/wakey-agent/config.toml +UCI_CONFIG=/etc/config/wakey start_service() { + local authorization='' log_filter='info' + if [ -f "$UCI_CONFIG" ]; then + config_load wakey || return 1 + config_get authorization main otlp_authorization + config_get log_filter main log_filter 'info' + fi procd_open_instance procd_set_param command "$BIN" serve --config "$CONFIG" procd_set_param file "$CONFIG" - # procd_set_param env RUST_LOG=wakey_agent=debug,wakey=debug + if [ -f "$UCI_CONFIG" ]; then + procd_append_param file "$UCI_CONFIG" + fi + procd_set_param env "RUST_LOG=$log_filter" + # Pass the complete header as one argument; never log its value. + if [ -n "$authorization" ]; then + procd_append_param env "WAKEY_OTLP_AUTHORIZATION=$authorization" + fi procd_set_param respawn 5 1 0 procd_set_param stdout 1 procd_set_param stderr 1 diff --git a/scripts/tests/test_openwrt_uci.sh b/scripts/tests/test_openwrt_uci.sh new file mode 100644 index 0000000..67d1190 --- /dev/null +++ b/scripts/tests/test_openwrt_uci.sh @@ -0,0 +1,69 @@ +#!/bin/sh +# Host-side contract test; OpenWrt's UCI and procd helpers are stubbed. +set -eu +ROOT="$(CDPATH='' cd -- "$(dirname -- "$0")/../.." && pwd)" +TMP="$(mktemp -d)" +trap 'rm -rf "$TMP"' EXIT HUP INT TERM + +config_load() { [ "$1" = wakey ]; loaded=1; return "$load_status"; } +config_get() { + [ "$2" = main ] + case "$3" in + otlp_authorization) authorization=$test_authorization ;; + log_filter) log_filter=${test_log_filter:-$4} ;; + *) return 1 ;; + esac +} +procd_open_instance() { opened=1; } +procd_close_instance() { :; } +procd_set_param() { + if [ "$1" = env ]; then + [ "$#" -eq 2 ] + received_filter=$2 + fi +} +procd_append_param() { + case "$1" in + env) [ "$#" -eq 2 ]; received_auth=$2 ;; + file) tracked_file=$2 ;; + esac +} +. "$ROOT/scripts/init/openwrt/wakey" +UCI_CONFIG="$TMP/wakey" +touch "$UCI_CONFIG" +load_status=0 +loaded=0 +opened=0 +test_authorization='Basic dGVzdDp0ZXN0==' +test_log_filter='wakey_agent=debug,opentelemetry=warn' +start_service +[ "$loaded" -eq 1 ] +[ "$opened" -eq 1 ] +[ "$received_auth" = "WAKEY_OTLP_AUTHORIZATION=$test_authorization" ] +[ "$received_filter" = "RUST_LOG=$test_log_filter" ] +[ "$tracked_file" = "$UCI_CONFIG" ] + +# Missing options retain the default filter and do not export empty credentials. +test_authorization='' +test_log_filter='' +received_auth='' +start_service +[ "$received_auth" = '' ] +[ "$received_filter" = 'RUST_LOG=info' ] + +load_status=1 +opened=0 +if start_service; then + echo 'failed UCI load was accepted' >&2 + exit 1 +fi +[ "$opened" -eq 0 ] + +UCI_CONFIG="$TMP/absent" +loaded=0 +tracked_file='' +start_service +[ "$loaded" -eq 0 ] +[ "$tracked_file" = '' ] +[ "$received_filter" = 'RUST_LOG=info' ] +echo 'PASS: OpenWrt UCI environment forwarding and optional configuration' diff --git a/wakey-agent/Cargo.toml b/wakey-agent/Cargo.toml index 5c1fe21..bcd901d 100644 --- a/wakey-agent/Cargo.toml +++ b/wakey-agent/Cargo.toml @@ -32,13 +32,7 @@ tokio = { version = "1", features = [ tokio-tungstenite = { version = "0.29", features = ["rustls-tls-webpki-roots"] } toml = "1.1" tracing = "0.1" -tracing-subscriber = { version = "0.3", features = [ - "env-filter", - "fmt", - "time", -] } -time = { version = "0.3", features = ["formatting", "local-offset"] } url = "2" vt100 = { package = "wakey-vt100", path = "../vendor/vt100", registry = "gitea", version = "=0.16.2-wakey.2" } wakey = { path = "..", registry = "gitea", version = "0" } -wakey-core = { path = "../wakey-core", registry = "gitea", version = "0" } +wakey-core = { path = "../wakey-core", registry = "gitea", version = "0", features = ["telemetry"] } diff --git a/wakey-agent/src/config.rs b/wakey-agent/src/config.rs index 3e26699..a68cb5e 100644 --- a/wakey-agent/src/config.rs +++ b/wakey-agent/src/config.rs @@ -42,6 +42,8 @@ pub struct AgentConfig { pub observation_store_path: PathBuf, #[serde(default)] pub terminal: TerminalConfig, + #[serde(default)] + pub telemetry: wakey_core::telemetry::TelemetryConfig, } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] @@ -88,6 +90,7 @@ pub static DEFAULT_CONFIG: LazyLock = LazyLock::new(|| AgentConfig mac_name_cache_path: default_mac_name_cache_path(), observation_store_path: default_observation_store_path(), terminal: TerminalConfig::default(), + telemetry: wakey_core::telemetry::TelemetryConfig::default(), }); impl fmt::Debug for AgentConfig { @@ -111,6 +114,7 @@ impl fmt::Debug for AgentConfig { .field("mac_name_cache_path", &self.mac_name_cache_path) .field("observation_store_path", &self.observation_store_path) .field("terminal", &self.terminal) + .field("telemetry", &self.telemetry) .finish() } } @@ -274,6 +278,12 @@ mod tests { dhcp_leases_path: "/tmp/test-dhcp.leases".into(), mac_name_cache_path: "/tmp/test-names.json".into(), observation_store_path: "/tmp/test-observations.json".into(), + telemetry: wakey_core::telemetry::TelemetryConfig { + otlp_endpoint: Some("http://10.0.0.1:4317".into()), + authorization_env: Some("WAKEY_OTLP_AUTHORIZATION".into()), + service_name: "router-agent".into(), + json_logs: true, + }, terminal: TerminalConfig { enabled: true, shell: "/bin/sh".into(), diff --git a/wakey-agent/src/enroll.rs b/wakey-agent/src/enroll.rs index b89125a..deb6582 100644 --- a/wakey-agent/src/enroll.rs +++ b/wakey-agent/src/enroll.rs @@ -151,6 +151,12 @@ mod tests { dhcp_leases_path: "/tmp/custom-dhcp.leases".into(), mac_name_cache_path: "/tmp/custom-names.json".into(), observation_store_path: "/tmp/custom-observations.json".into(), + telemetry: wakey_core::telemetry::TelemetryConfig { + otlp_endpoint: Some("http://10.0.0.1:4317".into()), + authorization_env: Some("WAKEY_OTLP_AUTHORIZATION".into()), + service_name: "router-agent".into(), + json_logs: true, + }, terminal: crate::config::TerminalConfig { enabled: true, shell: "/bin/ash".into(), diff --git a/wakey-agent/src/main.rs b/wakey-agent/src/main.rs index ae33be7..bea0064 100644 --- a/wakey-agent/src/main.rs +++ b/wakey-agent/src/main.rs @@ -15,9 +15,31 @@ use wakey_agent::protocol; #[tokio::main] async fn main() -> Result<()> { let cli = Cli::parse(); - tracing::init(cli.verbose); let global_config = cli.config.as_deref(); + // Only the long-lived daemon exports: hotplug/enroll invocations stay cheap. + let daemon_config = if let Command::Serve(args) = &cli.command { + Some(config::load_config(global_config.unwrap_or(&args.config))?) + } else { + None + }; + let local_only = wakey_core::telemetry::TelemetryConfig::default(); + let telemetry = tracing::init( + cli.verbose, + daemon_config + .as_ref() + .map(|c| &c.telemetry) + .unwrap_or(&local_only), + daemon_config.as_ref().map(|c| c.agent_id.as_str()), + )?; + + let result = run(cli).await; + tokio::task::spawn_blocking(move || telemetry.shutdown()).await?; + result +} + +async fn run(cli: Cli) -> Result<()> { + let global_config = cli.config.as_deref(); match cli.command { Command::Serve(mut args) => { if let Some(config) = global_config { diff --git a/wakey-agent/src/protocol.rs b/wakey-agent/src/protocol.rs index 77b42d5..2748a12 100644 --- a/wakey-agent/src/protocol.rs +++ b/wakey-agent/src/protocol.rs @@ -325,6 +325,8 @@ pub enum ServerMessage { Command { request_id: RequestId, command: AgentCommand, + #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")] + trace_context: std::collections::BTreeMap, }, SyncDeviceSnapshot, SyncTerminalSessions, @@ -354,6 +356,7 @@ mod tests { command: AgentCommand::Leases(LeasesRequest { include_state: true, }), + trace_context: Default::default(), }; let json = serde_json::to_string(&msg).expect("serialize"); @@ -367,6 +370,36 @@ mod tests { assert!(err.contains("must not be empty")); } + #[test] + fn old_commands_without_trace_context_remain_valid() { + let message: ServerMessage = serde_json::from_str(r#"{"type":"command","request_id":"req-old","command":{"kind":"leases","include_state":false}}"#).expect("old command"); + let ServerMessage::Command { trace_context, .. } = message else { + panic!("command expected") + }; + assert!(trace_context.is_empty()); + } + + #[test] + fn command_trace_context_roundtrips() { + let message = ServerMessage::Command { + request_id: RequestId::try_from("req-traced".to_string()).unwrap(), + command: AgentCommand::Leases(LeasesRequest { + include_state: false, + }), + trace_context: [( + "traceparent".into(), + "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01".into(), + )] + .into(), + }; + let ServerMessage::Command { trace_context, .. } = + serde_json::from_str(&serde_json::to_string(&message).unwrap()).unwrap() + else { + panic!("command expected") + }; + assert!(trace_context.contains_key("traceparent")); + } + #[test] fn terminal_id_deserialization_enforces_validation() { let valid: TerminalId = serde_json::from_str(r#""term-1""#).expect("valid terminal id"); diff --git a/wakey-agent/src/serve.rs b/wakey-agent/src/serve.rs index cea3e6c..52ebcf6 100644 --- a/wakey-agent/src/serve.rs +++ b/wakey-agent/src/serve.rs @@ -29,6 +29,8 @@ pub async fn serve(args: ServeArgs) -> Result<()> { { use tokio::signal::unix::{SignalKind, signal}; let mut hup = signal(SignalKind::hangup()).context("failed to install SIGHUP handler")?; + let mut terminate = + signal(SignalKind::terminate()).context("failed to install SIGTERM handler")?; loop { tokio::select! { @@ -37,9 +39,17 @@ pub async fn serve(args: ServeArgs) -> Result<()> { worker.abort(); break; } + _ = terminate.recv() => { + info!("SIGTERM received; shutting down wakey-agent"); + worker.abort(); + break; + } _ = hup.recv() => { match config::load_config(&args.config) { Ok(new_cfg) => { + if new_cfg.telemetry != cfg.telemetry { + warn!("telemetry configuration changed; restart the daemon to apply it"); + } cfg = new_cfg; info!(config_path = %args.config.display(), agent_id = %cfg.agent_id, "reload requested; restarting session with updated config"); worker.abort(); @@ -67,6 +77,8 @@ pub async fn serve(args: ServeArgs) -> Result<()> { worker.abort(); } + // Await the cancelled session task before flushing its telemetry. + let _ = worker.await; let _ = remove_pid_file(&pid_file); Ok(()) } diff --git a/wakey-agent/src/session.rs b/wakey-agent/src/session.rs index 865e5d2..73d90eb 100644 --- a/wakey-agent/src/session.rs +++ b/wakey-agent/src/session.rs @@ -195,11 +195,13 @@ where .await } +#[tracing::instrument(name = "collect_device_snapshot", skip_all)] async fn send_device_snapshot_ws(sink: &mut S, config: &AgentConfig) -> Result where S: SinkExt + Unpin, >::Error: std::error::Error + Send + Sync + 'static, { + let mut operation = wakey_core::telemetry::Operation::new("snapshot"); let req = crate::protocol::InventoryRequest { query: None, name: None, @@ -221,6 +223,11 @@ where ) .await?; debug!(agent_id = %config.agent_id, devices = count, "sent device snapshot over websocket"); + operation.outcome("ok"); + wakey_core::telemetry::count( + wakey_core::telemetry::Counter::SnapshotDevices, + count as u64, + ); Ok(count) } @@ -244,11 +251,16 @@ where ServerMessage::Command { request_id, command, + trace_context, } => { let kind = command_kind(&command); + let span = info_span!("execute_command", command = kind, request_id = %request_id); + wakey_core::telemetry::set_remote_parent(&span, &trace_context); + let mut operation = wakey_core::telemetry::Operation::new(kind); info!(request_id = %request_id, command = %kind, "received command from control-plane"); - match dispatch_command(command, config).await { + match tracing::Instrument::instrument(dispatch_command(command, config), span).await { Ok(result) => { + operation.outcome("ok"); info!(request_id = %request_id, command = %kind, "command execution completed"); send_command_result(sink, request_id, kind, result).await?; } diff --git a/wakey-agent/src/terminal.rs b/wakey-agent/src/terminal.rs index 619c24b..87bad61 100644 --- a/wakey-agent/src/terminal.rs +++ b/wakey-agent/src/terminal.rs @@ -80,6 +80,10 @@ impl RelayOutputBuffer { match sender.try_send(frame) { Ok(()) => {} Err(mpsc::error::TrySendError::Full(frame)) => { + wakey_core::telemetry::count( + wakey_core::telemetry::Counter::TerminalBackpressure, + 1, + ); self.pending.push_back(frame); } Err(mpsc::error::TrySendError::Closed(_)) => self.disconnect(), @@ -169,6 +173,7 @@ impl TerminalManager { ) } + #[tracing::instrument(name = "terminal_open", skip_all, fields(terminal_id = %terminal_id))] pub fn open( &self, config: &AgentConfig, @@ -214,6 +219,7 @@ impl TerminalManager { let active = Arc::downgrade(&self.active); let events = self.events.clone(); tokio::spawn(async move { + let _active_metric = wakey_core::telemetry::ActiveTerminal::new(); let result = run_terminal( &config, &terminal_id, @@ -238,6 +244,7 @@ impl TerminalManager { Ok(()) } + #[tracing::instrument(name = "terminal_close", skip_all, fields(terminal_id = %terminal_id))] pub fn close(&self, terminal_id: &TerminalId) -> bool { let mut active = self.active.lock().expect("terminal manager poisoned"); let Some(terminal) = active.get_mut(terminal_id.as_str()) else { @@ -253,6 +260,7 @@ impl TerminalManager { true } + #[tracing::instrument(name = "terminal_resume", skip_all, fields(terminal_id = %terminal_id))] pub fn resume(&self, terminal_id: &TerminalId, relay_token: String) -> Result<()> { let active = self.active.lock().expect("terminal manager poisoned"); let session = active @@ -453,6 +461,7 @@ async fn run_terminal( match read { Ok(0) => break, Ok(count) => { + wakey_core::telemetry::count(wakey_core::telemetry::Counter::TerminalOutputBytes, count as u64); terminal_state.process(&output[..count]); relay_output.queue_live(Message::Binary(output[..count].to_vec().into())); } @@ -575,6 +584,7 @@ where let Some(message) = incoming else { break; }; match message.context("terminal relay websocket receive failed")? { Message::Binary(bytes) => { + wakey_core::telemetry::count(wakey_core::telemetry::Counter::TerminalInputBytes, bytes.len() as u64); pending_input = Some(RelayInput::Binary { generation, bytes: bytes.to_vec(), diff --git a/wakey-agent/src/tracing.rs b/wakey-agent/src/tracing.rs index 8473799..c87af6d 100644 --- a/wakey-agent/src/tracing.rs +++ b/wakey-agent/src/tracing.rs @@ -1,28 +1,19 @@ -use time::UtcOffset; -use time::format_description::well_known::Rfc3339; -use tracing_subscriber::fmt::time::OffsetTime; -use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt}; +use anyhow::Result; +use wakey_core::telemetry::{TelemetryConfig, TelemetryGuard}; -pub fn init(verbose: u8) { - let filter = EnvFilter::try_from_default_env() - .or_else(|_| EnvFilter::try_new(default_filter(verbose))) - .expect("static tracing filter should parse"); - - tracing_subscriber::registry() - .with(filter) - .with(fmt::layer().with_timer(local_offset_timer())) - .init(); -} - -fn default_filter(verbose: u8) -> &'static str { - match verbose { - 0 => "wakey_agent=info,wakey=info", - 1 => "wakey_agent=debug,wakey=debug", - _ => "wakey_agent=trace,wakey=debug", - } -} - -fn local_offset_timer() -> OffsetTime { - let offset = UtcOffset::current_local_offset().unwrap_or(UtcOffset::UTC); - OffsetTime::new(offset, Rfc3339) +pub fn init( + verbose: u8, + config: &TelemetryConfig, + agent_id: Option<&str>, +) -> Result { + wakey_core::telemetry::init( + config, + "wakey-agent", + match verbose { + 0 => "wakey_agent=info,wakey=info,wakey_core=info,opentelemetry=warn", + 1 => "wakey_agent=debug,wakey=debug,wakey_core=info,opentelemetry=warn", + _ => "wakey_agent=trace,wakey=debug,wakey_core=info,opentelemetry=warn", + }, + agent_id, + ) } diff --git a/wakey-control-plane/Cargo.toml b/wakey-control-plane/Cargo.toml index 5b173b2..dc0c760 100644 --- a/wakey-control-plane/Cargo.toml +++ b/wakey-control-plane/Cargo.toml @@ -33,17 +33,6 @@ toml = "1.1" tokio = { workspace = true } tower-http = { version = "0.6.8", features = ["fs"] } tracing = { workspace = true } -tracing-opentelemetry = "0.32" -tracing-subscriber = { version = "0.3", features = [ - "env-filter", - "fmt", - "json", - "time", -] } -time = { version = "0.3", features = ["formatting", "local-offset"] } -opentelemetry = "0.31" -opentelemetry-otlp = { version = "0.31", features = ["grpc-tonic", "trace"] } -opentelemetry_sdk = { version = "0.31", features = ["rt-tokio"] } uuid = { workspace = true } wakey-agent = { path = "../wakey-agent", registry = "gitea", version = "0" } -wakey-core = { path = "../wakey-core", registry = "gitea", version = "0" } +wakey-core = { path = "../wakey-core", registry = "gitea", version = "0", features = ["telemetry"] } diff --git a/wakey-control-plane/src/api/commands.rs b/wakey-control-plane/src/api/commands.rs index 2f5b317..60ab262 100644 --- a/wakey-control-plane/src/api/commands.rs +++ b/wakey-control-plane/src/api/commands.rs @@ -4,7 +4,7 @@ use axum::http::StatusCode; use axum::response::IntoResponse; use serde::{Deserialize, Serialize}; use std::time::Instant; -use tracing::{info, info_span, warn}; +use tracing::{info, warn}; use uuid::Uuid; use wakey_agent::protocol::{ AgentCapability, AgentCapabilityOptions, AgentCommand, ErrorPayload, RequestId, ServerMessage, @@ -85,6 +85,7 @@ pub async fn run_command( } } +#[tracing::instrument(name = "relay_command", skip_all, fields(agent_id = %agent_id, request_id = tracing::field::Empty, command = tracing::field::Empty))] pub async fn relay_agent_command( state: &AppState, agent_id: &str, @@ -93,14 +94,11 @@ pub async fn relay_agent_command( ) -> Result { let request_id_string = format!("req-{}", Uuid::new_v4()); let command_kind = command_kind(&command); + tracing::Span::current() + .record("request_id", &request_id_string) + .record("command", command_kind); let started = Instant::now(); - let span = info_span!( - "relay_command", - agent_id = %agent_id, - request_id = %request_id_string, - command = %command_kind, - ); - let _span_guard = span.enter(); + let mut operation = wakey_core::telemetry::Operation::new(command_kind); let request_id = RequestId::try_from(request_id_string.clone()).map_err(|err| { ApiError::new( @@ -149,6 +147,7 @@ pub async fn relay_agent_command( if let Err(err) = tx.send(SessionEvent::Message(ServerMessage::Command { request_id, command, + trace_context: wakey_core::telemetry::current_trace_context(), })) { state.pending.lock().await.remove(&request_id_string); warn!(error = %err, "failed sending command to agent session"); @@ -249,6 +248,7 @@ pub async fn relay_agent_command( )); } Err(_) => { + operation.outcome("timeout"); state.pending.lock().await.remove(&request_id_string); warn!( timeout_ms = timeout.as_millis() as u64, @@ -276,6 +276,11 @@ pub async fn relay_agent_command( } }; + operation.outcome(if response.status == "ok" { + "ok" + } else { + "error" + }); Ok(response) } diff --git a/wakey-control-plane/src/api/terminals.rs b/wakey-control-plane/src/api/terminals.rs index 85a982e..a207a2a 100644 --- a/wakey-control-plane/src/api/terminals.rs +++ b/wakey-control-plane/src/api/terminals.rs @@ -49,6 +49,7 @@ pub struct TerminalSessionResponse { pub attachment_token: Option, } +#[tracing::instrument(name = "terminal_create", skip_all)] pub async fn create_terminal( State(state): State, Json(request): Json, @@ -183,6 +184,7 @@ pub async fn list_terminals(State(state): State) -> Json, Path(terminal_id): Path, @@ -210,6 +212,7 @@ pub async fn attach_terminal( })) } +#[tracing::instrument(name = "terminal_close", skip_all)] pub async fn close_terminal( State(state): State, Path(terminal_id): Path, diff --git a/wakey-control-plane/src/config/init.rs b/wakey-control-plane/src/config/init.rs index e818fde..18b495a 100644 --- a/wakey-control-plane/src/config/init.rs +++ b/wakey-control-plane/src/config/init.rs @@ -113,6 +113,9 @@ pub fn write_init_config(args: &InitConfigArgs) -> Result> { ui_dist_dir, bootstrap_enroll_tokens, telemetry: WritableTelemetry { + authorization_env: base_telemetry + .authorization_env + .or_else(|| defaults.telemetry.authorization_env.clone()), otlp_endpoint: args .telemetry_otlp_endpoint .clone() @@ -223,6 +226,7 @@ bootstrap_enroll_tokens = ["old-token"] [telemetry] otlp_endpoint = "http://otel.example" +authorization_env = "WAKEY_OTLP_AUTHORIZATION" service_name = "old-service" json_logs = true "#, @@ -261,6 +265,10 @@ json_logs = true Some("/tmp/wakey-base/old.sqlite3") ); assert_eq!(parsed["telemetry"]["json_logs"].as_bool(), Some(true)); + assert_eq!( + parsed["telemetry"]["authorization_env"].as_str(), + Some("WAKEY_OTLP_AUTHORIZATION") + ); assert_eq!( parsed["telemetry"]["service_name"].as_str(), Some("old-service") diff --git a/wakey-control-plane/src/config/resolve.rs b/wakey-control-plane/src/config/resolve.rs index 63e32f1..500e633 100644 --- a/wakey-control-plane/src/config/resolve.rs +++ b/wakey-control-plane/src/config/resolve.rs @@ -231,6 +231,7 @@ fn resolve_admin_public_url( fn resolve_telemetry(file: Option) -> TelemetryConfig { let mut out = TelemetryConfig::default(); if let Some(file) = file { + out.authorization_env = file.authorization_env; if let Some(endpoint) = file.otlp_endpoint { let trimmed = endpoint.trim().to_string(); if !trimmed.is_empty() { diff --git a/wakey-control-plane/src/config/types.rs b/wakey-control-plane/src/config/types.rs index c2a9287..2396c2f 100644 --- a/wakey-control-plane/src/config/types.rs +++ b/wakey-control-plane/src/config/types.rs @@ -23,6 +23,7 @@ pub struct DaemonConfig { #[derive(Debug, Clone)] pub struct TelemetryConfig { pub otlp_endpoint: Option, + pub authorization_env: Option, pub service_name: String, pub json_logs: bool, } @@ -32,6 +33,7 @@ impl Default for TelemetryConfig { let defaults = &DEFAULT_CONFIG.telemetry; Self { otlp_endpoint: defaults.otlp_endpoint.clone(), + authorization_env: defaults.authorization_env.clone(), service_name: defaults.service_name.clone(), json_logs: defaults.json_logs, } @@ -57,6 +59,7 @@ pub(crate) struct FileConfig { #[derive(Debug, Deserialize, Default)] pub(crate) struct FileTelemetryConfig { pub(crate) otlp_endpoint: Option, + pub(crate) authorization_env: Option, pub(crate) service_name: Option, pub(crate) json_logs: Option, } @@ -81,6 +84,8 @@ pub(crate) struct WritableConfig { pub(crate) struct WritableTelemetry { #[serde(skip_serializing_if = "Option::is_none")] pub(crate) otlp_endpoint: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) authorization_env: Option, pub(crate) service_name: String, pub(crate) json_logs: bool, } @@ -98,6 +103,7 @@ pub(crate) static DEFAULT_CONFIG: LazyLock = LazyLock::new(|| Wr bootstrap_enroll_tokens: Vec::new(), telemetry: WritableTelemetry { otlp_endpoint: None, + authorization_env: None, service_name: "wakey-control-plane".to_string(), json_logs: false, }, diff --git a/wakey-control-plane/src/main.rs b/wakey-control-plane/src/main.rs index cebef7b..8d6e7d2 100644 --- a/wakey-control-plane/src/main.rs +++ b/wakey-control-plane/src/main.rs @@ -28,8 +28,10 @@ async fn main() -> Result<()> { } } let daemon = config::DaemonConfig::from_serve_args(&args)?; - tracing::init(cli.verbose, &daemon.telemetry)?; - runtime::serve(daemon).await + let telemetry = tracing::init(cli.verbose, &daemon.telemetry)?; + let result = runtime::serve(daemon).await; + tokio::task::spawn_blocking(move || telemetry.shutdown()).await?; + result } Command::InitConfig(args) => { if let Some(path) = config::write_init_config(&args)? { diff --git a/wakey-control-plane/src/runtime/mod.rs b/wakey-control-plane/src/runtime/mod.rs index 44a749e..02208e6 100644 --- a/wakey-control-plane/src/runtime/mod.rs +++ b/wakey-control-plane/src/runtime/mod.rs @@ -213,6 +213,8 @@ pub async fn serve(daemon: config::DaemonConfig) -> Result<()> { use tokio::signal::unix::{SignalKind, signal}; let mut server = server; let mut hup = signal(SignalKind::hangup()).context("failed to install SIGHUP handler")?; + let mut terminate = + signal(SignalKind::terminate()).context("failed to install SIGTERM handler")?; let mut gc_tick = tokio::time::interval(Duration::from_secs(300)); gc_tick.set_missed_tick_behavior(MissedTickBehavior::Skip); @@ -224,6 +226,11 @@ pub async fn serve(daemon: config::DaemonConfig) -> Result<()> { server.abort(); break; } + _ = terminate.recv() => { + info!("SIGTERM received; shutting down control-plane"); + server.abort(); + break; + } _ = hup.recv() => { match app_state.store.reload_from_disk().await { Ok(()) => info!("reloaded state from disk"), diff --git a/wakey-control-plane/src/tracing.rs b/wakey-control-plane/src/tracing.rs index 50f9ceb..29e6c17 100644 --- a/wakey-control-plane/src/tracing.rs +++ b/wakey-control-plane/src/tracing.rs @@ -1,99 +1,22 @@ -use anyhow::{Context, Result}; -use opentelemetry::global; -use opentelemetry::trace::TracerProvider as _; -use opentelemetry_otlp::WithExportConfig; -use opentelemetry_sdk::Resource; -use opentelemetry_sdk::trace::{SdkTracerProvider, Tracer}; -use time::UtcOffset; -use time::format_description::well_known::Rfc3339; -use tracing_subscriber::fmt::time::OffsetTime; -use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt}; - use crate::config::TelemetryConfig; +use anyhow::Result; +pub use wakey_core::telemetry::TelemetryGuard; -pub fn init(verbose: u8, telemetry: &TelemetryConfig) -> Result<()> { - let filter = EnvFilter::try_from_default_env() - .or_else(|_| EnvFilter::try_new(default_filter(verbose))) - .expect("static tracing filter should parse"); - - let otel = build_otel_layer(telemetry)?; - - if telemetry.json_logs { - if let Some(otel_layer) = otel { - tracing_subscriber::registry() - .with(otel_layer) - .with(filter) - .with(fmt::layer().json().with_timer(local_offset_timer())) - .init(); - tracing::info!(endpoint = ?telemetry.otlp_endpoint, json_logs = telemetry.json_logs, "tracing initialized with otlp exporter"); - } else { - tracing_subscriber::registry() - .with(filter) - .with(fmt::layer().json().with_timer(local_offset_timer())) - .init(); - tracing::info!( - json_logs = telemetry.json_logs, - "tracing initialized without otlp exporter" - ); - } - } else if let Some(otel_layer) = otel { - tracing_subscriber::registry() - .with(otel_layer) - .with(filter) - .with(fmt::layer().with_timer(local_offset_timer())) - .init(); - tracing::info!(endpoint = ?telemetry.otlp_endpoint, json_logs = telemetry.json_logs, "tracing initialized with otlp exporter"); - } else { - tracing_subscriber::registry() - .with(filter) - .with(fmt::layer().with_timer(local_offset_timer())) - .init(); - tracing::info!( - json_logs = telemetry.json_logs, - "tracing initialized without otlp exporter" - ); - } - - Ok(()) -} - -fn build_otel_layer( - telemetry: &TelemetryConfig, -) -> Result>> -{ - let Some(endpoint) = telemetry.otlp_endpoint.as_deref() else { - return Ok(None); +pub fn init(verbose: u8, telemetry: &TelemetryConfig) -> Result { + let config = wakey_core::telemetry::TelemetryConfig { + otlp_endpoint: telemetry.otlp_endpoint.clone(), + authorization_env: telemetry.authorization_env.clone(), + service_name: telemetry.service_name.clone(), + json_logs: telemetry.json_logs, }; - - let exporter = opentelemetry_otlp::SpanExporter::builder() - .with_tonic() - .with_endpoint(endpoint) - .build() - .context("failed to build OTLP span exporter")?; - - let provider = SdkTracerProvider::builder() - .with_batch_exporter(exporter) - .with_resource( - Resource::builder_empty() - .with_service_name(telemetry.service_name.clone()) - .build(), - ) - .build(); - - let tracer = provider.tracer(telemetry.service_name.clone()); - global::set_tracer_provider(provider); - Ok(Some(tracing_opentelemetry::layer().with_tracer(tracer))) -} - -fn default_filter(verbose: u8) -> &'static str { - match verbose { - 0 => "wakey_control_plane=info", - 1 => "wakey_control_plane=debug", - _ => "wakey_control_plane=trace", - } -} - -fn local_offset_timer() -> OffsetTime { - let offset = UtcOffset::current_local_offset().unwrap_or(UtcOffset::UTC); - OffsetTime::new(offset, Rfc3339) + wakey_core::telemetry::init( + &config, + "wakey-control-plane", + match verbose { + 0 => "wakey_control_plane=info,wakey_core=info,opentelemetry=warn", + 1 => "wakey_control_plane=debug,wakey_core=info,opentelemetry=warn", + _ => "wakey_control_plane=trace,wakey_core=info,opentelemetry=warn", + }, + None, + ) } diff --git a/wakey-control-plane/src/ws.rs b/wakey-control-plane/src/ws.rs index 3adb01f..a227d2d 100644 --- a/wakey-control-plane/src/ws.rs +++ b/wakey-control-plane/src/ws.rs @@ -309,12 +309,18 @@ async fn process_agent_text( anyhow::bail!("device_snapshot for unauthenticated or mismatched agent"); } ensure_current_session(state, &agent_id, connection_id).await?; - match state - .store - .replace_agent_device_snapshot(&agent_id, &devices) - .await + let mut operation = wakey_core::telemetry::Operation::new("snapshot_store"); + let span = tracing::info_span!("store_device_snapshot", agent_id = %agent_id, devices = devices.len()); + match tracing::Instrument::instrument( + state + .store + .replace_agent_device_snapshot(&agent_id, &devices), + span, + ) + .await { Ok(accepted) => { + operation.outcome("ok"); debug!(agent_id = %agent_id, accepted, "agent websocket device snapshot accepted"); } Err(err) => { diff --git a/wakey-core/Cargo.toml b/wakey-core/Cargo.toml index 2dd0206..0c45d3d 100644 --- a/wakey-core/Cargo.toml +++ b/wakey-core/Cargo.toml @@ -5,11 +5,28 @@ edition = "2024" publish = ["gitea"] [dependencies] +anyhow = { workspace = true, optional = true } +opentelemetry = { version = "0.31", optional = true } +opentelemetry-otlp = { version = "0.31", default-features = false, features = ["grpc-tonic", "trace", "logs", "metrics"], optional = true } +opentelemetry_sdk = { version = "0.31", optional = true } +opentelemetry-appender-tracing = { version = "0.31", features = ["experimental_use_tracing_span_context"], optional = true } +tracing = { workspace = true, optional = true } +tracing-opentelemetry = { version = "0.32", optional = true } +tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "json", "time"], optional = true } +time = { version = "0.3", features = ["formatting", "local-offset"], optional = true } +tonic = { version = "0.14", default-features = false, optional = true } macaddr = { workspace = true } serde = { workspace = true } serde_with = { version = "3", features = ["json"] } strum = { version = "0", features = ["derive", "strum_macros"] } thiserror = "2" +[features] +telemetry = ["dep:anyhow", "dep:opentelemetry", "dep:opentelemetry-otlp", "dep:opentelemetry_sdk", "dep:opentelemetry-appender-tracing", "dep:tracing", "dep:tracing-opentelemetry", "dep:tracing-subscriber", "dep:time", "dep:tonic"] + [dev-dependencies] serde_json.workspace = true +tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "sync", "time"] } +tokio-stream = { version = "0.1", features = ["net"] } +tonic = "0.14" +opentelemetry-proto = { version = "0.31", default-features = false, features = ["gen-tonic", "trace", "logs", "metrics"] } diff --git a/wakey-core/src/lib.rs b/wakey-core/src/lib.rs index 3923a7a..c66e429 100644 --- a/wakey-core/src/lib.rs +++ b/wakey-core/src/lib.rs @@ -3,4 +3,7 @@ pub mod model; pub mod parse; +#[cfg(feature = "telemetry")] +pub mod telemetry; + pub use model::*; diff --git a/wakey-core/src/telemetry.rs b/wakey-core/src/telemetry.rs new file mode 100644 index 0000000..b666294 --- /dev/null +++ b/wakey-core/src/telemetry.rs @@ -0,0 +1,409 @@ +//! Shared, opt-in OTLP setup. Local logging remains available without a collector. + +use std::collections::BTreeMap; +use std::sync::OnceLock; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; +use opentelemetry::{ + KeyValue, global, + trace::{TraceContextExt as _, TracerProvider as _}, +}; +use opentelemetry_otlp::WithExportConfig; +use opentelemetry_otlp::WithTonicConfig; +use opentelemetry_sdk::{ + Resource, logs::SdkLoggerProvider, metrics::SdkMeterProvider, trace::SdkTracerProvider, +}; +use serde::{Deserialize, Serialize}; +use tracing_opentelemetry::OpenTelemetrySpanExt; +use tracing_subscriber::{EnvFilter, Layer, fmt, layer::SubscriberExt, util::SubscriberInitExt}; + +/// A single gRPC receiver accepts all three signals. No endpoint means local-only. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +pub struct TelemetryConfig { + pub otlp_endpoint: Option, + /// Name of the environment variable holding the complete Authorization header. + #[serde(skip_serializing_if = "Option::is_none")] + pub authorization_env: Option, + pub service_name: String, + pub json_logs: bool, +} + +/// Owns providers until all instrumented work has stopped; flush off the async workers. +#[derive(Default)] +pub struct TelemetryGuard { + traces: Option, + logs: Option, + metrics: Option, +} + +impl TelemetryGuard { + pub fn shutdown(self) { + // Export failures must not turn an otherwise successful daemon exit into a failure. + if let Some(provider) = self.metrics { + let _ = provider.shutdown(); + } + if let Some(provider) = self.logs { + let _ = provider.shutdown(); + } + if let Some(provider) = self.traces { + let _ = provider.shutdown(); + } + } +} + +pub fn init( + config: &TelemetryConfig, + service: &str, + default_filter: &str, + instance: Option<&str>, +) -> Result { + let filter = + EnvFilter::try_from_default_env().or_else(|_| EnvFilter::try_new(default_filter))?; + let service = if config.service_name.is_empty() { + service + } else { + &config.service_name + }; + let mut guard = TelemetryGuard::default(); + if let Some(endpoint) = config + .otlp_endpoint + .as_deref() + .filter(|s| !s.trim().is_empty()) + { + let metadata = authorization_metadata(config.authorization_env.as_deref(), |name| { + std::env::var(name).ok() + })?; + let mut resource = Resource::builder_empty().with_service_name(service.to_owned()); + if let Some(instance) = instance { + resource = + resource.with_attribute(KeyValue::new("service.instance.id", instance.to_owned())); + } + let resource = resource.build(); + // Small fixed queues are intentional: a disconnected collector cannot grow router RAM. + let trace_exporter = opentelemetry_otlp::SpanExporter::builder() + .with_tonic() + .with_metadata(metadata.clone()) + .with_endpoint(endpoint) + .with_timeout(Duration::from_secs(3)) + .build() + .context("OTLP trace exporter")?; + let trace_processor = opentelemetry_sdk::trace::BatchSpanProcessor::builder(trace_exporter) + .with_batch_config( + opentelemetry_sdk::trace::BatchConfigBuilder::default() + .with_max_queue_size(256) + .with_max_export_batch_size(64) + .build(), + ) + .build(); + let traces = SdkTracerProvider::builder() + .with_span_processor(trace_processor) + .with_resource(resource.clone()) + .build(); + let log_exporter = opentelemetry_otlp::LogExporter::builder() + .with_tonic() + .with_metadata(metadata.clone()) + .with_endpoint(endpoint) + .with_timeout(Duration::from_secs(3)) + .build() + .context("OTLP log exporter")?; + let log_processor = opentelemetry_sdk::logs::BatchLogProcessor::builder(log_exporter) + .with_batch_config( + opentelemetry_sdk::logs::BatchConfigBuilder::default() + .with_max_queue_size(256) + .with_max_export_batch_size(64) + .build(), + ) + .build(); + let logs = SdkLoggerProvider::builder() + .with_log_processor(log_processor) + .with_resource(resource.clone()) + .build(); + let metric_exporter = opentelemetry_otlp::MetricExporter::builder() + .with_tonic() + .with_metadata(metadata) + .with_endpoint(endpoint) + .with_timeout(Duration::from_secs(3)) + .build() + .context("OTLP metric exporter")?; + let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(metric_exporter) + .with_interval(Duration::from_secs(30)) + .build(); + let metrics = SdkMeterProvider::builder() + .with_reader(reader) + .with_resource(resource) + .build(); + global::set_tracer_provider(traces.clone()); + global::set_meter_provider(metrics.clone()); + let _ = METRICS.set(Instruments::new()); + let otel = tracing_opentelemetry::layer().with_tracer(traces.tracer(service.to_owned())); + // Exporter failures must remain local, never feed back into the failing exporter. + let logs_layer = + opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge::new(&logs) + .with_filter(tracing_subscriber::filter::filter_fn(|metadata| { + !metadata.target().starts_with("opentelemetry") + })); + let subscriber = tracing_subscriber::registry() + .with(filter) + .with(otel) + .with(logs_layer); + if config.json_logs { + subscriber + .with(fmt::layer().json().with_timer(local_timer())) + .try_init()?; + } else { + subscriber + .with(fmt::layer().with_timer(local_timer())) + .try_init()?; + } + guard = TelemetryGuard { + traces: Some(traces), + logs: Some(logs), + metrics: Some(metrics), + }; + } else if config.json_logs { + tracing_subscriber::registry() + .with(filter) + .with(fmt::layer().json().with_timer(local_timer())) + .try_init()?; + } else { + tracing_subscriber::registry() + .with(filter) + .with(fmt::layer().with_timer(local_timer())) + .try_init()?; + } + Ok(guard) +} + +fn authorization_metadata( + name: Option<&str>, + lookup: impl FnOnce(&str) -> Option, +) -> Result { + let mut metadata = tonic::metadata::MetadataMap::new(); + let Some(name) = name else { + return Ok(metadata); + }; + anyhow::ensure!( + !name.is_empty() + && name.bytes().all(|c| c.is_ascii_alphanumeric() || c == b'_') + && !name.as_bytes()[0].is_ascii_digit(), + "telemetry authorization_env must be a valid environment variable name" + ); + let raw = lookup(name).with_context(|| { + format!("telemetry authorization environment variable {name} is missing or not UTF-8") + })?; + let header = raw.trim(); + anyhow::ensure!( + !header.is_empty(), + "telemetry authorization environment variable {name} is empty" + ); + let valid_scheme = header.split_once(' ').is_some_and(|(scheme, credentials)| { + (scheme.eq_ignore_ascii_case("basic") || scheme.eq_ignore_ascii_case("bearer")) + && !credentials.trim().is_empty() + }); + anyhow::ensure!( + valid_scheme && !raw.contains(['\r', '\n']), + "telemetry authorization environment variable {name} must contain a single Basic or Bearer header" + ); + // Never include the supplied value/parser error in diagnostics. Sensitive + // metadata also hides it if a dependency Debug-prints an outgoing request. + let mut value = tonic::metadata::MetadataValue::try_from(header).map_err(|_| { + anyhow::anyhow!("invalid authorization metadata in environment variable {name}") + })?; + value.set_sensitive(true); + metadata.insert("authorization", value); + Ok(metadata) +} + +fn local_timer() -> fmt::time::OffsetTime { + fmt::time::OffsetTime::new( + time::UtcOffset::current_local_offset().unwrap_or(time::UtcOffset::UTC), + time::format_description::well_known::Rfc3339, + ) +} + +// Construct instruments once, after the real provider exists. Disabled telemetry +// therefore does no per-PTY-read allocations or instrument registration. +static METRICS: OnceLock = OnceLock::new(); +struct Instruments { + operations: opentelemetry::metrics::Counter, + duration: opentelemetry::metrics::Histogram, + snapshot_devices: opentelemetry::metrics::Counter, + terminal_active: opentelemetry::metrics::UpDownCounter, + terminal_input: opentelemetry::metrics::Counter, + terminal_output: opentelemetry::metrics::Counter, + terminal_backpressure: opentelemetry::metrics::Counter, +} +impl Instruments { + fn new() -> Self { + let meter = global::meter("wakey"); + Self { + operations: meter.u64_counter("wakey.operations").build(), + duration: meter + .f64_histogram("wakey.operation.duration") + .with_unit("s") + .build(), + snapshot_devices: meter.u64_counter("wakey.snapshot.devices").build(), + terminal_active: meter.i64_up_down_counter("wakey.terminal.active").build(), + terminal_input: meter + .u64_counter("wakey.terminal.input.bytes") + .with_unit("By") + .build(), + terminal_output: meter + .u64_counter("wakey.terminal.output.bytes") + .with_unit("By") + .build(), + terminal_backpressure: meter.u64_counter("wakey.terminal.backpressure").build(), + } + } +} + +/// W3C context only, never arbitrary application fields or credentials. +pub fn current_trace_context() -> BTreeMap { + use opentelemetry::propagation::TextMapPropagator; + let mut carrier = std::collections::HashMap::new(); + opentelemetry_sdk::propagation::TraceContextPropagator::new() + .inject_context(&tracing::Span::current().context(), &mut carrier); + carrier.into_iter().collect() +} + +pub fn set_remote_parent(span: &tracing::Span, carrier: &BTreeMap) { + use opentelemetry::propagation::TextMapPropagator; + let carrier: std::collections::HashMap = carrier + .iter() + .filter(|(key, value)| { + matches!(key.as_str(), "traceparent" | "tracestate") && value.len() <= 512 + }) + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + let context = opentelemetry_sdk::propagation::TraceContextPropagator::new().extract(&carrier); + if context.has_active_span() { + let _ = span.set_parent(context); + } +} + +/// Records completion on every return path, without high-cardinality identifiers. +pub struct Operation { + started: Instant, + name: &'static str, + outcome: &'static str, +} + +impl Operation { + pub fn new(name: &'static str) -> Self { + Self { + started: Instant::now(), + name, + outcome: "error", + } + } + pub fn outcome(&mut self, outcome: &'static str) { + self.outcome = outcome; + } +} + +impl Drop for Operation { + fn drop(&mut self) { + let Some(metrics) = METRICS.get() else { + return; + }; + let labels = [ + KeyValue::new("operation", self.name), + KeyValue::new("outcome", self.outcome), + ]; + metrics.operations.add(1, &labels); + metrics + .duration + .record(self.started.elapsed().as_secs_f64(), &labels); + } +} + +pub enum Counter { + SnapshotDevices, + TerminalInputBytes, + TerminalOutputBytes, + TerminalBackpressure, +} + +pub fn count(counter: Counter, value: u64) { + let Some(metrics) = METRICS.get() else { + return; + }; + let counter = match counter { + Counter::SnapshotDevices => &metrics.snapshot_devices, + Counter::TerminalInputBytes => &metrics.terminal_input, + Counter::TerminalOutputBytes => &metrics.terminal_output, + Counter::TerminalBackpressure => &metrics.terminal_backpressure, + }; + counter.add(value, &[]); +} + +/// An active session contributes until its worker finishes or is cancelled. +pub struct ActiveTerminal; +impl ActiveTerminal { + pub fn new() -> Self { + if let Some(metrics) = METRICS.get() { + metrics.terminal_active.add(1, &[]); + } + Self + } +} +impl Default for ActiveTerminal { + fn default() -> Self { + Self::new() + } +} +impl Drop for ActiveTerminal { + fn drop(&mut self) { + if let Some(metrics) = METRICS.get() { + metrics.terminal_active.add(-1, &[]); + } + } +} + +#[cfg(test)] +mod auth_tests { + use super::*; + + #[test] + fn authorization_supports_both_schemes_and_redacts_debug() { + for header in ["Basic dGVzdDp0ZXN0", "Bearer test-token"] { + let metadata = + authorization_metadata(Some("OTLP_AUTH"), |_| Some(header.into())).unwrap(); + assert_eq!( + metadata.get("authorization").unwrap().to_str().unwrap(), + header + ); + assert!(!format!("{metadata:?}").contains(header)); + } + assert!( + authorization_metadata(None, |_| panic!("no credential lookup")) + .unwrap() + .is_empty() + ); + } + + #[test] + fn missing_empty_and_invalid_auth_fail_without_echoing_secrets() { + for value in [ + None, + Some(""), + Some(" "), + Some("Basic "), + Some("Digest private-secret"), + Some("Basic private-secret\r\n"), + Some("Bearer private-secret\u{7f}"), + ] { + let error = authorization_metadata(Some("OTLP_AUTH"), |_| value.map(str::to_owned)) + .unwrap_err(); + assert!(!format!("{error:#}").contains("private-secret")); + } + assert!( + authorization_metadata(Some("BAD=NAME"), |_| panic!( + "invalid name must not be looked up" + )) + .is_err() + ); + } +} diff --git a/wakey-core/tests/telemetry.rs b/wakey-core/tests/telemetry.rs new file mode 100644 index 0000000..125e5c5 --- /dev/null +++ b/wakey-core/tests/telemetry.rs @@ -0,0 +1,196 @@ +#![cfg(feature = "telemetry")] + +use opentelemetry_proto::tonic::collector::{ + logs::v1 as logs, metrics::v1 as metrics, trace::v1 as traces, +}; +use std::sync::{Arc, Mutex}; +use tonic::{Request, Response, Status}; +use wakey_core::telemetry::{self, TelemetryConfig}; + +#[derive(Clone, Default)] +struct Collector { + traces: Arc>>, + logs: Arc>>, + metrics: Arc>>, +} + +fn check_auth(request: &Request) -> Result<(), Status> { + if request + .metadata() + .get("authorization") + .and_then(|v| v.to_str().ok()) + != Some("Basic dGVzdDp0ZXN0") + { + return Err(Status::unauthenticated( + "test receiver requires authorization", + )); + } + Ok(()) +} + +#[tonic::async_trait] +impl traces::trace_service_server::TraceService for Collector { + async fn export( + &self, + request: Request, + ) -> Result, Status> { + check_auth(&request)?; + self.traces.lock().unwrap().push(request.into_inner()); + Ok(Response::new(Default::default())) + } +} +#[tonic::async_trait] +impl logs::logs_service_server::LogsService for Collector { + async fn export( + &self, + request: Request, + ) -> Result, Status> { + check_auth(&request)?; + self.logs.lock().unwrap().push(request.into_inner()); + Ok(Response::new(Default::default())) + } +} +#[tonic::async_trait] +impl metrics::metrics_service_server::MetricsService for Collector { + async fn export( + &self, + request: Request, + ) -> Result, Status> { + check_auth(&request)?; + self.metrics.lock().unwrap().push(request.into_inner()); + Ok(Response::new(Default::default())) + } +} + +/// Shutdown is the synchronization point: no sleeps or export-interval polling. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exports_all_signals_and_preserves_remote_command_parent() { + // Exercise the real global subscriber without mutating this process's environment + // or inheriting a developer's RUST_LOG=warn/off setting. + if std::env::var_os("WAKEY_OTLP_TEST_CHILD").is_none() { + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "exports_all_signals_and_preserves_remote_command_parent", + "--nocapture", + ]) + .env("WAKEY_OTLP_TEST_CHILD", "1") + .env("RUST_LOG", "info") + .env("WAKEY_OTLP_TEST_AUTH", "Basic dGVzdDp0ZXN0") + .status() + .unwrap(); + assert!(status.success()); + return; + } + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let collector = Collector::default(); + let (stop_tx, stop_rx) = tokio::sync::oneshot::channel(); + let server = tonic::transport::Server::builder() + .add_service(traces::trace_service_server::TraceServiceServer::new( + collector.clone(), + )) + .add_service(logs::logs_service_server::LogsServiceServer::new( + collector.clone(), + )) + .add_service(metrics::metrics_service_server::MetricsServiceServer::new( + collector.clone(), + )) + .serve_with_incoming_shutdown( + tokio_stream::wrappers::TcpListenerStream::new(listener), + async { + let _ = stop_rx.await; + }, + ); + let server = tokio::spawn(server); + let guard = telemetry::init( + &TelemetryConfig { + otlp_endpoint: Some(endpoint), + authorization_env: Some("WAKEY_OTLP_TEST_AUTH".into()), + service_name: "wakey-test".into(), + json_logs: false, + }, + "unused", + "info", + Some("test-agent"), + ) + .unwrap(); + + let parent = tracing::info_span!("relay_command"); + let context = parent.in_scope(telemetry::current_trace_context); + assert!(context.contains_key("traceparent")); + let child = tracing::info_span!("execute_command"); + telemetry::set_remote_parent(&child, &context); + child.in_scope(|| { + tracing::info!(command = "inventory", "command complete"); + let mut operation = telemetry::Operation::new("inventory"); + operation.outcome("ok"); + telemetry::count(telemetry::Counter::SnapshotDevices, 3); + let _active = telemetry::ActiveTerminal::new(); + }); + drop(child); + drop(parent); + tracing::warn!(target: "opentelemetry_sdk", "synthetic exporter failure remains local"); + let invalid_parent = tracing::info_span!("invalid_remote_parent"); + telemetry::set_remote_parent( + &invalid_parent, + &[("traceparent".into(), "malformed".into())].into(), + ); + drop(invalid_parent); + tokio::task::spawn_blocking(move || guard.shutdown()) + .await + .unwrap(); + let _ = stop_tx.send(()); + server.await.unwrap().unwrap(); + + let traces = collector.traces.lock().unwrap(); + let spans: Vec<_> = traces + .iter() + .flat_map(|r| &r.resource_spans) + .flat_map(|r| &r.scope_spans) + .flat_map(|s| &s.spans) + .collect(); + let parent = spans.iter().find(|s| s.name == "relay_command").unwrap(); + let child = spans.iter().find(|s| s.name == "execute_command").unwrap(); + assert_eq!(parent.trace_id, child.trace_id); + assert_eq!(parent.span_id, child.parent_span_id); + let invalid_parent = spans + .iter() + .find(|s| s.name == "invalid_remote_parent") + .unwrap(); + assert!(invalid_parent.parent_span_id.is_empty()); + assert_ne!(parent.trace_id, invalid_parent.trace_id); + let logs = collector.logs.lock().unwrap(); + let records: Vec<_> = logs + .iter() + .flat_map(|r| &r.resource_logs) + .flat_map(|r| &r.scope_logs) + .flat_map(|s| &s.log_records) + .collect(); + assert!( + records + .iter() + .any(|r| r.trace_id == child.trace_id && r.span_id == child.span_id) + ); + assert!( + !records + .iter() + .any(|r| format!("{:?}", r.body).contains("synthetic exporter failure")) + ); + let metrics = collector.metrics.lock().unwrap(); + let names: Vec<_> = metrics + .iter() + .flat_map(|r| &r.resource_metrics) + .flat_map(|r| &r.scope_metrics) + .flat_map(|s| &s.metrics) + .map(|m| m.name.as_str()) + .collect(); + for name in [ + "wakey.operations", + "wakey.operation.duration", + "wakey.snapshot.devices", + "wakey.terminal.active", + ] { + assert!(names.contains(&name), "missing {name}"); + } +}