diff --git a/Cargo.lock b/Cargo.lock index b662802..c9e3626 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -504,6 +504,8 @@ dependencies = [ "time", "tokio", "tokio-tungstenite", + "tracing", + "tracing-subscriber", ] [[package]] diff --git a/crates/crank-runtime/Cargo.toml b/crates/crank-runtime/Cargo.toml index ed1bef6..a616160 100644 --- a/crates/crank-runtime/Cargo.toml +++ b/crates/crank-runtime/Cargo.toml @@ -21,6 +21,7 @@ serde.workspace = true serde_json.workspace = true sha2.workspace = true thiserror.workspace = true +tracing.workspace = true [dev-dependencies] axum.workspace = true @@ -29,3 +30,4 @@ futures-util = "0.3" time.workspace = true tokio.workspace = true tokio-tungstenite.workspace = true +tracing-subscriber.workspace = true diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index f1a79eb..be9311a 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/crates/crank-runtime/src/executor.rs @@ -7,6 +7,7 @@ use crank_adapter_soap::{SoapAdapter, SoapRequest}; use crank_adapter_websocket::{WebsocketAdapter, WebsocketWindowRequest}; use crank_core::{ExecutionMode, Target, TransportBehavior}; use serde_json::{Map, Value, json}; +use tracing::debug; use crate::{ AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeOperation, @@ -75,6 +76,7 @@ impl RuntimeExecutor { resolved_auth: Option<&ResolvedAuth>, request_context: Option<&RuntimeRequestContext>, ) -> Result { + log_runtime_event("unary.execute", operation, request_context); let prepared_request = self.prepare_request(operation, input)?; let prepared_request = apply_resolved_auth(prepared_request, resolved_auth); self.execute_prepared(operation, prepared_request, request_context) @@ -117,6 +119,7 @@ impl RuntimeExecutor { resolved_auth: Option<&ResolvedAuth>, request_context: Option<&RuntimeRequestContext>, ) -> Result { + log_runtime_event("window.execute", operation, request_context); let Some(streaming) = operation.execution_config.streaming.as_ref() else { return Err(RuntimeError::MissingStreamingConfig { operation_id: operation.operation_id.as_str().to_owned(), @@ -185,6 +188,7 @@ impl RuntimeExecutor { resolved_auth: Option<&ResolvedAuth>, request_context: Option<&RuntimeRequestContext>, ) -> Result { + log_runtime_event("session.seed", operation, request_context); let Some(streaming) = operation.execution_config.streaming.as_ref() else { return Err(RuntimeError::MissingStreamingConfig { operation_id: operation.operation_id.as_str().to_owned(), @@ -253,6 +257,7 @@ impl RuntimeExecutor { prepared_request: PreparedRequest, request_context: Option<&RuntimeRequestContext>, ) -> Result { + log_runtime_event("adapter.dispatch", operation, request_context); let context_headers = runtime_context_headers(request_context); match &operation.target { Target::Grpc(target) => { @@ -356,6 +361,7 @@ impl RuntimeExecutor { prepared_request: PreparedRequest, request_context: Option<&RuntimeRequestContext>, ) -> Result { + log_runtime_event("window.adapter.dispatch", operation, request_context); let context_headers = runtime_context_headers(request_context); match &operation.target { Target::Rest(target) => { @@ -550,6 +556,28 @@ fn runtime_context_headers( .unwrap_or_default() } +fn log_runtime_event( + stage: &'static str, + operation: &RuntimeOperation, + request_context: Option<&RuntimeRequestContext>, +) { + let request_id = request_context + .map(|context| context.request_id.as_str()) + .unwrap_or_default(); + let correlation_id = request_context + .map(|context| context.correlation_id.as_str()) + .unwrap_or_default(); + + debug!( + stage, + operation_id = %operation.operation_id, + protocol = ?operation.protocol, + request_id, + correlation_id, + "runtime execution" + ); +} + fn read_string_map( value: Option<&Value>, field_name: &str, @@ -606,6 +634,8 @@ fn non_empty_payload(value: Option) -> Option { #[cfg(test)] mod tests { use std::collections::BTreeMap; + use std::io; + use std::sync::{Arc, Mutex}; use axum::{ Json, Router, @@ -632,6 +662,7 @@ mod tests { accept_async, accept_hdr_async, tungstenite::handshake::server::{Request, Response}, }; + use tracing_subscriber::{filter::LevelFilter, fmt::MakeWriter, prelude::*}; use crate::{ PreparedRequest, RuntimeError, RuntimeExecutor, RuntimeOperation, RuntimeRequestContext, @@ -641,6 +672,42 @@ mod tests { OffsetDateTime::parse(value, &Rfc3339).unwrap() } + #[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 executes_rest_operation_end_to_end() { let base_url = spawn_runtime_server().await; @@ -678,6 +745,49 @@ mod tests { assert_eq!(output, json!({ "id": "req_rest_123|corr_rest_123" })); } + #[tokio::test] + async fn emits_runtime_tracing_with_request_context() { + let base_url = spawn_runtime_server().await; + let executor = RuntimeExecutor::new(); + let mut operation = test_rest_operation(&base_url, false, false); + let context = RuntimeRequestContext::new("req_log_123", "corr_log_123"); + 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::DEBUG), + ); + let dispatch = tracing::Dispatch::new(subscriber); + + if let Target::Rest(target) = &mut operation.target { + target.path_template = "/leads-context".to_owned(); + } + + let _guard = tracing::dispatcher::set_default(&dispatch); + let output = executor + .execute_with_context( + &operation, + &json!({ "email": "user@example.com" }), + Some(&context), + ) + .await + .unwrap(); + + assert_eq!(output, json!({ "id": "req_log_123|corr_log_123" })); + + let logs = writer.output(); + assert!(logs.contains("runtime execution")); + assert!(logs.contains("unary.execute")); + assert!(logs.contains("adapter.dispatch")); + assert!(logs.contains("req_log_123")); + assert!(logs.contains("corr_log_123")); + assert!(logs.contains("op_rest_runtime")); + } + #[tokio::test] async fn executes_graphql_operation_end_to_end() { let endpoint = spawn_graphql_server().await;