lda/wakey
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 18 Wiki Activity

logging and config sweep

Browse Source
This commit is contained in:
lda
2026-04-11 18:02:12 +07:00 Unverified
parent ce1c0c0558
commit 78b12ad0ee
16 changed files with 839 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+284
View File
@@ -91,6 +91,17 @@ version = "1.0.102"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
[[package]]
name = "async-trait"
version = "0.1.89"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "atomic-waker"
version = "1.1.2"
@@ -464,6 +475,12 @@ version = "1.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
[[package]]
name = "either"
version = "1.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
[[package]]
name = "equivalent"
version = "1.0.2"
@@ -496,6 +513,12 @@ version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
[[package]]
name = "fnv"
version = "1.0.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
[[package]]
name = "foldhash"
version = "0.1.5"
@@ -655,6 +678,25 @@ version = "0.32.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7"
[[package]]
name = "h2"
version = "0.4.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54"
dependencies = [
"atomic-waker",
"bytes",
"fnv",
"futures-core",
"futures-sink",
"http",
"indexmap 2.13.1",
"slab",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "hashbrown"
version = "0.12.3"
@@ -743,6 +785,7 @@ dependencies = [
"bytes",
"futures-channel",
"futures-core",
"h2",
"http",
"http-body",
"httparse",
@@ -771,6 +814,19 @@ dependencies = [
"webpki-roots 1.0.6",
]
[[package]]
name = "hyper-timeout"
version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
dependencies = [
"hyper",
"hyper-util",
"pin-project-lite",
"tokio",
"tower-service",
]
[[package]]
name = "hyper-util"
version = "0.1.20"
@@ -984,6 +1040,15 @@ version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
[[package]]
name = "itertools"
version = "0.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285"
dependencies = [
"either",
]
[[package]]
name = "itoa"
version = "1.0.18"
@@ -1236,6 +1301,82 @@ version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
[[package]]
name = "opentelemetry"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0"
dependencies = [
"futures-core",
"futures-sink",
"js-sys",
"pin-project-lite",
"thiserror 2.0.18",
"tracing",
]
[[package]]
name = "opentelemetry-http"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d"
dependencies = [
"async-trait",
"bytes",
"http",
"opentelemetry",
"reqwest",
]
[[package]]
name = "opentelemetry-otlp"
version = "0.31.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f"
dependencies = [
"http",
"opentelemetry",
"opentelemetry-http",
"opentelemetry-proto",
"opentelemetry_sdk",
"prost",
"reqwest",
"thiserror 2.0.18",
"tokio",
"tonic",
"tracing",
]
[[package]]
name = "opentelemetry-proto"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f"
dependencies = [
"opentelemetry",
"opentelemetry_sdk",
"prost",
"tonic",
"tonic-prost",
]
[[package]]
name = "opentelemetry_sdk"
version = "0.31.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd"
dependencies = [
"futures-channel",
"futures-executor",
"futures-util",
"opentelemetry",
"percent-encoding",
"rand 0.9.2",
"thiserror 2.0.18",
"tokio",
"tokio-stream",
]
[[package]]
name = "owo-colors"
version = "4.3.0"
@@ -1277,6 +1418,26 @@ version = "2.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
[[package]]
name = "pin-project"
version = "1.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1749c7ed4bcaf4c3d0a3efc28538844fb29bcdd7d2b67b2be7e20ba861ff517"
dependencies = [
"pin-project-internal",
]
[[package]]
name = "pin-project-internal"
version = "1.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d9b20ed30f105399776b9c883e68e536ef602a16ae6f596d2c473591d6ad64c6"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "pin-project-lite"
version = "0.2.17"
@@ -1326,6 +1487,29 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "prost"
version = "0.14.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568"
dependencies = [
"bytes",
"prost-derive",
]
[[package]]
name = "prost-derive"
version = "0.14.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b"
dependencies = [
"anyhow",
"itertools",
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "quinn"
version = "0.11.9"
@@ -1515,7 +1699,9 @@ checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147"
dependencies = [
"base64",
"bytes",
"futures-channel",
"futures-core",
"futures-util",
"http",
"http-body",
"http-body-util",
@@ -2061,6 +2247,17 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-stream"
version = "0.1.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
dependencies = [
"futures-core",
"pin-project-lite",
"tokio",
]
[[package]]
name = "tokio-tungstenite"
version = "0.24.0"
@@ -2089,6 +2286,19 @@ dependencies = [
"tungstenite 0.28.0",
]
[[package]]
name = "tokio-util"
version = "0.7.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
dependencies = [
"bytes",
"futures-core",
"futures-sink",
"pin-project-lite",
"tokio",
]
[[package]]
name = "toml"
version = "0.8.23"
@@ -2130,6 +2340,43 @@ version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801"
[[package]]
name = "tonic"
version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fec7c61a0695dc1887c1b53952990f3ad2e3a31453e1f49f10e75424943a93ec"
dependencies = [
"async-trait",
"base64",
"bytes",
"http",
"http-body",
"http-body-util",
"hyper",
"hyper-timeout",
"hyper-util",
"percent-encoding",
"pin-project",
"sync_wrapper",
"tokio",
"tokio-stream",
"tower",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
name = "tonic-prost"
version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a55376a0bbaa4975a3f10d009ad763d8f4108f067c7c2e74f3001fb49778d309"
dependencies = [
"bytes",
"prost",
"tonic",
]
[[package]]
name = "tower"
version = "0.5.3"
@@ -2138,9 +2385,12 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
dependencies = [
"futures-core",
"futures-util",
"indexmap 2.13.1",
"pin-project-lite",
"slab",
"sync_wrapper",
"tokio",
"tokio-util",
"tower-layer",
"tower-service",
"tracing",
@@ -2230,6 +2480,32 @@ dependencies = [
"tracing-core",
]
[[package]]
name = "tracing-opentelemetry"
version = "0.32.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ac28f2d093c6c477eaa76b23525478f38de514fa9aeb1285738d4b97a9552fc"
dependencies = [
"js-sys",
"opentelemetry",
"smallvec",
"tracing",
"tracing-core",
"tracing-log",
"tracing-subscriber",
"web-time",
]
[[package]]
name = "tracing-serde"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1"
dependencies = [
"serde",
"tracing-core",
]
[[package]]
name = "tracing-subscriber"
version = "0.3.23"
@@ -2240,12 +2516,15 @@ dependencies = [
"nu-ansi-term",
"once_cell",
"regex-automata",
"serde",
"serde_json",
"sharded-slab",
"smallvec",
"thread_local",
"tracing",
"tracing-core",
"tracing-log",
"tracing-serde",
]
[[package]]
@@ -2434,11 +2713,16 @@ dependencies = [
"clap",
"futures-util",
"nix",
"opentelemetry",
"opentelemetry-otlp",
"opentelemetry_sdk",
"reqwest",
"serde",
"serde_json",
"tokio",
"toml",
"tracing",
"tracing-opentelemetry",
"tracing-subscriber",
"uuid",
"wakey-agent",
README.md
+76
View File
@@ -70,6 +70,82 @@ The project now uses:
- `wakey-agent` for outbound enrollment and websocket execution
- `wakey-control-plane` for enrollment, registry, and command relay
## Logging and troubleshooting
Both daemons emit structured tracing logs to stderr. If you are not seeing
enrollment/command activity, run with increased verbosity.
Control-plane also supports a config file at
`/etc/wakey-control-plane/config.toml` (override with `--config-file`) so you
can persist telemetry settings instead of passing flags.
Example:
```toml
bind = "0.0.0.0:8080"
public_url = "https://cp.example.com"
state_file = "/var/lib/wakey-control-plane/state.json"
pid_file = "/var/run/wakey-control-plane.pid"
command_timeout_ms = 30000
[telemetry]
otlp_endpoint = "http://127.0.0.1:4317"
service_name = "wakey-control-plane"
json_logs = false
```
If `telemetry.otlp_endpoint` is omitted, logs still work normally and only local
structured logs are emitted.
### Quick start
Control-plane:
```sh
wakey-control-plane -v serve --bind 0.0.0.0:8787 --public-url https://cp.example.com
```
Agent:
```sh
wakey-agent -v serve --config /etc/wakey-agent/config.toml
```
### Fine-grained log filters
Use `RUST_LOG` when you want to focus on websocket/API internals.
```sh
RUST_LOG=wakey_control_plane=debug,wakey_agent=debug wakey-control-plane serve
RUST_LOG=wakey_agent=debug wakey-agent serve
```
### What you should see
During registration/enroll:
- control-plane: `issued enroll token`, then `agent enrollment accepted`
- agent: `starting agent enrollment`, then `agent enrollment succeeded and config was written`
During live connectivity:
- control-plane: `agent websocket upgraded`, `agent authenticated`, `agent disconnected`
- agent: `connecting agent websocket`, `agent websocket session authenticated`, `heartbeat sent` (debug)
During command relay:
- control-plane: `dispatching command to agent`
- agent: `received command from control-plane`, then `command execution completed` (or `command dispatch failed`)
- control-plane: `agent command completed` (or timeout/error warnings)
During daemon control/state operations:
- control-plane: `wrote control-plane pid file`, `saved control-plane store`, `reloaded control-plane store from disk`
- agent: `wrote wakey-agent pid file`, `sending wakey-agent reload signal`
If commands still appear silent, verify both processes are running with `-v`
and that `RUST_LOG` is not overriding to a stricter level.
## CLI
`wakey` is usable as a local/operator CLI.
wakey-agent/src/dispatch.rs
+17 -1
View File
@@ -1,5 +1,5 @@
use anyhow::Result;
use tracing::{debug, instrument};
use tracing::{debug, info, instrument};
use crate::protocol::{
AgentCommand, CommandResult, DevsRequest, InventoryRequest, LeasesRequest, StatusRequest,
@@ -8,6 +8,8 @@ use crate::protocol::{
#[instrument(skip_all)]
pub async fn dispatch_command(command: AgentCommand) -> Result<CommandResult> {
let kind = command_kind(&command);
info!(command = %kind, "dispatching command into local wakey services");
match command {
AgentCommand::Status(req) => dispatch_status(req).await,
AgentCommand::Leases(req) => dispatch_leases(req).await,
@@ -38,6 +40,7 @@ async fn dispatch_leases(req: LeasesRequest) -> Result<CommandResult> {
include_state: req.include_state,
})
.await?;
debug!(rows = leases.len(), include_state = req.include_state, "dispatched leases command");
Ok(CommandResult::Leases(leases))
}
@@ -50,11 +53,13 @@ async fn dispatch_devs(req: DevsRequest) -> Result<CommandResult> {
if req.up_only {
devs.retain(|dev| dev.operstate == "up");
}
debug!(rows = devs.len(), up_only = req.up_only, "dispatched devs command");
Ok(CommandResult::Devs(devs))
}
async fn dispatch_inventory(req: InventoryRequest) -> Result<CommandResult> {
let inventory = wakey::inventory(req.into_device_query()).await?;
debug!(rows = inventory.devices.len(), "dispatched inventory command");
Ok(CommandResult::Inventory(inventory))
}
@@ -65,9 +70,20 @@ async fn dispatch_wake(req: WakeRequest) -> Result<CommandResult> {
(None, Some(mac), ip) => wakey::wake_explicit(mac, ip).await?,
_ => unreachable!("wake request validated before dispatch"),
};
debug!("dispatched wake command");
Ok(CommandResult::Wake(result))
}
fn command_kind(command: &AgentCommand) -> &'static str {
match command {
AgentCommand::Status(_) => "status",
AgentCommand::Leases(_) => "leases",
AgentCommand::Devs(_) => "devs",
AgentCommand::Inventory(_) => "inventory",
AgentCommand::Wake(_) => "wake",
}
}
pub fn validate_wake_request(req: &WakeRequest) -> Result<()> {
let has_query = req.query.is_some();
let has_mac = req.mac.is_some();
wakey-agent/src/enroll.rs
+4 -2
View File
@@ -2,7 +2,7 @@ use anyhow::{Context, Result};
use reqwest::StatusCode;
use serde::{Deserialize, Serialize};
use std::path::Path;
use tracing::info;
use tracing::{info, warn};
use crate::config::{AgentConfig, save_config};
@@ -21,6 +21,7 @@ struct EnrollResponse {
pub async fn enroll(server_url: &str, enroll_token: &str, config_path: &Path) -> Result<AgentConfig> {
let server_url = normalize_server_url(server_url);
let endpoint = format!("{server_url}/api/v1/agents/enroll");
info!(endpoint = %endpoint, config_path = %config_path.display(), "starting agent enrollment");
let client = reqwest::Client::new();
let response = client
.post(endpoint)
@@ -35,6 +36,7 @@ pub async fn enroll(server_url: &str, enroll_token: &str, config_path: &Path) ->
.text()
.await
.unwrap_or_else(|_| "<unable to read enrollment error body>".into());
warn!(status = %status, "agent enrollment request rejected by control-plane");
anyhow::bail!("enrollment failed with {status}: {body}");
}
@@ -51,7 +53,7 @@ pub async fn enroll(server_url: &str, enroll_token: &str, config_path: &Path) ->
reconnect_max_ms: 30_000,
};
save_config(config_path, &config)?;
info!(config_path = %config_path.display(), "wrote agent config");
info!(agent_id = %config.agent_id, config_path = %config_path.display(), "agent enrollment succeeded and config was written");
Ok(config)
}
wakey-agent/src/main.rs
+13 -3
View File
@@ -17,8 +17,12 @@ async fn main() -> Result<()> {
tracing::init(cli.verbose);
match cli.command {
Command::Serve(args) => serve::serve(args).await?,
Command::Serve(args) => {
::tracing::info!("wakey-agent command: serve");
serve::serve(args).await?
}
Command::Enroll(args) => {
::tracing::info!(server_url = %args.server_url, config = %args.config.display(), "wakey-agent command: enroll");
let config = enroll::enroll(&args.server_url, &args.enroll_token, &args.config).await?;
println!("agent_id={}", config.agent_id);
println!("config={}", args.config.display());
@@ -29,8 +33,14 @@ async fn main() -> Result<()> {
}
}
}
Command::InitConfig(args) => init_config(args)?,
Command::Reload(args) => serve::reload_daemon(&args.pid_file)?,
Command::InitConfig(args) => {
::tracing::info!(config = %args.config.display(), force = args.force, "wakey-agent command: init-config");
init_config(args)?
}
Command::Reload(args) => {
::tracing::info!(pid_file = %args.pid_file.display(), "wakey-agent command: reload");
serve::reload_daemon(&args.pid_file)?
}
}
Ok(())
wakey-agent/src/serve.rs
+3
View File
@@ -16,6 +16,7 @@ pub async fn serve(args: ServeArgs) -> Result<()> {
}
write_pid_file(&args.pid_file)?;
info!(pid_file = %args.pid_file.display(), "wrote wakey-agent pid file");
let mut cfg = config::load_config(&args.config)?;
info!(config_path = %args.config.display(), agent_id = %cfg.agent_id, "starting wakey-agent");
@@ -48,6 +49,7 @@ pub async fn serve(args: ServeArgs) -> Result<()> {
}
}
join = &mut worker => {
warn!("agent worker task exited; shutting down daemon");
let _ = remove_pid_file(&args.pid_file);
return join.context("agent session join failed")?;
}
@@ -69,6 +71,7 @@ pub async fn serve(args: ServeArgs) -> Result<()> {
pub fn reload_daemon(pid_file: &Path) -> Result<()> {
let pid = read_pid(pid_file)?;
info!(pid, pid_file = %pid_file.display(), "sending wakey-agent reload signal");
send_hup(pid)
}
wakey-agent/src/session.rs
+64 -21
View File
@@ -2,11 +2,11 @@ use anyhow::{Context, Result};
use futures_util::{SinkExt, StreamExt};
use tokio::time::{Duration, MissedTickBehavior, interval, sleep};
use tokio_tungstenite::{connect_async, tungstenite::Message};
use tracing::{debug, error, info, warn};
use tracing::{debug, error, info, info_span, warn};
use crate::config::AgentConfig;
use crate::dispatch::dispatch_command;
use crate::protocol::{ClientMessage, ErrorPayload, ServerMessage};
use crate::protocol::{AgentCommand, ClientMessage, ErrorPayload, ServerMessage};
pub async fn run(config: AgentConfig) -> Result<()> {
let mut backoff = config.reconnect_base_ms.max(100);
@@ -25,6 +25,17 @@ pub async fn run(config: AgentConfig) -> Result<()> {
}
async fn run_once(config: &AgentConfig) -> Result<()> {
let session_id = format!(
"{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis())
.unwrap_or_default()
);
let span = info_span!("agent_session", session_id = %session_id, agent_id = %config.agent_id);
let _span_guard = span.enter();
let ws_url = websocket_url(&config.server_url)?;
info!(%ws_url, agent_id = %config.agent_id, "connecting agent websocket");
let (stream, _) = connect_async(ws_url.as_str())
@@ -44,6 +55,7 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
},
)
.await?;
info!(agent_id = %config.agent_id, "agent websocket session authenticated");
let mut heartbeat = interval(Duration::from_secs(30));
heartbeat.set_missed_tick_behavior(MissedTickBehavior::Skip);
@@ -54,11 +66,15 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
send_json(&mut sink, &ClientMessage::Heartbeat {
agent_id: config.agent_id.clone(),
}).await?;
debug!(agent_id = %config.agent_id, "heartbeat sent");
}
maybe_msg = source.next() => {
let msg = match maybe_msg {
Some(msg) => msg.context("websocket frame failed")?,
None => anyhow::bail!("websocket closed by server"),
None => {
info!("websocket stream closed by server");
anyhow::bail!("websocket closed by server");
}
};
match msg {
@@ -79,6 +95,7 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
}
Message::Pong(_) => {}
Message::Close(frame) => {
info!(close = ?frame, "received close frame from server");
anyhow::bail!("websocket closed: {:?}", frame);
}
Message::Binary(_) => {
@@ -105,26 +122,31 @@ where
ServerMessage::Command {
request_id,
command,
} => match dispatch_command(command).await {
Ok(result) => {
send_json(sink, &ClientMessage::Result { request_id, result }).await?;
}
Err(err) => {
error!(request_id = %request_id, error = %err, "command dispatch failed");
send_json(
sink,
&ClientMessage::Error {
request_id,
error: ErrorPayload {
code: "command_dispatch_failed".into(),
message: err.to_string(),
retryable: None,
} => {
let kind = command_kind(&command);
info!(request_id = %request_id, command = %kind, "received command from control-plane");
match dispatch_command(command).await {
Ok(result) => {
info!(request_id = %request_id, command = %kind, "command execution completed");
send_json(sink, &ClientMessage::Result { request_id, result }).await?;
}
Err(err) => {
error!(request_id = %request_id, command = %kind, error = %err, "command dispatch failed");
send_json(
sink,
&ClientMessage::Error {
request_id,
error: ErrorPayload {
code: "command_dispatch_failed".into(),
message: err.to_string(),
retryable: None,
},
},
},
)
.await?;
)
.await?;
}
}
},
}
}
Ok(())
}
@@ -135,12 +157,33 @@ where
<S as futures_util::Sink<Message>>::Error: std::error::Error + Send + Sync + 'static,
{
let payload = serde_json::to_string(message).context("failed to serialize websocket message")?;
debug!(message_type = %client_message_kind(message), "sending websocket message");
sink.send(Message::Text(payload))
.await
.context("failed to send websocket message")?;
Ok(())
}
fn command_kind(command: &AgentCommand) -> &'static str {
match command {
AgentCommand::Status(_) => "status",
AgentCommand::Leases(_) => "leases",
AgentCommand::Devs(_) => "devs",
AgentCommand::Inventory(_) => "inventory",
AgentCommand::Wake(_) => "wake",
}
}
fn client_message_kind(message: &ClientMessage) -> &'static str {
match message {
ClientMessage::Hello { .. } => "hello",
ClientMessage::Auth { .. } => "auth",
ClientMessage::Heartbeat { .. } => "heartbeat",
ClientMessage::Result { .. } => "result",
ClientMessage::Error { .. } => "error",
}
}
pub fn websocket_url(server_url: &str) -> Result<url::Url> {
let base = url::Url::parse(server_url).context("invalid server_url")?;
let scheme = match base.scheme() {
wakey-control-plane/Cargo.toml
+6 -1
View File
@@ -12,6 +12,7 @@ nix = { version = "0.30", default-features = false, features = ["signal", "proce
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
toml = "0.8"
tokio = { version = "1", features = [
"macros",
"rt-multi-thread",
@@ -21,6 +22,10 @@ tokio = { version = "1", features = [
"signal",
] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
tracing-opentelemetry = "0.32"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "json"] }
opentelemetry = "0.31"
opentelemetry-otlp = { version = "0.31", features = ["grpc-tonic", "trace"] }
opentelemetry_sdk = { version = "0.31", features = ["rt-tokio"] }
uuid = { version = "1", features = ["v4", "serde"] }
wakey-agent = { path = "../wakey-agent", registry = "gitea" }
wakey-control-plane/src/api.rs
+82 -36
View File
@@ -3,6 +3,7 @@ use axum::extract::{Path as AxumPath, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use serde::{Deserialize, Serialize};
use tracing::{info, info_span, warn};
use uuid::Uuid;
use wakey_agent::protocol::{AgentCommand, ErrorPayload, RequestId, ServerMessage};
@@ -56,19 +57,25 @@ pub async fn enroll(
Json(req): Json<EnrollRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.enroll(&req.enroll_token).await {
Ok(issued) => Ok((
StatusCode::OK,
Json(EnrollResponse {
agent_id: issued.agent_id,
agent_token: issued.agent_token,
server_url: state.public_url,
}),
)),
Err(err) => Err(json_error(
StatusCode::UNAUTHORIZED,
"enrollment_rejected",
&err.to_string(),
)),
Ok(issued) => {
info!(agent_id = %issued.agent_id, "agent enrollment accepted");
Ok((
StatusCode::OK,
Json(EnrollResponse {
agent_id: issued.agent_id,
agent_token: issued.agent_token,
server_url: state.public_url,
}),
))
}
Err(err) => {
warn!(error = %err, "agent enrollment rejected");
Err(json_error(
StatusCode::UNAUTHORIZED,
"enrollment_rejected",
&err.to_string(),
))
}
}
}
@@ -76,17 +83,23 @@ pub async fn issue_enroll_token(
State(state): State<AppState>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
match state.store.issue_enroll_token().await {
Ok(token) => Ok((
StatusCode::OK,
Json(IssueEnrollTokenResponse {
enroll_token: token,
}),
)),
Err(err) => Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"issue_enroll_token_failed",
&err.to_string(),
)),
Ok(token) => {
info!("issued enroll token");
Ok((
StatusCode::OK,
Json(IssueEnrollTokenResponse {
enroll_token: token,
}),
))
}
Err(err) => {
warn!(error = %err, "failed to issue enroll token");
Err(json_error(
StatusCode::INTERNAL_SERVER_ERROR,
"issue_enroll_token_failed",
&err.to_string(),
))
}
}
}
@@ -113,6 +126,15 @@ pub async fn run_command(
Json(req): Json<RelayCommandRequest>,
) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
let request_id_string = format!("req-{}", Uuid::new_v4());
let command = command_kind(&req.command);
let span = info_span!(
"relay_command",
agent_id = %agent_id,
request_id = %request_id_string,
command = %command,
);
let _span_guard = span.enter();
let request_id = RequestId::try_from(request_id_string.clone()).map_err(|err| {
json_error(
StatusCode::INTERNAL_SERVER_ERROR,
@@ -126,6 +148,7 @@ pub async fn run_command(
sessions.get(&agent_id).cloned()
}
.ok_or_else(|| {
warn!("command rejected: agent not connected");
json_error(
StatusCode::NOT_FOUND,
"agent_not_connected",
@@ -140,11 +163,16 @@ pub async fn run_command(
.await
.insert(request_id_string.clone(), pending_tx);
info!(
"dispatching command to agent"
);
if let Err(err) = tx.send(ServerMessage::Command {
request_id,
command: req.command,
}) {
state.pending.lock().await.remove(&request_id_string);
warn!(error = %err, "failed sending command to agent session");
return Err(json_error(
StatusCode::BAD_GATEWAY,
"agent_send_failed",
@@ -159,19 +187,26 @@ pub async fn run_command(
);
let outcome = tokio::time::timeout(timeout, pending_rx).await;
let response = match outcome {
Ok(Ok(AgentReply::Result(result))) => RelayCommandResponse {
request_id: request_id_string,
status: "ok".into(),
result: Some(result),
error: None,
},
Ok(Ok(AgentReply::Error(error))) => RelayCommandResponse {
request_id: request_id_string,
status: "error".into(),
result: None,
error: Some(error),
},
Ok(Ok(AgentReply::Result(result))) => {
info!("agent command completed");
RelayCommandResponse {
request_id: request_id_string,
status: "ok".into(),
result: Some(result),
error: None,
}
}
Ok(Ok(AgentReply::Error(error))) => {
warn!(code = %error.code, "agent command returned error");
RelayCommandResponse {
request_id: request_id_string,
status: "error".into(),
result: None,
error: Some(error),
}
}
Ok(Err(_)) => {
warn!("agent response channel dropped");
return Err(json_error(
StatusCode::BAD_GATEWAY,
"agent_response_dropped",
@@ -180,6 +215,7 @@ pub async fn run_command(
}
Err(_) => {
state.pending.lock().await.remove(&request_id_string);
warn!(timeout_ms = timeout.as_millis() as u64, "agent command timed out");
return Err(json_error(
StatusCode::GATEWAY_TIMEOUT,
"agent_timeout",
@@ -206,3 +242,13 @@ pub fn json_error(
})),
)
}
fn command_kind(command: &AgentCommand) -> &'static str {
match command {
AgentCommand::Status(_) => "status",
AgentCommand::Leases(_) => "leases",
AgentCommand::Devs(_) => "devs",
AgentCommand::Inventory(_) => "inventory",
AgentCommand::Wake(_) => "wake",
}
}
wakey-control-plane/src/cli.rs
+14 -10
View File
@@ -5,6 +5,7 @@ use clap::{ArgAction, Args, Parser, Subcommand};
pub const DEFAULT_STATE_FILE: &str = "/var/lib/wakey-control-plane/state.json";
pub const DEFAULT_PID_FILE: &str = "/var/run/wakey-control-plane.pid";
pub const DEFAULT_CONFIG_FILE: &str = "/etc/wakey-control-plane/config.toml";
#[derive(Parser)]
#[command(name = "wakey-control-plane")]
@@ -29,23 +30,26 @@ pub enum Command {
#[derive(Args, Clone)]
pub struct ServeArgs {
#[arg(long, default_value = "0.0.0.0:8080")]
pub bind: SocketAddr,
#[arg(long)]
pub bind: Option<SocketAddr>,
#[arg(long, default_value = "http://127.0.0.1:8080")]
pub public_url: String,
#[arg(long)]
pub public_url: Option<String>,
#[arg(long, default_value = DEFAULT_STATE_FILE)]
pub state_file: PathBuf,
#[arg(long)]
pub state_file: Option<PathBuf>,
#[arg(long = "enroll-token")]
pub enroll_tokens: Vec<String>,
#[arg(long, default_value_t = 30_000)]
pub command_timeout_ms: u64,
#[arg(long)]
pub command_timeout_ms: Option<u64>,
#[arg(long, default_value = DEFAULT_PID_FILE)]
pub pid_file: PathBuf,
#[arg(long)]
pub pid_file: Option<PathBuf>,
#[arg(long, default_value = DEFAULT_CONFIG_FILE)]
pub config_file: PathBuf,
}
#[derive(Args)]
wakey-control-plane/src/config.rs
+129 -7
View File
@@ -1,7 +1,11 @@
use std::net::SocketAddr;
use std::path::Path;
use std::path::PathBuf;
use std::time::Duration;
use anyhow::{Context, Result};
use serde::Deserialize;
use crate::cli::ServeArgs;
#[derive(Debug, Clone)]
@@ -11,18 +15,136 @@ pub struct DaemonConfig {
pub state_file: PathBuf,
pub command_timeout: Duration,
pub pid_file: PathBuf,
pub enroll_tokens: Vec<String>,
pub telemetry: TelemetryConfig,
}
#[derive(Debug, Clone)]
pub struct TelemetryConfig {
pub otlp_endpoint: Option<String>,
pub service_name: String,
pub json_logs: bool,
}
impl Default for TelemetryConfig {
fn default() -> Self {
Self {
otlp_endpoint: None,
service_name: "wakey-control-plane".to_string(),
json_logs: false,
}
}
}
#[derive(Debug, Deserialize, Default)]
struct FileConfig {
bind: Option<String>,
public_url: Option<String>,
state_file: Option<PathBuf>,
command_timeout_ms: Option<u64>,
pid_file: Option<PathBuf>,
enroll_tokens: Option<Vec<String>>,
telemetry: Option<FileTelemetryConfig>,
}
#[derive(Debug, Deserialize, Default)]
struct FileTelemetryConfig {
otlp_endpoint: Option<String>,
service_name: Option<String>,
json_logs: Option<bool>,
}
impl DaemonConfig {
pub fn from_serve_args(args: &ServeArgs) -> Self {
Self {
bind: args.bind,
public_url: normalize_public_url(&args.public_url),
state_file: args.state_file.clone(),
command_timeout: Duration::from_millis(args.command_timeout_ms.max(1)),
pid_file: args.pid_file.clone(),
pub fn from_serve_args(args: &ServeArgs) -> Result<Self> {
let file = load_file_config(&args.config_file)?;
let bind = match args.bind {
Some(bind) => bind,
None => match file.bind {
Some(ref bind) => bind
.parse::<SocketAddr>()
.with_context(|| format!("invalid bind address `{bind}` in {}", args.config_file.display()))?,
None => "0.0.0.0:8080".parse().expect("static default bind should parse"),
},
};
let public_url = normalize_public_url(
args.public_url
.as_deref()
.or(file.public_url.as_deref())
.unwrap_or("http://127.0.0.1:8080"),
);
let state_file = args
.state_file
.clone()
.or(file.state_file)
.unwrap_or_else(|| PathBuf::from(crate::cli::DEFAULT_STATE_FILE));
let command_timeout = Duration::from_millis(
args.command_timeout_ms
.or(file.command_timeout_ms)
.unwrap_or(30_000)
.max(1),
);
let pid_file = args
.pid_file
.clone()
.or(file.pid_file)
.unwrap_or_else(|| PathBuf::from(crate::cli::DEFAULT_PID_FILE));
let enroll_tokens = if args.enroll_tokens.is_empty() {
file.enroll_tokens.unwrap_or_default()
} else {
args.enroll_tokens.clone()
};
let telemetry = resolve_telemetry(file.telemetry);
Ok(Self {
bind,
public_url,
state_file,
command_timeout,
pid_file,
enroll_tokens,
telemetry,
})
}
}
fn load_file_config(path: &Path) -> Result<FileConfig> {
if !path.exists() {
return Ok(FileConfig::default());
}
let raw = std::fs::read_to_string(path)
.with_context(|| format!("failed to read config file {}", path.display()))?;
toml::from_str::<FileConfig>(&raw)
.with_context(|| format!("failed to parse config file {}", path.display()))
}
fn resolve_telemetry(file: Option<FileTelemetryConfig>) -> TelemetryConfig {
let mut out = TelemetryConfig::default();
if let Some(file) = file {
if let Some(endpoint) = file.otlp_endpoint {
let trimmed = endpoint.trim().to_string();
if !trimmed.is_empty() {
out.otlp_endpoint = Some(trimmed);
}
}
if let Some(name) = file.service_name {
let trimmed = name.trim();
if !trimmed.is_empty() {
out.service_name = trimmed.to_string();
}
}
if let Some(json_logs) = file.json_logs {
out.json_logs = json_logs;
}
}
out
}
pub fn normalize_public_url(url: &str) -> String {
wakey-control-plane/src/main.rs
+13 -4
View File
@@ -13,11 +13,20 @@ use cli::{Cli, Command};
#[tokio::main]
async fn main() -> Result<()> {
let cli = Cli::parse();
tracing::init(cli.verbose);
match cli.command {
Command::Serve(args) => runtime::serve(args).await,
Command::IssueEnrollToken(args) => runtime::issue_enroll_token(args).await,
Command::Reload(args) => runtime::reload_daemon(&args.pid_file),
Command::Serve(args) => {
let daemon = config::DaemonConfig::from_serve_args(&args)?;
tracing::init(cli.verbose, &daemon.telemetry)?;
runtime::serve(daemon).await
}
Command::IssueEnrollToken(args) => {
tracing::init(cli.verbose, &config::TelemetryConfig::default())?;
runtime::issue_enroll_token(args).await
}
Command::Reload(args) => {
tracing::init(cli.verbose, &config::TelemetryConfig::default())?;
runtime::reload_daemon(&args.pid_file)
}
}
}
wakey-control-plane/src/runtime.rs
+8 -4
View File
@@ -12,7 +12,7 @@ use tracing::{info, warn};
use wakey_agent::protocol::{ErrorPayload, ServerMessage};
use crate::api;
use crate::cli::{IssueEnrollTokenArgs, ServeArgs};
use crate::cli::IssueEnrollTokenArgs;
use crate::config;
use crate::state;
use crate::ws;
@@ -31,11 +31,11 @@ pub enum AgentReply {
Error(ErrorPayload),
}
pub async fn serve(args: ServeArgs) -> Result<()> {
let daemon = config::DaemonConfig::from_serve_args(&args);
pub async fn serve(daemon: config::DaemonConfig) -> Result<()> {
write_pid_file(&daemon.pid_file)?;
info!(pid_file = %daemon.pid_file.display(), "wrote control-plane pid file");
let store = state::Store::load_or_init(&daemon.state_file, args.enroll_tokens)
let store = state::Store::load_or_init(&daemon.state_file, daemon.enroll_tokens.clone())
.await
.with_context(|| format!("failed to initialize store {}", daemon.state_file.display()))?;
@@ -109,6 +109,7 @@ pub async fn issue_enroll_token(args: IssueEnrollTokenArgs) -> Result<()> {
if let Some(url) = args.public_url {
let base = config::normalize_public_url(&url);
let endpoint = config::issue_token_endpoint(&base);
info!(endpoint = %endpoint, "requesting live enroll token from running control-plane daemon");
let client = reqwest::Client::new();
let response = client
@@ -130,6 +131,7 @@ pub async fn issue_enroll_token(args: IssueEnrollTokenArgs) -> Result<()> {
.json()
.await
.context("failed to decode live issuance response")?;
info!("received live enroll token response");
println!("enroll_token={}", payload.enroll_token);
println!(
@@ -140,6 +142,7 @@ pub async fn issue_enroll_token(args: IssueEnrollTokenArgs) -> Result<()> {
}
// Fallback for offline tooling: writes to state file, requires daemon reload to pick up.
info!(state_file = %args.state_file.display(), "issuing enroll token via offline state file fallback");
let store = state::Store::load_or_init(&args.state_file, args.enroll_tokens)
.await
.with_context(|| format!("failed to initialize store {}", args.state_file.display()))?;
@@ -154,6 +157,7 @@ pub async fn issue_enroll_token(args: IssueEnrollTokenArgs) -> Result<()> {
pub fn reload_daemon(pid_file: &Path) -> Result<()> {
let pid = read_pid(pid_file)?;
info!(pid, pid_file = %pid_file.display(), "sending control-plane reload signal");
send_hup(pid)
}
wakey-control-plane/src/state.rs
+28
View File
@@ -4,6 +4,7 @@ use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use tracing::{debug, info, warn};
use uuid::Uuid;
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -35,9 +36,11 @@ impl Store {
let raw = tokio::fs::read_to_string(path)
.await
.with_context(|| format!("failed to read store {}", path.display()))?;
debug!(path = %path.display(), bytes = raw.len(), "loading persisted control-plane store");
serde_json::from_str::<PersistedState>(&raw)
.with_context(|| format!("failed to decode store {}", path.display()))?
} else {
info!(path = %path.display(), seeded_enroll_tokens = seeded_tokens.len(), "initializing new control-plane store");
PersistedState {
enroll_tokens: seeded_tokens,
agents: HashMap::new(),
@@ -49,12 +52,23 @@ impl Store {
state: RwLock::new(initial),
};
store.save().await?;
let (enroll_tokens, agents) = {
let snapshot = store.state.read().await;
(snapshot.enroll_tokens.len(), snapshot.agents.len())
};
info!(
path = %store.path.display(),
enroll_tokens,
agents,
"control-plane store ready"
);
Ok(store)
}
pub async fn enroll(&self, enroll_token: &str) -> Result<IssuedAgent> {
let mut state = self.state.write().await;
if !state.enroll_tokens.remove(enroll_token) {
warn!("rejecting enroll attempt with invalid or consumed token");
anyhow::bail!("invalid or already-used enroll token");
}
@@ -63,6 +77,7 @@ impl Store {
state.agents.insert(agent_id.clone(), agent_token.clone());
drop(state);
self.save().await?;
info!(agent_id = %agent_id, "issued persistent agent credentials");
Ok(IssuedAgent {
agent_id,
@@ -76,11 +91,13 @@ impl Store {
state.enroll_tokens.insert(token.clone());
drop(state);
self.save().await?;
info!("persisted new enroll token");
Ok(token)
}
pub async fn reload_from_disk(&self) -> Result<()> {
if !self.path.exists() {
warn!(path = %self.path.display(), "reload requested but store file is missing");
return Ok(());
}
@@ -90,7 +107,10 @@ impl Store {
let decoded = serde_json::from_str::<PersistedState>(&raw)
.with_context(|| format!("failed to decode store {}", self.path.display()))?;
let enroll_tokens = decoded.enroll_tokens.len();
let agents = decoded.agents.len();
*self.state.write().await = decoded;
info!(path = %self.path.display(), enroll_tokens, agents, "reloaded control-plane store from disk");
Ok(())
}
@@ -121,6 +141,7 @@ impl Store {
let snapshot = self.state.read().await;
let body =
serde_json::to_string_pretty(&*snapshot).context("failed to serialize store state")?;
let body_len = body.len();
if let Some(parent) = self.path.parent() {
tokio::fs::create_dir_all(parent)
@@ -139,6 +160,13 @@ impl Store {
self.path.display()
)
})?;
debug!(
path = %self.path.display(),
bytes = body_len,
enroll_tokens = snapshot.enroll_tokens.len(),
agents = snapshot.agents.len(),
"saved control-plane store"
);
Ok(())
}
}
wakey-control-plane/src/tracing.rs
+76 -5
View File
@@ -1,14 +1,85 @@
use anyhow::{Context, Result};
use opentelemetry::trace::TracerProvider as _;
use opentelemetry::global;
use opentelemetry_otlp::WithExportConfig;
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::trace::{SdkTracerProvider, Tracer};
use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
pub fn init(verbose: u8) {
use crate::config::TelemetryConfig;
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");
tracing_subscriber::registry()
.with(filter)
.with(fmt::layer())
.init();
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())
.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())
.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())
.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())
.init();
tracing::info!(json_logs = telemetry.json_logs, "tracing initialized without otlp exporter");
}
Ok(())
}
fn build_otel_layer(
telemetry: &TelemetryConfig,
) -> Result<
Option<
tracing_opentelemetry::OpenTelemetryLayer<
tracing_subscriber::Registry,
Tracer,
>,
>,
> {
let Some(endpoint) = telemetry.otlp_endpoint.as_deref() else {
return Ok(None);
};
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 {
wakey-control-plane/src/ws.rs
+22 -3
View File
@@ -5,7 +5,8 @@ use axum::response::IntoResponse;
use futures_util::{SinkExt, StreamExt};
use serde::Deserialize;
use tokio::sync::mpsc;
use tracing::{debug, info, warn};
use tracing::{debug, info, info_span, warn};
use uuid::Uuid;
use wakey_agent::protocol::{ErrorPayload, RequestId, ServerMessage};
use crate::runtime::{AgentReply, AppState};
@@ -38,6 +39,11 @@ pub async fn agent_ws(ws: WebSocketUpgrade, State(state): State<AppState>) -> im
}
async fn handle_agent_socket(state: AppState, socket: WebSocket) {
let connection_id = Uuid::new_v4().to_string();
let span = info_span!("agent_ws_connection", connection_id = %connection_id);
let _span_guard = span.enter();
info!("agent websocket upgraded");
let (mut write, mut read) = socket.split();
let (tx, mut rx) = mpsc::unbounded_channel::<ServerMessage>();
@@ -55,6 +61,7 @@ async fn handle_agent_socket(state: AppState, socket: WebSocket) {
break;
}
}
debug!("websocket writer loop ended");
});
let mut authed_agent_id: Option<String> = None;
@@ -67,7 +74,10 @@ async fn handle_agent_socket(state: AppState, socket: WebSocket) {
warn!(error = %err, "agent websocket receive error");
break;
}
None => break,
None => {
info!("agent websocket stream ended by peer");
break;
}
};
match msg {
@@ -80,7 +90,10 @@ async fn handle_agent_socket(state: AppState, socket: WebSocket) {
}
Message::Ping(_) => {}
Message::Pong(_) => {}
Message::Close(_) => break,
Message::Close(frame) => {
info!(close = ?frame, "agent websocket close frame received");
break;
}
Message::Binary(_) => {
debug!("ignoring unexpected binary websocket frame");
}
@@ -93,6 +106,7 @@ async fn handle_agent_socket(state: AppState, socket: WebSocket) {
}
writer.abort();
debug!("agent websocket connection cleanup complete");
}
async fn process_agent_text(
@@ -117,6 +131,7 @@ async fn process_agent_text(
.verify_agent_token(&agent_id, &agent_token)
.await
{
warn!(agent_id = %agent_id, "agent auth rejected");
anyhow::bail!("agent auth rejected");
}
state
@@ -140,6 +155,8 @@ async fn process_agent_text(
let key = request_id.as_str().to_string();
if let Some(waiter) = state.pending.lock().await.remove(&key) {
let _ = waiter.send(AgentReply::Result(result));
} else {
debug!(request_id = %key, "dropping unsolicited result from agent");
}
}
IncomingClientMessage::Error { request_id, error } => {
@@ -149,6 +166,8 @@ async fn process_agent_text(
let key = request_id.as_str().to_string();
if let Some(waiter) = state.pending.lock().await.remove(&key) {
let _ = waiter.send(AgentReply::Error(error));
} else {
debug!(request_id = %key, "dropping unsolicited error from agent");
}
}
}