From 3d2d97c1e68dc0db268887274894bf8e4376ae52 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Fri, 1 May 2026 11:57:58 +0000 Subject: [PATCH] api: log ingress request ids --- apps/admin-api/src/request_context.rs | 103 ++++++++++++++++++++- apps/mcp-server/src/app.rs | 8 ++ apps/mcp-server/src/main.rs | 124 +++++++++++++++++++++++++- 3 files changed, 233 insertions(+), 2 deletions(-) diff --git a/apps/admin-api/src/request_context.rs b/apps/admin-api/src/request_context.rs index 3b1ec06..cda184e 100644 --- a/apps/admin-api/src/request_context.rs +++ b/apps/admin-api/src/request_context.rs @@ -4,6 +4,7 @@ use axum::{ middleware::Next, response::Response, }; +use tracing::info; use uuid::Uuid; pub const REQUEST_ID_HEADER: HeaderName = HeaderName::from_static("x-request-id"); @@ -18,9 +19,18 @@ pub async fn apply_request_context(mut request: Request, next: Next) -> Response let context = RequestContext { request_id: resolve_request_id(request.headers()), }; + let method = request.method().clone(); + let path = request.uri().path().to_owned(); request.extensions_mut().insert(context.clone()); let mut response = next.run(request).await; + info!( + request_id = %context.request_id, + method = %method, + path, + status = response.status().as_u16(), + "admin request completed" + ); if let Ok(value) = HeaderValue::from_str(&context.request_id) { response.headers_mut().insert(REQUEST_ID_HEADER, value); } @@ -47,7 +57,15 @@ fn is_valid_request_id(value: &str) -> bool { #[cfg(test)] mod tests { - use super::is_valid_request_id; + use std::io; + use std::sync::{Arc, Mutex}; + + use axum::{Router, routing::get}; + use reqwest::Client; + use tokio::net::TcpListener; + use tracing_subscriber::{filter::LevelFilter, fmt::MakeWriter, prelude::*}; + + use super::{REQUEST_ID_HEADER, apply_request_context, is_valid_request_id}; #[test] fn accepts_visible_ascii_request_ids() { @@ -61,4 +79,87 @@ mod tests { assert!(!is_valid_request_id("bad value")); assert!(!is_valid_request_id("bad\nvalue")); } + + #[derive(Clone, Default)] + struct SharedLogWriter { + buffer: Arc>>, + } + + impl SharedLogWriter { + fn output(&self) -> String { + String::from_utf8(self.buffer.lock().unwrap().clone()).unwrap() + } + } + + impl<'a> MakeWriter<'a> for SharedLogWriter { + type Writer = SharedLogGuard; + + fn make_writer(&'a self) -> Self::Writer { + SharedLogGuard { + buffer: Arc::clone(&self.buffer), + } + } + } + + struct SharedLogGuard { + buffer: Arc>>, + } + + impl io::Write for SharedLogGuard { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.buffer.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + #[tokio::test] + async fn logs_request_completion_with_request_id() { + let writer = SharedLogWriter::default(); + let subscriber = tracing_subscriber::registry().with( + tracing_subscriber::fmt::layer() + .with_writer(writer.clone()) + .without_time() + .with_ansi(false) + .with_target(false) + .compact() + .with_filter(LevelFilter::INFO), + ); + let dispatch = tracing::Dispatch::new(subscriber); + let app = Router::new() + .route("/probe", get(|| async { "ok" })) + .layer(axum::middleware::from_fn(apply_request_context)); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + + let _guard = tracing::dispatcher::set_default(&dispatch); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let response = Client::new() + .get(format!("http://{address}/probe")) + .header(REQUEST_ID_HEADER.as_str(), "req_admin_trace_123") + .send() + .await + .unwrap(); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + assert_eq!( + response.headers()[REQUEST_ID_HEADER.as_str()] + .to_str() + .unwrap(), + "req_admin_trace_123" + ); + + let logs = writer.output(); + assert!(logs.contains("admin request completed")); + assert!(logs.contains("req_admin_trace_123")); + assert!(logs.contains("GET")); + assert!(logs.contains("/probe")); + assert!(logs.contains("status=200")); + } } diff --git a/apps/mcp-server/src/app.rs b/apps/mcp-server/src/app.rs index 57b072b..0c53833 100644 --- a/apps/mcp-server/src/app.rs +++ b/apps/mcp-server/src/app.rs @@ -38,6 +38,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; +use tracing::info; use crate::{ catalog::PublishedToolCatalog, @@ -240,6 +241,13 @@ async fn mcp_post( Json(message): Json, ) -> Response { let transport_request_id = resolve_request_id(&headers); + info!( + request_id = %transport_request_id, + workspace_slug = %path.workspace_slug, + agent_slug = %path.agent_slug, + jsonrpc_method = method_name(&message).unwrap_or(""), + "mcp request received" + ); if let Err(status) = validate_origin(&state.allowed_origins, &headers) { return with_request_id_header(status.into_response(), &transport_request_id); diff --git a/apps/mcp-server/src/main.rs b/apps/mcp-server/src/main.rs index 23a9a26..2a5514e 100644 --- a/apps/mcp-server/src/main.rs +++ b/apps/mcp-server/src/main.rs @@ -79,7 +79,12 @@ fn database_options_from_env() -> Result>>, + } + + impl SharedLogWriter { + fn output(&self) -> String { + String::from_utf8(self.buffer.lock().unwrap().clone()).unwrap() + } + } + + impl<'a> MakeWriter<'a> for SharedLogWriter { + type Writer = SharedLogGuard; + + fn make_writer(&'a self) -> Self::Writer { + SharedLogGuard { + buffer: Arc::clone(&self.buffer), + } + } + } + + struct SharedLogGuard { + buffer: Arc>>, + } + + impl io::Write for SharedLogGuard { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.buffer.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + fn test_workspace_slug() -> &'static str { "default" } @@ -410,6 +452,86 @@ mod tests { assert_eq!(logs[0].log.request_id.as_deref(), Some(request_id.as_str())); } + #[tokio::test] + async fn emits_request_id_in_mcp_ingress_logs() { + let registry = test_registry().await; + let upstream_base_url = spawn_upstream_server().await; + let operation = test_operation(&upstream_base_url, "crm_request_trace"); + + registry + .create_operation(&test_workspace_id(), &operation, Some("alice")) + .await + .unwrap(); + registry + .publish_operation(PublishRequest { + workspace_id: &test_workspace_id(), + operation_id: &operation.id, + version: 1, + published_at: &OffsetDateTime::parse("2026-03-26T10:00:00Z", &Rfc3339).unwrap(), + published_by: Some("alice"), + }) + .await + .unwrap(); + publish_agent_for_operation(®istry, &operation, "sales-request-trace").await; + let api_key = create_platform_api_key( + ®istry, + "mcp-request-trace", + &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write], + ) + .await; + + let base_url = spawn_mcp_server(build_test_app( + registry, + Duration::from_millis(0), + Some("https://crank.example.com".to_owned()), + )) + .await; + let client = reqwest::Client::new(); + let mcp_url = agent_mcp_url(&base_url, "sales-request-trace"); + let writer = SharedLogWriter::default(); + let subscriber = tracing_subscriber::registry().with( + tracing_subscriber::fmt::layer() + .with_writer(writer.clone()) + .without_time() + .with_ansi(false) + .with_target(false) + .compact() + .with_filter(LevelFilter::INFO), + ); + let dispatch = tracing::Dispatch::new(subscriber); + + let _guard = tracing::dispatcher::set_default(&dispatch); + let response = post_jsonrpc_response( + &client, + &mcp_url, + &api_key, + None, + Some("req_mcp_trace_123"), + json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-03-26" + } + }), + ) + .await; + + assert_eq!( + response.headers()["x-request-id"].to_str().unwrap(), + "req_mcp_trace_123" + ); + assert_eq!(response.status(), reqwest::StatusCode::OK); + + let logs = writer.output(); + assert!(logs.contains("mcp request received")); + assert!(logs.contains("req_mcp_trace_123")); + assert!(logs.contains("sales-request-trace")); + assert!(logs.contains("default")); + assert!(logs.contains("initialize")); + } + #[tokio::test] async fn initializes_and_calls_published_graphql_tool_via_mcp() { let registry = test_registry().await;