chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
Each MCP tool call is logged, named, in the trace of its request
crates/knotless/src/telemetry.rs+21−0
| @@ -237,6 +237,21 @@ pub fn trace_id() -> Option<String> { | ||
TRACE.try_with(String::clone).ok() | ||
} | ||
| ||
/// The trace a request is served in, as a request extension, for what | ||
/// serves it in a task of its own, out of reach of the task-local trace. | ||
#[derive(Clone, Debug)] | ||
pub struct Trace(pub String); | ||
| ||
impl Trace { | ||
/// The trace `given` names, or a new one, for what is served with none. | ||
pub fn or_new(given: Option<&Trace>) -> String { | ||
match given { | ||
Some(trace) => trace.0.clone(), | ||
None => hex(16), | ||
} | ||
} | ||
} | ||
| ||
/// Runs `f` in the trace `id`. | ||
pub async fn traced<F: Future>(id: String, f: F) -> F::Output { | ||
TRACE.scope(id, f).await | ||
| @@ -935,6 +950,12 @@ mod tests { | ||
assert_eq!(stats.call("new", "200", 1, false, false), "other"); | ||
} | ||
| ||
#[test] | ||
fn traces_handed_on() { | ||
assert_eq!(Trace::or_new(Some(&Trace("t".into()))), "t"); | ||
assert_eq!(Trace::or_new(None).len(), 32); | ||
} | ||
| ||
/// A long poll is never slow, however long it waits. | ||
#[test] | ||
fn long_polls_are_not_slow() { | ||
crates/knotless-mcp/src/http.rs+20−14
| @@ -618,12 +618,15 @@ impl ServerHandler for Remote { | ||
request: CallToolRequestParams, | ||
context: RequestContext<RoleServer>, | ||
) -> Result<CallToolResponse, McpError> { | ||
let did = context | ||
.extensions | ||
.get::<Parts>() | ||
let parts = context.extensions.get::<Parts>(); | ||
let did = parts | ||
.and_then(|p| p.extensions.get::<Caller>()) | ||
.map(|c| c.0.clone()) | ||
.unwrap_or_default(); | ||
// rmcp calls the tool in a task of its own: the request's trace, | ||
// handed on, so what the tool does is in it. | ||
let trace = parts.and_then(|p| p.extensions.get::<telemetry::Trace>()); | ||
let trace = telemetry::Trace::or_new(trace); | ||
let server = Server::new(self.mcp.ctx_for(&did)); | ||
let (started, tool) = (Instant::now(), request.name.to_string()); | ||
let waits = request | ||
| @@ -634,7 +637,7 @@ impl ServerHandler for Remote { | ||
.unwrap_or_default() | ||
> 0; | ||
let res = match self.mcp.turn(&did, waits, started) { | ||
Ok(_turn) => server.call_tool(request, context).await, | ||
Ok(_turn) => telemetry::traced(trace.clone(), server.call_tool(request, context)).await, | ||
Err(why) => Ok(CallToolResponse::Complete(CallToolResult::error(vec![ | ||
ContentBlock::text(why), | ||
]))), | ||
| @@ -650,16 +653,19 @@ impl ServerHandler for Remote { | ||
Err(err) => Some(err.message.to_string()), | ||
}; | ||
let took = started.elapsed(); | ||
if let Some(error) = &failure { | ||
let f = Fields { | ||
did: Some(&did), | ||
tool: Some(&tool), | ||
error: Some(error.clone()), | ||
duration_ms: Some(telemetry::millis(took)), | ||
..Fields::default() | ||
}; | ||
telemetry::note(Level::WARN, "mcp tool failed", &f); | ||
} | ||
let f = Fields { | ||
did: Some(&did), | ||
tool: Some(&tool), | ||
error: failure.clone(), | ||
duration_ms: Some(telemetry::millis(took)), | ||
..Fields::default() | ||
}; | ||
// A line a call, named for its tool, in the request's trace: at | ||
// warn when it failed, or was slow and not asked to wait. | ||
let warn = failure.is_some() || (telemetry::slow(took) && !waits); | ||
let level = [Level::INFO, Level::WARN][usize::from(warn)]; | ||
let message = ["mcp tool", "mcp tool failed"][usize::from(failure.is_some())]; | ||
telemetry::traced(trace, async { telemetry::note(level, message, &f) }).await; | ||
self.mcp.tell(Event { | ||
name: "mcp.tool", | ||
subject: &tool, | ||
crates/knotless-mcp/src/lib.rs+1−1
| @@ -826,7 +826,7 @@ pub struct Logs { | ||
/// `x-trace-id` and each of its lines carry. | ||
pub trace: Option<String>, | ||
/// Only lines with this message: `request`, `outgoing`, `sql`, `mcp | ||
/// tool failed`, `git`. | ||
/// tool`, `mcp tool failed`, `git`. | ||
pub message: Option<String>, | ||
/// Only lines at this level or above: `error`, `warn`, `info`, `debug`. | ||
pub level: Option<String>, | ||
crates/knotless-serve/src/telemetry.rs+1−0
| @@ -72,6 +72,7 @@ async fn observe(mut req: Request, next: Next) -> Response { | ||
let agent = agent.map(str::to_string); | ||
let parent = req.headers().get("traceparent"); | ||
let trace = telemetry::trace_from(parent.and_then(|v| v.to_str().ok())); | ||
req.extensions_mut().insert(telemetry::Trace(trace.clone())); | ||
let mut res = telemetry::traced(trace.clone(), next.run(req)).await; | ||
let id = HeaderValue::try_from(trace.as_str()).expect("hex is a header value"); | ||
res.headers_mut().insert("x-trace-id", id); | ||
docs/observability.md+6−0
| @@ -17,6 +17,12 @@ to read it is in [deploy/README.md](../deploy/README.md#logs). | ||
over one kind of line, not a join across many. Slow ones (over | ||
`KNOTLESS_SLOW_MS`, 1000) are at warn, but for long polls, which wait | ||
on purpose; 5xx at error with the cause. | ||
- **A line for each MCP tool call**: `mcp tool`, with the tool's name | ||
(`gen_ai.tool.name`), the caller's DID and `duration_ms`, at info, or | ||
at warn when slow and not asked to wait; `mcp tool failed`, at warn, | ||
with why. The MCP server calls each tool in a task of its own, so the | ||
request's trace is handed to it: the call's line, and its calls and | ||
slow statements, are in the trace of the `/mcp` request that made it. | ||
- **A line for each call made**, to a PDS, the PLC directory, Bluesky or | ||
a handle's host, from the one place every call goes through | ||
(`knotless::telemetry::send`): host, path, XRPC method, status, error, | ||
lexicons/dev/knotless/getLogs.json+1−0
| @@ -20,6 +20,7 @@ | ||
"outgoing", | ||
"sql", | ||
"git", | ||
"mcp tool", | ||
"mcp tool failed", | ||
"logs read" | ||
] | ||