dataflow_rs/engine/functions/
log.rs

1use crate::engine::error::Result;
2use crate::engine::executor::{ArenaContext, with_arena};
3use crate::engine::message::{Change, Message};
4use crate::engine::task_outcome::TaskOutcome;
5use datalogic_rs::{Engine, Logic};
6use datavalue::DataValue;
7use log::{debug, error, info, trace, warn};
8use serde::Deserialize;
9use serde_json::Value;
10use std::collections::HashMap;
11use std::sync::Arc;
12
13/// Log levels supported by the log function
14#[derive(Debug, Clone, Default, Deserialize)]
15#[serde(rename_all = "lowercase")]
16pub enum LogLevel {
17    Trace,
18    Debug,
19    #[default]
20    Info,
21    Warn,
22    Error,
23}
24
25impl LogLevel {
26    /// The `log` crate level this variant emits at.
27    fn as_log_level(&self) -> log::Level {
28        match self {
29            LogLevel::Trace => log::Level::Trace,
30            LogLevel::Debug => log::Level::Debug,
31            LogLevel::Info => log::Level::Info,
32            LogLevel::Warn => log::Level::Warn,
33            LogLevel::Error => log::Level::Error,
34        }
35    }
36}
37
38/// Configuration for the log function.
39///
40/// The message and field expressions are pre-compiled at startup.
41#[derive(Debug, Clone, Deserialize)]
42pub struct LogConfig {
43    /// Log level to emit at
44    #[serde(default)]
45    pub level: LogLevel,
46
47    /// JSONLogic expression to produce the log message string
48    pub message: Value,
49
50    /// Additional structured fields: each value is a JSONLogic expression
51    #[serde(default)]
52    pub fields: HashMap<String, Value>,
53
54    /// Pre-compiled `message` JSONLogic, populated by `LogicCompiler`.
55    #[serde(skip)]
56    pub compiled_message: Option<Arc<Logic>>,
57
58    /// Pre-compiled JSONLogic for each `fields` entry, populated by
59    /// `LogicCompiler`. The inner `Option` is `None` for fields whose logic
60    /// failed to compile (logged at engine construction).
61    #[serde(skip)]
62    pub compiled_fields: Vec<(String, Option<Arc<Logic>>)>,
63}
64
65impl LogConfig {
66    /// Execute the log function, opening a fresh thread-local arena scope.
67    ///
68    /// Use this entry point when calling `LogConfig` outside an existing
69    /// `with_arena` scope (direct API users, tests). Inside a workflow sync
70    /// stretch the dispatch goes through [`Self::execute_in_arena`] to reuse
71    /// the cached `ArenaContext` and avoid a redundant `to_arena` walk.
72    pub fn execute(
73        &self,
74        message: &mut Message,
75        engine: &Arc<Engine>,
76    ) -> Result<(TaskOutcome, Vec<Change>)> {
77        with_arena(|arena| {
78            let mut arena_ctx = ArenaContext::from_owned(&message.context, arena);
79            self.execute_in_arena(message, &mut arena_ctx, engine)
80        })
81    }
82
83    /// Execute against an externally-provided `ArenaContext` so the cached
84    /// arena form of `message.context` (built once at the top of the workflow
85    /// sync stretch) is reused across every JSONLogic eval performed here.
86    pub(crate) fn execute_in_arena(
87        &self,
88        _message: &mut Message,
89        arena_ctx: &mut ArenaContext<'_>,
90        engine: &Arc<Engine>,
91    ) -> Result<(TaskOutcome, Vec<Change>)> {
92        // If the target level is filtered out by the active logger, every
93        // JSONLogic eval and string alloc below would feed a message the
94        // logger drops on arrival. Skip the whole task.
95        if !log::log_enabled!(target: "dataflow::log", self.level.as_log_level()) {
96            return Ok((TaskOutcome::Success, vec![]));
97        }
98
99        let arena = arena_ctx.arena();
100        let ctx_av = arena_ctx.as_data_value();
101
102        // Stringify a single eval result. For the common String case we copy
103        // the str directly; otherwise the JSON emitter writes straight from
104        // `&DataValue<'_>` without an intermediate `to_owned()` deep clone.
105        let stringify = |compiled: &Logic| -> String {
106            match engine.evaluate(compiled, ctx_av, arena) {
107                Ok(DataValue::String(s)) => (*s).to_string(),
108                Ok(other) => other.to_string(),
109                Err(e) => {
110                    error!("Log: Failed to evaluate expression: {:?}", e);
111                    "<eval error>".to_string()
112                }
113            }
114        };
115
116        let log_message = match &self.compiled_message {
117            Some(compiled) => stringify(compiled),
118            None => "<uncompiled message>".to_string(),
119        };
120
121        let mut field_parts = Vec::with_capacity(self.compiled_fields.len());
122        for (key, compiled_opt) in &self.compiled_fields {
123            let val = match compiled_opt {
124                Some(compiled) => stringify(compiled),
125                None => "<uncompiled>".to_string(),
126            };
127            field_parts.push(format!("{}={}", key, val));
128        }
129
130        let full_message = if field_parts.is_empty() {
131            log_message
132        } else {
133            format!("{} [{}]", log_message, field_parts.join(", "))
134        };
135
136        match self.level {
137            LogLevel::Trace => trace!(target: "dataflow::log", "{}", full_message),
138            LogLevel::Debug => debug!(target: "dataflow::log", "{}", full_message),
139            LogLevel::Info => info!(target: "dataflow::log", "{}", full_message),
140            LogLevel::Warn => warn!(target: "dataflow::log", "{}", full_message),
141            LogLevel::Error => error!(target: "dataflow::log", "{}", full_message),
142        }
143
144        // Log function never modifies message, never fails
145        Ok((TaskOutcome::Success, vec![]))
146    }
147}
dataflow_rs/engine/functions/log.rs

dataflow_rs/engine/functions/
log.rs
