feat(telemetry): link agent snapshot collection to CC persistence
This commit is contained in:
7 files changed
+106
-3
No files matched your search
@@ -17,6 +17,9 @@ and reuse it. The helper checks the key again after building: if inputs changed
|
|||||||
while database reads/grouping were in flight, it discards the result and retries.
|
while database reads/grouping were in flight, it discards the result and retries.
|
||||||
Build errors are returned, not cached; cancelling a reader releases the lock.
|
Build errors are returned, not cached; cancelling a reader releases the lock.
|
||||||
The previous view is not served as a fallback after a rebuild error.
|
The previous view is not served as a fallback after a rebuild error.
|
||||||
|
Retries currently have no fixed limit. Sustained mutations faster than rebuilding
|
||||||
|
can delay the rebuilding reader and readers waiting on its lock; bounding this
|
||||||
|
behavior is tracked as a reliability follow-up.
|
||||||
|
|
||||||
Wake uses the same view, but command dispatch still checks the current session;
|
Wake uses the same view, but command dispatch still checks the current session;
|
||||||
a cached route cannot grant access to a disconnected agent.
|
a cached route cannot grant access to a disconnected agent.
|
||||||
|
|||||||
+2
-1
@@ -43,7 +43,7 @@ of remembered identifiers based on observation expiry.
|
|||||||
- [x] Separate console, OTLP log, and trace filters using per-layer filtering. Explicit signal config overrides `RUST_LOG`, then CLI verbosity supplies defaults. Unconfigured defaults remain info-level; exporter-failure logs stay local. Live validation pending.
|
- [x] Separate console, OTLP log, and trace filters using per-layer filtering. Explicit signal config overrides `RUST_LOG`, then CLI verbosity supplies defaults. Unconfigured defaults remain info-level; exporter-failure logs stay local. Live validation pending.
|
||||||
- [ ] Record hostname-resolution success/failure without exporting hostnames or resolved addresses. Current spans show timing but do not explicitly classify the result.
|
- [ ] Record hostname-resolution success/failure without exporting hostnames or resolved addresses. Current spans show timing but do not explicitly classify the result.
|
||||||
- [ ] Investigate the slow resolver with a reproducible device-side check. Timing isolated the phase, not the cause; do not treat instrumentation as a fix.
|
- [ ] Investigate the slow resolver with a reproducible device-side check. Timing isolated the phase, not the cause; do not treat instrumentation as a fix.
|
||||||
- [ ] Propagate trace context for periodic snapshots so agent collection and CC persistence can be inspected together. Commands already have this linkage.
|
- [x] Propagate optional W3C trace context for periodic/requested WS snapshots so agent collection and CC persistence can be inspected together. Wire compatibility and context linkage tested; live validation pending.
|
||||||
- [ ] Add current-state device gauges: stored rows per agent and grouped fleet devices split known/unknown. Define scope and update lifecycle first; never derive gauges from the last filtered UI query.
|
- [ ] Add current-state device gauges: stored rows per agent and grouped fleet devices split known/unknown. Define scope and update lifecycle first; never derive gauges from the last filtered UI query.
|
||||||
- [ ] Decide whether snapshots need persistence acknowledgements. The existing sent-device counter does not prove CC storage; any confirmed-storage metric needs an authoritative measurement point.
|
- [ ] Decide whether snapshots need persistence acknowledgements. The existing sent-device counter does not prove CC storage; any confirmed-storage metric needs an authoritative measurement point.
|
||||||
|
|
||||||
@@ -52,6 +52,7 @@ or current fleet size. Metric labels should remain bounded; no per-device labels
|
|||||||
|
|
||||||
## Validation And Reliability
|
## Validation And Reliability
|
||||||
|
|
||||||
|
- [ ] Bound fleet-cache rebuild retries under sustained mutations; current stale-build rejection can keep a request waiting while inputs continuously change. Do not silently serve stale wake routes as a fallback.
|
||||||
- [ ] Reproduce/check Ctrl-C cleanup on the test device: process groups, PTYs, relay tasks, and telemetry shutdown. Leftovers were suspected, not confirmed.
|
- [ ] Reproduce/check Ctrl-C cleanup on the test device: process groups, PTYs, relay tasks, and telemetry shutdown. Leftovers were suspected, not confirmed.
|
||||||
- [ ] Exercise collector unavailability and recovery with running agent/CC instances. Verify normal operations continue and telemetry queues remain bounded; outage losses are expected under the current best-effort policy.
|
- [ ] Exercise collector unavailability and recovery with running agent/CC instances. Verify normal operations continue and telemetry queues remain bounded; outage losses are expected under the current best-effort policy.
|
||||||
- [ ] Validate terminal recovery instrumentation live: generation timing/size, transport attach/resume, and backpressure episodes ending in drain or disconnect.
|
- [ ] Validate terminal recovery instrumentation live: generation timing/size, transport attach/resume, and backpressure episodes ending in drain or disconnect.
|
||||||
|
|||||||
+5
-1
@@ -96,7 +96,11 @@ existing debug events too; inspect them before enabling it broadly.
|
|||||||
- `relay_command` on CC propagates W3C context in an optional command field.
|
- `relay_command` on CC propagates W3C context in an optional command field.
|
||||||
`execute_command` on the agent adopts it; older messages/agents remain compatible.
|
`execute_command` on the agent adopts it; older messages/agents remain compatible.
|
||||||
- `collect_device_snapshot` and `store_device_snapshot` measure collection and
|
- `collect_device_snapshot` and `store_device_snapshot` measure collection and
|
||||||
persistence separately. Snapshot traces are not yet linked across the wire.
|
persistence as linked spans across the agent WebSocket. The snapshot carries
|
||||||
|
optional W3C context from collection; CC adopts it after authentication checks.
|
||||||
|
Periodic and requested syncs use the same path. Old messages and empty/invalid
|
||||||
|
context remain valid but do not create a cross-process link. Both processes
|
||||||
|
need tracing enabled and suitable filters/sampling to see both sides.
|
||||||
- Terminal create/attach/close/resume spans describe lifecycle operations, not
|
- Terminal create/attach/close/resume spans describe lifecycle operations, not
|
||||||
output chunks. Trace-linked OTLP logs use the SDK's tracing-context bridge.
|
output chunks. Trace-linked OTLP logs use the SDK's tracing-context bridge.
|
||||||
- Inventory source spans separate neighbor, DHCP lease, and hook-store reads
|
- Inventory source spans separate neighbor, DHCP lease, and hook-store reads
|
||||||
|
|||||||
@@ -301,6 +301,9 @@ pub enum ClientMessage {
|
|||||||
DeviceSnapshot {
|
DeviceSnapshot {
|
||||||
agent_id: String,
|
agent_id: String,
|
||||||
devices: Vec<Device>,
|
devices: Vec<Device>,
|
||||||
|
/// W3C trace context for the collection that produced this snapshot.
|
||||||
|
#[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
|
||||||
|
trace_context: std::collections::BTreeMap<String, String>,
|
||||||
},
|
},
|
||||||
Result {
|
Result {
|
||||||
request_id: RequestId,
|
request_id: RequestId,
|
||||||
@@ -427,11 +430,13 @@ mod tests {
|
|||||||
let msg = ClientMessage::DeviceSnapshot {
|
let msg = ClientMessage::DeviceSnapshot {
|
||||||
agent_id: "agent-a".into(),
|
agent_id: "agent-a".into(),
|
||||||
devices: vec![],
|
devices: vec![],
|
||||||
|
trace_context: Default::default(),
|
||||||
};
|
};
|
||||||
|
|
||||||
let json = serde_json::to_string(&msg).expect("serialize");
|
let json = serde_json::to_string(&msg).expect("serialize");
|
||||||
assert!(json.contains("\"type\":\"device_snapshot\""));
|
assert!(json.contains("\"type\":\"device_snapshot\""));
|
||||||
assert!(json.contains("\"devices\""));
|
assert!(json.contains("\"devices\""));
|
||||||
|
assert!(!json.contains("trace_context"));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
@@ -224,6 +224,7 @@ where
|
|||||||
&ClientMessage::DeviceSnapshot {
|
&ClientMessage::DeviceSnapshot {
|
||||||
agent_id: config.agent_id.clone(),
|
agent_id: config.agent_id.clone(),
|
||||||
devices: inventory.devices,
|
devices: inventory.devices,
|
||||||
|
trace_context: wakey_core::telemetry::current_trace_context(),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|||||||
@@ -38,6 +38,8 @@ enum IncomingClientMessage {
|
|||||||
DeviceSnapshot {
|
DeviceSnapshot {
|
||||||
agent_id: String,
|
agent_id: String,
|
||||||
devices: Vec<Device>,
|
devices: Vec<Device>,
|
||||||
|
#[serde(default)]
|
||||||
|
trace_context: std::collections::BTreeMap<String, String>,
|
||||||
},
|
},
|
||||||
Result {
|
Result {
|
||||||
request_id: RequestId,
|
request_id: RequestId,
|
||||||
@@ -305,13 +307,20 @@ async fn process_agent_text(
|
|||||||
ensure_current_session(state, &agent_id, connection_id).await?;
|
ensure_current_session(state, &agent_id, connection_id).await?;
|
||||||
debug!(agent_id = %agent_id, "heartbeat received");
|
debug!(agent_id = %agent_id, "heartbeat received");
|
||||||
}
|
}
|
||||||
IncomingClientMessage::DeviceSnapshot { agent_id, devices } => {
|
IncomingClientMessage::DeviceSnapshot {
|
||||||
|
agent_id,
|
||||||
|
devices,
|
||||||
|
trace_context,
|
||||||
|
} => {
|
||||||
if connection.authed_agent_id.as_deref() != Some(agent_id.as_str()) {
|
if connection.authed_agent_id.as_deref() != Some(agent_id.as_str()) {
|
||||||
anyhow::bail!("device_snapshot for unauthenticated or mismatched agent");
|
anyhow::bail!("device_snapshot for unauthenticated or mismatched agent");
|
||||||
}
|
}
|
||||||
ensure_current_session(state, &agent_id, connection_id).await?;
|
ensure_current_session(state, &agent_id, connection_id).await?;
|
||||||
let mut operation = wakey_core::telemetry::Operation::new("snapshot_store");
|
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());
|
let span = tracing::info_span!("store_device_snapshot", agent_id = %agent_id, devices = devices.len());
|
||||||
|
// Continue collection's trace rather than parenting persistence to
|
||||||
|
// the long-lived agent connection. Empty/invalid context is ignored.
|
||||||
|
wakey_core::telemetry::set_remote_parent(&span, &trace_context);
|
||||||
match tracing::Instrument::instrument(
|
match tracing::Instrument::instrument(
|
||||||
state
|
state
|
||||||
.store
|
.store
|
||||||
@@ -457,6 +466,42 @@ mod tests {
|
|||||||
AgentConnectionState, IncomingClientMessage, is_current_session, validate_auth_identity,
|
AgentConnectionState, IncomingClientMessage, is_current_session, validate_auth_identity,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn device_snapshot_trace_context_is_optional_and_cross_crate_compatible() {
|
||||||
|
let legacy: IncomingClientMessage =
|
||||||
|
serde_json::from_str(r#"{"type":"device_snapshot","agent_id":"agent-a","devices":[]}"#)
|
||||||
|
.unwrap();
|
||||||
|
let IncomingClientMessage::DeviceSnapshot { trace_context, .. } = legacy else {
|
||||||
|
panic!("expected snapshot");
|
||||||
|
};
|
||||||
|
assert!(trace_context.is_empty());
|
||||||
|
|
||||||
|
let expected = [(
|
||||||
|
"traceparent".to_string(),
|
||||||
|
"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01".to_string(),
|
||||||
|
)]
|
||||||
|
.into_iter()
|
||||||
|
.collect();
|
||||||
|
let message = wakey_agent::protocol::ClientMessage::DeviceSnapshot {
|
||||||
|
agent_id: "agent-a".into(),
|
||||||
|
devices: vec![],
|
||||||
|
trace_context: expected,
|
||||||
|
};
|
||||||
|
let IncomingClientMessage::DeviceSnapshot {
|
||||||
|
agent_id,
|
||||||
|
trace_context,
|
||||||
|
..
|
||||||
|
} = serde_json::from_str(&serde_json::to_string(&message).unwrap()).unwrap()
|
||||||
|
else {
|
||||||
|
panic!("expected snapshot");
|
||||||
|
};
|
||||||
|
assert_eq!(agent_id, "agent-a");
|
||||||
|
assert_eq!(
|
||||||
|
trace_context["traceparent"],
|
||||||
|
"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn current_session_check_rejects_stale_connection_ids() {
|
fn current_session_check_rejects_stale_connection_ids() {
|
||||||
let (tx, _rx) = mpsc::unbounded_channel();
|
let (tx, _rx) = mpsc::unbounded_channel();
|
||||||
|
|||||||
@@ -450,6 +450,50 @@ mod auth_tests {
|
|||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn trace_context_connects_collection_to_remote_persistence() {
|
||||||
|
let provider = SdkTracerProvider::builder().build();
|
||||||
|
let subscriber = tracing_subscriber::registry()
|
||||||
|
.with(tracing_opentelemetry::layer().with_tracer(provider.tracer("snapshot-test")));
|
||||||
|
// Scoped subscriber and synchronous entered guards keep this isolated
|
||||||
|
// from other tests without installing a global exporter or collector.
|
||||||
|
tracing::subscriber::with_default(subscriber, || {
|
||||||
|
let collect = tracing::info_span!("collect_device_snapshot");
|
||||||
|
let carrier = {
|
||||||
|
let _entered = collect.enter();
|
||||||
|
current_trace_context()
|
||||||
|
};
|
||||||
|
let collection = carrier["traceparent"].split('-').collect::<Vec<_>>();
|
||||||
|
let store = tracing::info_span!("store_device_snapshot");
|
||||||
|
set_remote_parent(&store, &carrier);
|
||||||
|
let persisted = {
|
||||||
|
let _entered = store.enter();
|
||||||
|
current_trace_context()
|
||||||
|
};
|
||||||
|
let persistence = persisted["traceparent"].split('-').collect::<Vec<_>>();
|
||||||
|
assert_eq!(collection[1], persistence[1], "trace IDs must match");
|
||||||
|
assert_ne!(
|
||||||
|
collection[2], persistence[2],
|
||||||
|
"spans must have distinct IDs"
|
||||||
|
);
|
||||||
|
|
||||||
|
let unrelated = tracing::info_span!("unrelated");
|
||||||
|
set_remote_parent(&unrelated, &BTreeMap::new());
|
||||||
|
set_remote_parent(
|
||||||
|
&unrelated,
|
||||||
|
&[("traceparent".into(), "invalid".into())]
|
||||||
|
.into_iter()
|
||||||
|
.collect(),
|
||||||
|
);
|
||||||
|
let _entered = unrelated.enter();
|
||||||
|
let local = current_trace_context();
|
||||||
|
assert_ne!(
|
||||||
|
collection[1],
|
||||||
|
local["traceparent"].split('-').nth(1).unwrap()
|
||||||
|
);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn authorization_supports_both_schemes_and_redacts_debug() {
|
fn authorization_supports_both_schemes_and_redacts_debug() {
|
||||||
for header in ["Basic dGVzdDp0ZXN0", "Bearer test-token"] {
|
for header in ["Basic dGVzdDp0ZXN0", "Bearer test-token"] {
|
||||||
|
|||||||
Reference in new issue
Block a user