vector/transforms/lua/v2/
mod.rs

1use std::{path::PathBuf, sync::Arc, time::Duration};
2
3use serde_with::serde_as;
4use snafu::{ResultExt, Snafu};
5use vector_lib::codecs::MetricTagValues;
6use vector_lib::configurable::configurable_component;
7pub use vector_lib::event::lua;
8use vector_lib::transform::runtime_transform::{RuntimeTransform, Timer};
9
10use crate::config::{ComponentKey, OutputId};
11use crate::event::lua::event::LuaEvent;
12use crate::schema::Definition;
13use crate::{
14    config::{self, DataType, Input, TransformOutput, CONFIG_PATHS},
15    event::Event,
16    internal_events::{LuaBuildError, LuaGcTriggered},
17    schema,
18    transforms::Transform,
19};
20
21#[derive(Debug, Snafu)]
22pub enum BuildError {
23    #[snafu(display("Invalid \"search_dirs\": {}", source))]
24    InvalidSearchDirs { source: mlua::Error },
25    #[snafu(display("Cannot evaluate Lua code in \"source\": {}", source))]
26    InvalidSource { source: mlua::Error },
27
28    #[snafu(display("Cannot evaluate Lua code defining \"hooks.init\": {}", source))]
29    InvalidHooksInit { source: mlua::Error },
30    #[snafu(display("Cannot evaluate Lua code defining \"hooks.process\": {}", source))]
31    InvalidHooksProcess { source: mlua::Error },
32    #[snafu(display("Cannot evaluate Lua code defining \"hooks.shutdown\": {}", source))]
33    InvalidHooksShutdown { source: mlua::Error },
34    #[snafu(display("Cannot evaluate Lua code defining timer handler: {}", source))]
35    InvalidTimerHandler { source: mlua::Error },
36
37    #[snafu(display("Runtime error in \"hooks.init\" function: {}", source))]
38    RuntimeErrorHooksInit { source: mlua::Error },
39    #[snafu(display("Runtime error in \"hooks.process\" function: {}", source))]
40    RuntimeErrorHooksProcess { source: mlua::Error },
41    #[snafu(display("Runtime error in \"hooks.shutdown\" function: {}", source))]
42    RuntimeErrorHooksShutdown { source: mlua::Error },
43    #[snafu(display("Runtime error in timer handler: {}", source))]
44    RuntimeErrorTimerHandler { source: mlua::Error },
45
46    #[snafu(display("Cannot call GC in Lua runtime: {}", source))]
47    RuntimeErrorGc { source: mlua::Error },
48}
49
50/// Configuration for the version two of the `lua` transform.
51#[configurable_component]
52#[derive(Clone, Debug)]
53#[serde(deny_unknown_fields)]
54pub struct LuaConfig {
55    /// The Lua program to initialize the transform with.
56    ///
57    /// The program can be used to import external dependencies, as well as define the functions
58    /// used for the various lifecycle hooks. However, it's not strictly required, as the lifecycle
59    /// hooks can be configured directly with inline Lua source for each respective hook.
60    #[configurable(metadata(
61        docs::examples = "function init()\n\tcount = 0\nend\n\nfunction process()\n\tcount = count + 1\nend\n\nfunction timer_handler(emit)\n\temit(make_counter(counter))\n\tcounter = 0\nend\n\nfunction shutdown(emit)\n\temit(make_counter(counter))\nend\n\nfunction make_counter(value)\n\treturn metric = {\n\t\tname = \"event_counter\",\n\t\tkind = \"incremental\",\n\t\ttimestamp = os.date(\"!*t\"),\n\t\tcounter = {\n\t\t\tvalue = value\n\t\t}\n \t}\nend",
62        docs::examples = "-- external file with hooks and timers defined\nrequire('custom_module')",
63    ))]
64    source: Option<String>,
65
66    /// A list of directories to search when loading a Lua file via the `require` function.
67    ///
68    /// If not specified, the modules are looked up in the configuration directories.
69    #[serde(default = "default_config_paths")]
70    #[configurable(metadata(docs::examples = "/etc/vector/lua"))]
71    #[configurable(metadata(docs::human_name = "Search Directories"))]
72    search_dirs: Vec<PathBuf>,
73
74    #[configurable(derived)]
75    hooks: HooksConfig,
76
77    /// A list of timers which should be configured and executed periodically.
78    #[serde(default)]
79    timers: Vec<TimerConfig>,
80
81    /// When set to `single`, metric tag values are exposed as single strings, the
82    /// same as they were before this config option. Tags with multiple values show the last assigned value, and null values
83    /// are ignored.
84    ///
85    /// When set to `full`, all metric tags are exposed as arrays of either string or null
86    /// values.
87    #[serde(default)]
88    metric_tag_values: MetricTagValues,
89}
90
91fn default_config_paths() -> Vec<PathBuf> {
92    match CONFIG_PATHS.lock().ok() {
93        Some(config_paths) => config_paths
94            .clone()
95            .into_iter()
96            .map(|config_path| match config_path {
97                config::ConfigPath::File(mut path, _format) => {
98                    path.pop();
99                    path
100                }
101                config::ConfigPath::Dir(path) => path,
102            })
103            .collect(),
104        None => vec![],
105    }
106}
107
108/// Lifecycle hooks.
109///
110/// These hooks can be set to perform additional processing during the lifecycle of the transform.
111#[configurable_component]
112#[derive(Clone, Debug)]
113#[serde(deny_unknown_fields)]
114struct HooksConfig {
115    /// The function called when the first event comes in, before `hooks.process` is called.
116    ///
117    /// It can produce new events using the `emit` function.
118    ///
119    /// This can either be inline Lua that defines a closure to use, or the name of the Lua function to call. In both
120    /// cases, the closure/function takes a single parameter, `emit`, which is a reference to a function for emitting events.
121    #[configurable(metadata(
122        docs::examples = "function (emit)\n\t-- Custom Lua code here\nend",
123        docs::examples = "init",
124    ))]
125    init: Option<String>,
126
127    /// The function called for each incoming event.
128    ///
129    /// It can produce new events using the `emit` function.
130    ///
131    /// This can either be inline Lua that defines a closure to use, or the name of the Lua function to call. In both
132    /// cases, the closure/function takes two parameters. The first parameter, `event`, is the event being processed,
133    /// while the second parameter, `emit`, is a reference to a function for emitting events.
134    #[configurable(metadata(
135        docs::examples = "function (event, emit)\n\tevent.log.field = \"value\" -- set value of a field\n\tevent.log.another_field = nil -- remove field\n\tevent.log.first, event.log.second = nil, event.log.first -- rename field\n\t-- Very important! Emit the processed event.\n\temit(event)\nend",
136        docs::examples = "process",
137    ))]
138    process: String,
139
140    /// The function called when the transform is stopped.
141    ///
142    /// It can produce new events using the `emit` function.
143    ///
144    /// This can either be inline Lua that defines a closure to use, or the name of the Lua function to call. In both
145    /// cases, the closure/function takes a single parameter, `emit`, which is a reference to a function for emitting events.
146    #[configurable(metadata(
147        docs::examples = "function (emit)\n\t-- Custom Lua code here\nend",
148        docs::examples = "shutdown",
149    ))]
150    shutdown: Option<String>,
151}
152
153/// A Lua timer.
154#[serde_as]
155#[configurable_component]
156#[derive(Clone, Debug)]
157struct TimerConfig {
158    /// The interval to execute the handler, in seconds.
159    #[serde_as(as = "serde_with::DurationSeconds<u64>")]
160    #[configurable(metadata(docs::human_name = "Interval"))]
161    interval_seconds: Duration,
162
163    /// The handler function which is called when the timer ticks.
164    ///
165    /// It can produce new events using the `emit` function.
166    ///
167    /// This can either be inline Lua that defines a closure to use, or the name of the Lua function
168    /// to call. In both cases, the closure/function takes a single parameter, `emit`, which is a
169    /// reference to a function for emitting events.
170    #[configurable(metadata(docs::examples = "timer_handler"))]
171    handler: String,
172}
173
174impl LuaConfig {
175    pub fn build(&self, key: ComponentKey) -> crate::Result<Transform> {
176        Lua::new(self, key).map(Transform::event_task)
177    }
178
179    pub fn input(&self) -> Input {
180        Input::new(DataType::Metric | DataType::Log)
181    }
182
183    pub fn outputs(
184        &self,
185        input_definitions: &[(OutputId, schema::Definition)],
186    ) -> Vec<TransformOutput> {
187        // Lua causes the type definition to be reset
188        let namespaces = input_definitions
189            .iter()
190            .flat_map(|(_output, definition)| definition.log_namespaces().clone())
191            .collect();
192
193        let definition = input_definitions
194            .iter()
195            .map(|(output, _definition)| {
196                (
197                    output.clone(),
198                    Definition::default_for_namespace(&namespaces),
199                )
200            })
201            .collect();
202
203        vec![TransformOutput::new(
204            DataType::Metric | DataType::Log,
205            definition,
206        )]
207    }
208}
209
210// Lua's garbage collector sometimes seems to be not executed automatically on high event rates,
211// which leads to leak-like RAM consumption pattern. This constant sets the number of invocations of
212// the Lua transform after which GC would be called, thus ensuring that the RAM usage is not too high.
213//
214// This constant is larger than 1 because calling GC is an expensive operation, so doing it
215// after each transform would have significant footprint on the performance.
216const GC_INTERVAL: usize = 16;
217
218pub struct Lua {
219    lua: mlua::Lua,
220    invocations_after_gc: usize,
221    hook_init: Option<mlua::RegistryKey>,
222    hook_process: mlua::RegistryKey,
223    hook_shutdown: Option<mlua::RegistryKey>,
224    timers: Vec<(Timer, mlua::RegistryKey)>,
225    multi_value_tags: bool,
226    source_id: Arc<ComponentKey>,
227}
228
229// Helper to create `RegistryKey` from Lua function code
230fn make_registry_value(lua: &mlua::Lua, source: &str) -> mlua::Result<mlua::RegistryKey> {
231    lua.load(source)
232        .eval::<mlua::Function>()
233        .and_then(|f| lua.create_registry_value(f))
234}
235
236impl Lua {
237    pub fn new(config: &LuaConfig, key: ComponentKey) -> crate::Result<Self> {
238        // In order to support loading C modules in Lua, we need to create unsafe instance
239        // without debug library.
240        let lua = unsafe {
241            mlua::Lua::unsafe_new_with(mlua::StdLib::ALL_SAFE, mlua::LuaOptions::default())
242        };
243
244        let additional_paths = config
245            .search_dirs
246            .iter()
247            .map(|d| format!("{}/?.lua", d.to_string_lossy()))
248            .collect::<Vec<_>>()
249            .join(";");
250
251        let mut timers = Vec::new();
252
253        if !additional_paths.is_empty() {
254            let package = lua.globals().get::<mlua::Table>("package")?;
255            let current_paths = package
256                .get::<String>("path")
257                .unwrap_or_else(|_| ";".to_string());
258            let paths = format!("{additional_paths};{current_paths}");
259            package.set("path", paths)?;
260        }
261
262        if let Some(source) = &config.source {
263            lua.load(source).eval::<()>().context(InvalidSourceSnafu)?;
264        }
265
266        let hook_init_code = config.hooks.init.as_ref();
267        let hook_init = hook_init_code
268            .map(|code| make_registry_value(&lua, code))
269            .transpose()
270            .context(InvalidHooksInitSnafu)?;
271
272        let hook_process =
273            make_registry_value(&lua, &config.hooks.process).context(InvalidHooksProcessSnafu)?;
274
275        let hook_shutdown_code = config.hooks.shutdown.as_ref();
276        let hook_shutdown = hook_shutdown_code
277            .map(|code| make_registry_value(&lua, code))
278            .transpose()
279            .context(InvalidHooksShutdownSnafu)?;
280
281        for (id, timer) in config.timers.iter().enumerate() {
282            let handler_key = lua
283                .load(&timer.handler)
284                .eval::<mlua::Function>()
285                .and_then(|f| lua.create_registry_value(f))
286                .context(InvalidTimerHandlerSnafu)?;
287
288            let timer = Timer {
289                id: id as u32,
290                interval: timer.interval_seconds,
291            };
292            timers.push((timer, handler_key));
293        }
294
295        let multi_value_tags = config.metric_tag_values == MetricTagValues::Full;
296
297        Ok(Self {
298            lua,
299            invocations_after_gc: 0,
300            timers,
301            hook_init,
302            hook_process,
303            hook_shutdown,
304            multi_value_tags,
305            source_id: Arc::new(key),
306        })
307    }
308
309    #[cfg(test)]
310    fn process(&mut self, event: Event, output: &mut Vec<Event>) -> Result<(), mlua::Error> {
311        let source_id = event.source_id().cloned();
312        let lua = &self.lua;
313        let result = lua.scope(|scope| {
314            let emit = scope.create_function_mut(|_, mut event: Event| {
315                if let Some(source_id) = &source_id {
316                    event.set_source_id(Arc::clone(source_id));
317                }
318                output.push(event);
319                Ok(())
320            })?;
321
322            lua.registry_value::<mlua::Function>(&self.hook_process)?
323                .call((
324                    LuaEvent {
325                        event,
326                        metric_multi_value_tags: self.multi_value_tags,
327                    },
328                    emit,
329                ))
330        });
331
332        self.attempt_gc();
333        result
334    }
335
336    #[cfg(test)]
337    fn process_single(&mut self, event: Event) -> Result<Option<Event>, mlua::Error> {
338        let mut out = Vec::new();
339        self.process(event, &mut out)?;
340        assert!(out.len() <= 1);
341        Ok(out.into_iter().next())
342    }
343
344    fn attempt_gc(&mut self) {
345        self.invocations_after_gc += 1;
346        if self.invocations_after_gc % GC_INTERVAL == 0 {
347            emit!(LuaGcTriggered {
348                used_memory: self.lua.used_memory()
349            });
350            _ = self
351                .lua
352                .gc_collect()
353                .context(RuntimeErrorGcSnafu)
354                .map_err(|error| error!(%error, rate_limit = 30));
355            self.invocations_after_gc = 0;
356        }
357    }
358}
359
360// A helper that reduces code duplication.
361fn wrap_emit_fn<'scope, 'env, F: 'scope + FnMut(Event)>(
362    scope: &'scope mlua::Scope<'scope, 'env>,
363    mut emit_fn: F,
364    source_id: Arc<ComponentKey>,
365) -> mlua::Result<mlua::Function> {
366    scope.create_function_mut(move |_, mut event: Event| -> mlua::Result<()> {
367        event.set_source_id(Arc::clone(&source_id));
368        emit_fn(event);
369        Ok(())
370    })
371}
372
373impl RuntimeTransform for Lua {
374    fn hook_process<F>(&mut self, event: Event, emit_fn: F)
375    where
376        F: FnMut(Event),
377    {
378        let lua = &self.lua;
379        let source_id = Arc::clone(event.source_id().unwrap_or(&self.source_id));
380        _ = lua
381            .scope(|scope| -> mlua::Result<()> {
382                lua.registry_value::<mlua::Function>(&self.hook_process)?
383                    .call((
384                        LuaEvent {
385                            event,
386                            metric_multi_value_tags: self.multi_value_tags,
387                        },
388                        wrap_emit_fn(scope, emit_fn, source_id)?,
389                    ))
390            })
391            .context(RuntimeErrorHooksProcessSnafu)
392            .map_err(|e| emit!(LuaBuildError { error: e }));
393
394        self.attempt_gc();
395    }
396
397    fn hook_init<F>(&mut self, emit_fn: F)
398    where
399        F: FnMut(Event),
400    {
401        let lua = &self.lua;
402        _ = lua
403            .scope(|scope| -> mlua::Result<()> {
404                match &self.hook_init {
405                    Some(key) => lua
406                        .registry_value::<mlua::Function>(key)?
407                        .call(wrap_emit_fn(scope, emit_fn, Arc::clone(&self.source_id))?),
408                    None => Ok(()),
409                }
410            })
411            .context(RuntimeErrorHooksInitSnafu)
412            .map_err(|error| error!(%error, rate_limit = 30));
413
414        self.attempt_gc();
415    }
416
417    fn hook_shutdown<F>(&mut self, emit_fn: F)
418    where
419        F: FnMut(Event),
420    {
421        let lua = &self.lua;
422        _ = lua
423            .scope(|scope| -> mlua::Result<()> {
424                match &self.hook_shutdown {
425                    Some(key) => lua
426                        .registry_value::<mlua::Function>(key)?
427                        .call(wrap_emit_fn(scope, emit_fn, Arc::clone(&self.source_id))?),
428                    None => Ok(()),
429                }
430            })
431            .context(RuntimeErrorHooksShutdownSnafu)
432            .map_err(|error| error!(%error, rate_limit = 30));
433
434        self.attempt_gc();
435    }
436
437    fn timer_handler<F>(&mut self, timer: Timer, emit_fn: F)
438    where
439        F: FnMut(Event),
440    {
441        let lua = &self.lua;
442        _ = lua
443            .scope(|scope| -> mlua::Result<()> {
444                let handler_key = &self.timers[timer.id as usize].1;
445                lua.registry_value::<mlua::Function>(handler_key)?
446                    .call(wrap_emit_fn(scope, emit_fn, Arc::clone(&self.source_id))?)
447            })
448            .context(RuntimeErrorTimerHandlerSnafu)
449            .map_err(|error| error!(%error, rate_limit = 30));
450
451        self.attempt_gc();
452    }
453
454    fn timers(&self) -> Vec<Timer> {
455        self.timers.iter().map(|(timer, _)| *timer).collect()
456    }
457}
458
459#[cfg(test)]
460mod tests {
461    use std::{future::Future, sync::Arc};
462
463    use similar_asserts::assert_eq;
464    use tokio::sync::mpsc::{self, Receiver, Sender};
465    use tokio::sync::Mutex;
466    use tokio_stream::wrappers::ReceiverStream;
467
468    use super::*;
469    use crate::test_util::{components::assert_transform_compliance, random_string};
470    use crate::transforms::test::create_topology;
471    use crate::{
472        event::{
473            metric::{Metric, MetricKind, MetricValue},
474            Event, LogEvent, Value,
475        },
476        test_util,
477    };
478
479    fn format_error(error: &mlua::Error) -> String {
480        match error {
481            mlua::Error::CallbackError { traceback, cause } => {
482                format_error(cause) + "\n" + traceback
483            }
484            err => err.to_string(),
485        }
486    }
487
488    fn from_config(config: &str) -> crate::Result<Box<Lua>> {
489        Lua::new(&toml::from_str(config).unwrap(), "transform".into()).map(Box::new)
490    }
491
492    async fn run_transform<T: Future>(
493        config: &str,
494        func: impl FnOnce(Sender<Event>, Arc<Mutex<Receiver<Event>>>) -> T,
495    ) -> T::Output {
496        test_util::trace_init();
497        assert_transform_compliance(async move {
498            let config = super::super::LuaConfig::V2(toml::from_str(config).unwrap());
499            let (tx, rx) = mpsc::channel(1);
500            let (topology, out) = create_topology(ReceiverStream::new(rx), config).await;
501
502            let out = Arc::new(tokio::sync::Mutex::new(out));
503
504            let result = func(tx, Arc::clone(&out)).await;
505
506            topology.stop().await;
507            assert_eq!(out.lock().await.recv().await, None);
508
509            result
510        })
511        .await
512    }
513
514    async fn next_event(out: &Arc<Mutex<Receiver<Event>>>, source: &str) -> Event {
515        let event = out
516            .lock()
517            .await
518            .recv()
519            .await
520            .expect("Event was not received");
521        assert_eq!(
522            event.source_id(),
523            Some(&Arc::new(ComponentKey::from(source)))
524        );
525        event
526    }
527
528    #[tokio::test]
529    async fn lua_runs_init_hook() {
530        let line1 = random_string(9);
531        run_transform(
532            &format!(
533                r#"
534            version = "2"
535            hooks.init = """function (emit)
536                event = {{log={{message="{line1}"}}}}
537                emit(event)
538            end
539            """
540            hooks.process = """function (event, emit)
541                emit(event)
542            end
543            """
544            "#
545            ),
546            |tx, out| async move {
547                let line2 = random_string(9);
548                tx.send(Event::Log(LogEvent::from(line2.as_str())))
549                    .await
550                    .unwrap();
551                drop(tx);
552                assert_eq!(
553                    next_event(&out, "transform").await.as_log()["message"],
554                    line1.into()
555                );
556                assert_eq!(
557                    next_event(&out, "in").await.as_log()["message"],
558                    line2.into(),
559                );
560            },
561        )
562        .await;
563    }
564
565    #[tokio::test]
566    async fn lua_add_field() {
567        run_transform(
568            r#"
569            version = "2"
570            hooks.process = """function (event, emit)
571                event["log"]["hello"] = "goodbye"
572                emit(event)
573            end
574            """
575            "#,
576            |tx, out| async move {
577                let event = Event::Log(LogEvent::from("program me"));
578                tx.send(event).await.unwrap();
579
580                assert_eq!(
581                    next_event(&out, "in").await.as_log()["hello"],
582                    "goodbye".into()
583                );
584            },
585        )
586        .await;
587    }
588
589    #[tokio::test]
590    async fn lua_read_field() {
591        run_transform(
592            r#"
593            version = "2"
594            hooks.process = """function (event, emit)
595                _, _, name = string.find(event.log.message, "Hello, my name is (%a+).")
596                event.log.name = name
597                emit(event)
598            end
599            """
600            "#,
601            |tx, out| async move {
602                let event = Event::Log(LogEvent::from("Hello, my name is Bob."));
603                tx.send(event).await.unwrap();
604
605                assert_eq!(next_event(&out, "in").await.as_log()["name"], "Bob".into());
606            },
607        )
608        .await;
609    }
610
611    #[tokio::test]
612    async fn lua_remove_field() {
613        run_transform(
614            r#"
615            version = "2"
616            hooks.process = """function (event, emit)
617                event.log.name = nil
618                emit(event)
619            end
620            """
621            "#,
622            |tx, out| async move {
623                let mut event = LogEvent::default();
624                event.insert("name", "Bob");
625
626                tx.send(event.into()).await.unwrap();
627
628                assert_eq!(next_event(&out, "in").await.as_log().get("name"), None);
629            },
630        )
631        .await;
632    }
633
634    #[tokio::test]
635    async fn lua_drop_event() {
636        run_transform(
637            r#"
638            version = "2"
639            hooks.process = """function (event, emit)
640                -- emit nothing
641            end
642            """
643            "#,
644            |tx, _out| async move {
645                let event = LogEvent::default().into();
646                tx.send(event).await.unwrap();
647
648                // "run_transform" will assert that the output stream is empty
649            },
650        )
651        .await;
652    }
653
654    #[tokio::test]
655    async fn lua_duplicate_event() {
656        run_transform(
657            r#"
658            version = "2"
659            hooks.process = """function (event, emit)
660                emit(event)
661                emit(event)
662            end
663            """
664            "#,
665            |tx, out| async move {
666                let mut event = LogEvent::default();
667                event.insert("host", "127.0.0.1");
668                tx.send(event.into()).await.unwrap();
669
670                assert!(out.lock().await.recv().await.is_some());
671                assert!(out.lock().await.recv().await.is_some());
672            },
673        )
674        .await;
675    }
676
677    #[tokio::test]
678    async fn lua_read_empty_field() {
679        run_transform(
680            r#"
681            version = "2"
682            hooks.process = """function (event, emit)
683                if event["log"]["non-existant"] == nil then
684                  event["log"]["result"] = "empty"
685                else
686                  event["log"]["result"] = "found"
687                end
688                emit(event)
689            end
690            """
691            "#,
692            |tx, out| async move {
693                let event = LogEvent::default();
694                tx.send(event.into()).await.unwrap();
695
696                assert_eq!(
697                    next_event(&out, "in").await.as_log()["result"],
698                    "empty".into()
699                );
700            },
701        )
702        .await;
703    }
704
705    #[tokio::test]
706    async fn lua_integer_value() {
707        run_transform(
708            r#"
709            version = "2"
710            hooks.process = """function (event, emit)
711                event["log"]["number"] = 3
712                emit(event)
713            end
714            """
715            "#,
716            |tx, out| async move {
717                let event = LogEvent::default();
718                tx.send(event.into()).await.unwrap();
719
720                assert_eq!(
721                    next_event(&out, "in").await.as_log()["number"],
722                    Value::Integer(3)
723                );
724            },
725        )
726        .await;
727    }
728
729    #[tokio::test]
730    async fn lua_numeric_value() {
731        run_transform(
732            r#"
733            version = "2"
734            hooks.process = """function (event, emit)
735                event["log"]["number"] = 3.14159
736                emit(event)
737            end
738            """
739            "#,
740            |tx, out| async move {
741                let event = LogEvent::default();
742                tx.send(event.into()).await.unwrap();
743
744                assert_eq!(
745                    next_event(&out, "in").await.as_log()["number"],
746                    Value::from(3.14159)
747                );
748            },
749        )
750        .await;
751    }
752
753    #[tokio::test]
754    async fn lua_boolean_value() {
755        run_transform(
756            r#"
757            version = "2"
758            hooks.process = """function (event, emit)
759                event["log"]["bool"] = true
760                emit(event)
761            end
762            """
763            "#,
764            |tx, out| async move {
765                let event = LogEvent::default();
766                tx.send(event.into()).await.unwrap();
767
768                assert_eq!(
769                    next_event(&out, "in").await.as_log()["bool"],
770                    Value::Boolean(true)
771                );
772            },
773        )
774        .await;
775    }
776
777    #[tokio::test]
778    async fn lua_non_coercible_value() {
779        run_transform(
780            r#"
781            version = "2"
782            hooks.process = """function (event, emit)
783                event["log"]["junk"] = nil
784                emit(event)
785            end
786            """
787            "#,
788            |tx, out| async move {
789                let event = LogEvent::default();
790                tx.send(event.into()).await.unwrap();
791
792                assert_eq!(next_event(&out, "in").await.as_log().get("junk"), None);
793            },
794        )
795        .await;
796    }
797
798    #[tokio::test]
799    async fn lua_non_string_key_write() -> crate::Result<()> {
800        let mut transform = from_config(
801            r#"
802            hooks.process = """function (event, emit)
803                event["log"][false] = "hello"
804                emit(event)
805            end
806            """
807            "#,
808        )
809        .unwrap();
810
811        let err = transform
812            .process_single(LogEvent::default().into())
813            .unwrap_err();
814        let err = format_error(&err);
815        assert!(
816            err.contains("error converting Lua boolean to String"),
817            "{}",
818            err
819        );
820        Ok(())
821    }
822
823    #[tokio::test]
824    async fn lua_non_string_key_read() {
825        run_transform(
826            r#"
827            version = "2"
828            hooks.process = """function (event, emit)
829                event.log.result = event.log[false]
830                emit(event)
831            end
832            """
833            "#,
834            |tx, out| async move {
835                let event = LogEvent::default();
836                tx.send(event.into()).await.unwrap();
837
838                assert_eq!(next_event(&out, "in").await.as_log().get("result"), None);
839            },
840        )
841        .await;
842    }
843
844    #[tokio::test]
845    async fn lua_script_error() -> crate::Result<()> {
846        let mut transform = from_config(
847            r#"
848            hooks.process = """function (event, emit)
849                error("this is an error")
850            end
851            """
852            "#,
853        )
854        .unwrap();
855
856        let err = transform
857            .process_single(LogEvent::default().into())
858            .unwrap_err();
859        let err = format_error(&err);
860        assert!(err.contains("this is an error"), "{}", err);
861        Ok(())
862    }
863
864    #[tokio::test]
865    async fn lua_syntax_error() -> crate::Result<()> {
866        let err = from_config(
867            r#"
868            hooks.process = """function (event, emit)
869                1234 = sadf <>&*!#@
870            end
871            """
872            "#,
873        )
874        .map(|_| ())
875        .unwrap_err()
876        .to_string();
877
878        assert!(err.contains("syntax error:"), "{}", err);
879        Ok(())
880    }
881
882    #[tokio::test]
883    async fn lua_load_file() {
884        use std::{fs::File, io::Write};
885
886        let dir = tempfile::tempdir().unwrap();
887        let mut file = File::create(dir.path().join("script2.lua")).unwrap();
888        write!(
889            &mut file,
890            r#"
891            local M = {{}}
892
893            local function modify(event2)
894              event2["log"]["new field"] = "new value"
895            end
896            M.modify = modify
897
898            return M
899            "#
900        )
901        .unwrap();
902
903        run_transform(
904            &format!(
905                r#"
906            version = "2"
907            hooks.process = """function (event, emit)
908                local script2 = require("script2")
909                script2.modify(event)
910                emit(event)
911            end
912            """
913            search_dirs = [{:?}]
914            "#,
915                dir.path().as_os_str() // This seems a bit weird, but recall we also support windows.
916            ),
917            |tx, out| async move {
918                let event = LogEvent::default();
919                tx.send(event.into()).await.unwrap();
920
921                assert_eq!(
922                    next_event(&out, "in").await.as_log()["\"new field\""],
923                    "new value".into()
924                );
925            },
926        )
927        .await;
928    }
929
930    #[tokio::test]
931    async fn lua_pairs() {
932        run_transform(
933            r#"
934            version = "2"
935            hooks.process = """function (event, emit)
936                for k,v in pairs(event.log) do
937                  event.log[k] = k .. v
938                end
939                emit(event)
940            end
941            """
942            "#,
943            |tx, out| async move {
944                let mut event = LogEvent::default();
945                event.insert("name", "Bob");
946                event.insert("friend", "Alice");
947                tx.send(event.into()).await.unwrap();
948
949                let output = next_event(&out, "in").await;
950
951                assert_eq!(output.as_log()["name"], "nameBob".into());
952                assert_eq!(output.as_log()["friend"], "friendAlice".into());
953            },
954        )
955        .await;
956    }
957
958    #[tokio::test]
959    async fn lua_metric() {
960        run_transform(
961            r#"
962            version = "2"
963                hooks.process = """function (event, emit)
964                event.metric.counter.value = event.metric.counter.value + 1
965                emit(event)
966            end
967            """
968            "#,
969            |tx, out| async move {
970                let metric = Metric::new(
971                    "example counter",
972                    MetricKind::Absolute,
973                    MetricValue::Counter { value: 1.0 },
974                );
975
976                let mut expected = metric
977                    .clone()
978                    .with_value(MetricValue::Counter { value: 2.0 });
979                let metadata = expected.metadata_mut();
980                metadata.set_upstream_id(Arc::new(OutputId::from("transform")));
981                metadata.set_source_id(Arc::new(ComponentKey::from("in")));
982
983                tx.send(metric.into()).await.unwrap();
984
985                assert_eq!(next_event(&out, "in").await.as_metric(), &expected);
986            },
987        )
988        .await;
989    }
990
991    #[tokio::test]
992    async fn lua_multiple_events() {
993        run_transform(
994            r#"
995            version = "2"
996            hooks.process = """function (event, emit)
997                event["log"]["hello"] = "goodbye"
998                emit(event)
999            end
1000            """
1001            "#,
1002            |tx, out| async move {
1003                let n: usize = 10;
1004                let events = (0..n).map(|i| Event::Log(LogEvent::from(format!("program me {i}"))));
1005                for event in events {
1006                    tx.send(event).await.unwrap();
1007                    assert!(out.lock().await.recv().await.is_some());
1008                }
1009            },
1010        )
1011        .await;
1012    }
1013}