api: log ingress request ids
This commit is contained in:
@@ -4,6 +4,7 @@ use axum::{
|
|||||||
middleware::Next,
|
middleware::Next,
|
||||||
response::Response,
|
response::Response,
|
||||||
};
|
};
|
||||||
|
use tracing::info;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
pub const REQUEST_ID_HEADER: HeaderName = HeaderName::from_static("x-request-id");
|
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 {
|
let context = RequestContext {
|
||||||
request_id: resolve_request_id(request.headers()),
|
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());
|
request.extensions_mut().insert(context.clone());
|
||||||
|
|
||||||
let mut response = next.run(request).await;
|
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) {
|
if let Ok(value) = HeaderValue::from_str(&context.request_id) {
|
||||||
response.headers_mut().insert(REQUEST_ID_HEADER, value);
|
response.headers_mut().insert(REQUEST_ID_HEADER, value);
|
||||||
}
|
}
|
||||||
@@ -47,7 +57,15 @@ fn is_valid_request_id(value: &str) -> bool {
|
|||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
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]
|
#[test]
|
||||||
fn accepts_visible_ascii_request_ids() {
|
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 value"));
|
||||||
assert!(!is_valid_request_id("bad\nvalue"));
|
assert!(!is_valid_request_id("bad\nvalue"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Default)]
|
||||||
|
struct SharedLogWriter {
|
||||||
|
buffer: Arc<Mutex<Vec<u8>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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<Mutex<Vec<u8>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl io::Write for SharedLogGuard {
|
||||||
|
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
|
||||||
|
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"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -38,6 +38,7 @@ use serde::{Deserialize, Serialize};
|
|||||||
use serde_json::{Value, json};
|
use serde_json::{Value, json};
|
||||||
use sha2::{Digest, Sha256};
|
use sha2::{Digest, Sha256};
|
||||||
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
|
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
|
||||||
|
use tracing::info;
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
catalog::PublishedToolCatalog,
|
catalog::PublishedToolCatalog,
|
||||||
@@ -240,6 +241,13 @@ async fn mcp_post(
|
|||||||
Json(message): Json<Value>,
|
Json(message): Json<Value>,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
let transport_request_id = resolve_request_id(&headers);
|
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("<unknown>"),
|
||||||
|
"mcp request received"
|
||||||
|
);
|
||||||
|
|
||||||
if let Err(status) = validate_origin(&state.allowed_origins, &headers) {
|
if let Err(status) = validate_origin(&state.allowed_origins, &headers) {
|
||||||
return with_request_id_header(status.into_response(), &transport_request_id);
|
return with_request_id_header(status.into_response(), &transport_request_id);
|
||||||
|
|||||||
+123
-1
@@ -79,7 +79,12 @@ fn database_options_from_env() -> Result<PgConnectOptions, Box<dyn std::error::E
|
|||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use std::{collections::BTreeMap, env, time::Duration};
|
use std::{
|
||||||
|
collections::BTreeMap,
|
||||||
|
env, io,
|
||||||
|
sync::{Arc, Mutex},
|
||||||
|
time::Duration,
|
||||||
|
};
|
||||||
|
|
||||||
use axum::{
|
use axum::{
|
||||||
Json, Router,
|
Json, Router,
|
||||||
@@ -110,6 +115,7 @@ mod tests {
|
|||||||
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
|
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
|
||||||
use tokio::net::TcpListener;
|
use tokio::net::TcpListener;
|
||||||
use tokio::time::sleep;
|
use tokio::time::sleep;
|
||||||
|
use tracing_subscriber::{filter::LevelFilter, fmt::MakeWriter, prelude::*};
|
||||||
|
|
||||||
use crate::app::build_app;
|
use crate::app::build_app;
|
||||||
|
|
||||||
@@ -117,6 +123,42 @@ mod tests {
|
|||||||
WorkspaceId::new("ws_default")
|
WorkspaceId::new("ws_default")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Default)]
|
||||||
|
struct SharedLogWriter {
|
||||||
|
buffer: Arc<Mutex<Vec<u8>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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<Mutex<Vec<u8>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl io::Write for SharedLogGuard {
|
||||||
|
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
|
||||||
|
self.buffer.lock().unwrap().extend_from_slice(bytes);
|
||||||
|
Ok(bytes.len())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn flush(&mut self) -> io::Result<()> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn test_workspace_slug() -> &'static str {
|
fn test_workspace_slug() -> &'static str {
|
||||||
"default"
|
"default"
|
||||||
}
|
}
|
||||||
@@ -410,6 +452,86 @@ mod tests {
|
|||||||
assert_eq!(logs[0].log.request_id.as_deref(), Some(request_id.as_str()));
|
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]
|
#[tokio::test]
|
||||||
async fn initializes_and_calls_published_graphql_tool_via_mcp() {
|
async fn initializes_and_calls_published_graphql_tool_via_mcp() {
|
||||||
let registry = test_registry().await;
|
let registry = test_registry().await;
|
||||||
|
|||||||
Reference in New Issue
Block a user