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#[configurable_component]
52#[derive(Clone, Debug)]
53#[serde(deny_unknown_fields)]
54pub struct LuaConfig {
55 #[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 #[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 #[serde(default)]
79 timers: Vec<TimerConfig>,
80
81 #[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#[configurable_component]
112#[derive(Clone, Debug)]
113#[serde(deny_unknown_fields)]
114struct HooksConfig {
115 #[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 #[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 #[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#[serde_as]
155#[configurable_component]
156#[derive(Clone, Debug)]
157struct TimerConfig {
158 #[serde_as(as = "serde_with::DurationSeconds<u64>")]
160 #[configurable(metadata(docs::human_name = "Interval"))]
161 interval_seconds: Duration,
162
163 #[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 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
210const 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
229fn 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 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
360fn 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 },
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() ),
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}