camel_processor/
log.rs

1use camel_api::{CamelError, Exchange, IdentityProcessor, Value, ValueSource};
2use std::future::Future;
3use std::pin::Pin;
4use std::task::{Context, Poll};
5use tower::Service;
6use tracing::{debug, error, info, trace, warn};
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub enum LogLevel {
10    Trace,
11    Debug,
12    Info,
13    Warn,
14    Error,
15}
16
17#[derive(Clone)]
18pub struct LogProcessor {
19    inner: IdentityProcessor,
20    level: LogLevel,
21    message: String,
22}
23
24// TODO(PROC-004): Add metrics instrumentation — processed count, error count, latency histograms
25// are not yet instrumented on processors. Consider wiring MetricsCollector into LogProcessor and
26// incrementing a counter on each call, recording elapsed time, and tracking errors.
27
28impl LogProcessor {
29    pub fn new(level: LogLevel, message: String) -> Self {
30        Self {
31            inner: IdentityProcessor,
32            level,
33            message,
34        }
35    }
36}
37
38impl Service<Exchange> for LogProcessor {
39    type Response = Exchange;
40    type Error = CamelError;
41    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
42
43    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
44        self.inner.poll_ready(cx)
45    }
46
47    fn call(&mut self, exchange: Exchange) -> Self::Future {
48        let msg = self.message.clone();
49        let exchange_id = exchange.correlation_id.clone();
50        let body_preview = sanitize_preview(
51            &exchange
52                .input
53                .body
54                .as_text()
55                .unwrap_or("")
56                .chars()
57                .take(64)
58                .collect::<String>(),
59        );
60        debug!(exchange_id = %exchange_id, body_preview = %body_preview, "LogProcessor processing exchange");
61        match self.level {
62            LogLevel::Trace => trace!(exchange_id = %exchange_id, "{}", msg),
63            LogLevel::Debug => debug!(exchange_id = %exchange_id, "{}", msg),
64            LogLevel::Info => info!(exchange_id = %exchange_id, "{}", msg),
65            LogLevel::Warn => warn!(exchange_id = %exchange_id, "{}", msg),
66            // Deliberately unannotated: the `log` step is a user-output
67            // mechanism — the route selects this level — so the call sits
68            // outside ADR-0012's operational convention. The lint-log-levels
69            // exclusion is symbol-bound (structural, in scripts/xtask).
70            LogLevel::Error => error!(exchange_id = %exchange_id, "{}", msg),
71        }
72        self.inner.call(exchange)
73    }
74}
75
76/// A log processor that evaluates a message expression against the Exchange at call-time.
77/// Analogous to [`DynamicSetHeader`](crate::dynamic_set_header::DynamicSetHeader).
78///
79/// A failed evaluation fails the step BEFORE logging — no log record is
80/// emitted with a null message.
81#[derive(Clone)]
82pub struct DynamicLog {
83    inner: IdentityProcessor,
84    level: LogLevel,
85    source: ValueSource,
86}
87
88impl DynamicLog {
89    pub fn new(level: LogLevel, source: impl Into<ValueSource>) -> Self {
90        Self {
91            inner: IdentityProcessor,
92            level,
93            source: source.into(),
94        }
95    }
96}
97
98/// Render an evaluated log-message value: strings pass through unquoted,
99/// everything else uses its JSON representation.
100fn log_value_to_string(value: Value) -> String {
101    match value {
102        Value::String(s) => s,
103        other => other.to_string(),
104    }
105}
106
107impl Service<Exchange> for DynamicLog {
108    type Response = Exchange;
109    type Error = CamelError;
110    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
111
112    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
113        self.inner.poll_ready(cx)
114    }
115
116    fn call(&mut self, exchange: Exchange) -> Self::Future {
117        let source = self.source.clone();
118        let level = self.level;
119        Box::pin(async move {
120            let value = source.evaluate(&exchange).await?;
121            let exchange_id = exchange.correlation_id.clone();
122            let msg = sanitize_preview(&log_value_to_string(value));
123            match level {
124                LogLevel::Trace => trace!(exchange_id = %exchange_id, "{}", msg),
125                LogLevel::Debug => debug!(exchange_id = %exchange_id, "{}", msg),
126                LogLevel::Info => info!(exchange_id = %exchange_id, "{}", msg),
127                LogLevel::Warn => warn!(exchange_id = %exchange_id, "{}", msg),
128                // Deliberately unannotated: user-output mechanism, see
129                // `LogProcessor::call` (symbol-bound lint exclusion).
130                LogLevel::Error => error!(exchange_id = %exchange_id, "{}", msg),
131            }
132            Ok(exchange)
133        })
134    }
135}
136
137/// Strip control characters that could enable log injection.
138/// Replaces control chars with U+FFFD (replacement char) to preserve
139/// alignment while making injection visible.
140fn sanitize_preview(s: &str) -> String {
141    s.chars()
142        .map(|c| if c.is_control() { '\u{FFFD}' } else { c })
143        .collect()
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149    use camel_api::body::Body;
150    use camel_api::{Message, Value};
151    use tower::ServiceExt;
152
153    #[tokio::test]
154    async fn test_log_processor_passes_exchange_through() {
155        let mut processor = LogProcessor::new(LogLevel::Info, "test message".into());
156        let exchange = Exchange::default();
157        let result = processor.call(exchange).await;
158        assert!(result.is_ok());
159    }
160
161    #[tokio::test]
162    async fn test_log_processor_preserves_exchange_body() {
163        let mut processor = LogProcessor::new(LogLevel::Debug, "debug message".into());
164        let mut exchange = Exchange::default();
165        exchange.input.body = Body::Text("test body".into());
166        let result = processor.call(exchange).await.unwrap();
167        assert_eq!(result.input.body.as_text(), Some("test body"));
168    }
169
170    fn sync_source<F>(f: F) -> camel_api::ValueSource
171    where
172        F: Fn(&Exchange) -> camel_api::Value + Send + Sync + 'static,
173    {
174        camel_api::ValueSource::Sync(std::sync::Arc::new(f))
175    }
176
177    #[tokio::test]
178    async fn test_dynamic_log_evaluates_body() {
179        let svc = DynamicLog::new(
180            LogLevel::Info,
181            sync_source(|ex: &Exchange| {
182                camel_api::Value::String(format!("body={}", ex.input.body.as_text().unwrap_or("")))
183            }),
184        );
185        let exchange = Exchange::new(Message::new("hello"));
186        let result = svc.oneshot(exchange).await.unwrap();
187        // Exchange passes through unchanged
188        assert_eq!(result.input.body.as_text(), Some("hello"));
189    }
190
191    #[tokio::test]
192    async fn test_dynamic_log_evaluates_header() {
193        let svc = DynamicLog::new(
194            LogLevel::Info,
195            sync_source(|ex: &Exchange| {
196                let counter = ex
197                    .input
198                    .header("CamelTimerCounter")
199                    .and_then(|v| v.as_i64())
200                    .unwrap_or(0);
201                camel_api::Value::String(format!("{} World", counter))
202            }),
203        );
204        let mut msg = Message::new("");
205        msg.set_header("CamelTimerCounter", Value::Number(42.into()));
206        let exchange = Exchange::new(msg);
207        let result = svc.oneshot(exchange).await.unwrap();
208        // Exchange passes through unchanged
209        assert_eq!(
210            result.input.header("CamelTimerCounter"),
211            Some(&Value::Number(42.into()))
212        );
213    }
214
215    #[tokio::test]
216    async fn dynamic_log_error_fails_step() {
217        use std::sync::Arc;
218
219        use camel_api::{BoxValueFuture, ValueSource};
220
221        let source = ValueSource::Async(Arc::new(|_: &Exchange| {
222            Box::pin(async { Err(CamelError::ProcessorError("log boom".into())) }) as BoxValueFuture
223        }));
224        let svc = DynamicLog::new(LogLevel::Info, source);
225
226        let result = svc.oneshot(Exchange::new(Message::new("x"))).await;
227        assert!(
228            result.is_err(),
229            "a failed log-message expression must fail the step, not log a null message"
230        );
231    }
232
233    #[test]
234    fn body_preview_strips_control_chars() {
235        let body_with_injection = "no\ttab\nnl\rcr\0null\u{1B}esc";
236        let preview: String = body_with_injection.chars().take(64).collect();
237        let sanitized = sanitize_preview(&preview);
238        assert!(!sanitized.contains('\n'), "newlines must be replaced");
239        assert!(
240            !sanitized.contains('\r'),
241            "carriage returns must be replaced"
242        );
243        assert!(!sanitized.contains('\t'), "tabs must be replaced");
244        assert!(!sanitized.contains('\0'), "null must be replaced");
245        assert!(
246            !sanitized.contains('\u{1B}'),
247            "ESC (ANSI injection) must be replaced"
248        );
249        // U+FFFD replacement char must appear (replacement, not deletion)
250        assert!(
251            sanitized.contains('\u{FFFD}'),
252            "control chars must be replaced with U+FFFD, not deleted"
253        );
254    }
255
256    #[tokio::test]
257    async fn dynamic_log_sanitizes_control_chars() {
258        let svc = DynamicLog::new(
259            LogLevel::Info,
260            sync_source(|_ex: &Exchange| {
261                camel_api::Value::String("fake\nline\rinjection".to_string())
262            }),
263        );
264        // The exchange passes through; sanitization happens inside call().
265        // We verify the exchange is unchanged and the call succeeds — the
266        // sanitization itself is validated by body_preview_strips_control_chars.
267        let exchange = Exchange::default();
268        let result = svc.oneshot(exchange).await;
269        assert!(result.is_ok());
270    }
271
272    /// Run `f` with a thread-local subscriber and return the captured output.
273    fn capture_logs<F: FnOnce()>(f: F) -> String {
274        use std::io::Write;
275        use std::sync::{Arc, Mutex};
276
277        #[derive(Clone)]
278        struct Buf(Arc<Mutex<Vec<u8>>>);
279        impl Write for Buf {
280            fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
281                self.0.lock().unwrap().extend_from_slice(b); // allow-unwrap
282                Ok(b.len())
283            }
284            fn flush(&mut self) -> std::io::Result<()> {
285                Ok(())
286            }
287        }
288        impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Buf {
289            type Writer = Buf;
290            fn make_writer(&'a self) -> Buf {
291                self.clone()
292            }
293        }
294
295        let buf = Buf(Arc::new(Mutex::new(Vec::new())));
296        let subscriber = tracing_subscriber::fmt()
297            .with_writer(buf.clone())
298            .with_ansi(false)
299            .with_max_level(tracing::Level::TRACE)
300            .finish();
301        tracing::subscriber::with_default(subscriber, f);
302        let bytes = buf.0.lock().unwrap().clone(); // allow-unwrap
303        String::from_utf8(bytes).unwrap() // allow-unwrap
304    }
305
306    #[test]
307    fn log_processor_error_level_emits_error_not_warn() {
308        let out = capture_logs(|| {
309            let mut p = LogProcessor::new(LogLevel::Error, "boom-static".into());
310            drop(p.call(Exchange::default()));
311        });
312        assert!(
313            out.contains("ERROR") && out.contains("boom-static"),
314            "level error must emit ERROR, got: {out}"
315        );
316        assert!(!out.contains("WARN"), "must not downgrade to WARN: {out}");
317    }
318
319    #[test]
320    fn dynamic_log_error_level_emits_error_not_warn() {
321        let out = capture_logs(|| {
322            let mut svc = DynamicLog::new(
323                LogLevel::Error,
324                sync_source(|_ex: &Exchange| camel_api::Value::String("boom-dynamic".into())),
325            );
326            let fut = svc.call(Exchange::default());
327            futures::executor::block_on(fut).unwrap(); // allow-unwrap
328        });
329        assert!(
330            out.contains("ERROR") && out.contains("boom-dynamic"),
331            "level error must emit ERROR, got: {out}"
332        );
333        assert!(!out.contains("WARN"), "must not downgrade to WARN: {out}");
334    }
335
336    #[test]
337    fn log_processor_warn_level_still_emits_warn() {
338        let out = capture_logs(|| {
339            let mut p = LogProcessor::new(LogLevel::Warn, "w-static".into());
340            drop(p.call(Exchange::default()));
341        });
342        assert!(out.contains("WARN") && !out.contains("ERROR"), "got: {out}");
343    }
344}
camel_processor/log.rs

camel_processor/
log.rs
