Function transform

pkg/acquisition/acquisition.go:458–518

(
	transformChan chan pipeline.Event,
	output chan pipeline.Event,
	acquisTomb *tomb.Tomb,
	transformRuntime *vm.Program,
	logger *log.Entry,
)

}

func transform(
	transformChan chan pipeline.Event,
	output chan pipeline.Event,
	acquisTomb *tomb.Tomb,
	transformRuntime *vm.Program,
	logger *log.Entry,
) {
	logger.Info("transformer started")

	for {
		select {
		case <-acquisTomb.Dying():
			logger.Debugf("transformer is dying")
			return
		case evt := <-transformChan:
			logger.Tracef("Received event %s", evt.Line.Raw)

			out, err := expr.Run(transformRuntime, map[string]any{"evt": &evt})
			if err != nil {
				logger.Errorf("while running transform expression: %s, sending event as-is", err)
				output <- evt
				continue
			}

			if out == nil {
				logger.Errorf("transform expression returned nil, sending event as-is")
				output <- evt
				continue
			}

			switch v := out.(type) {
			case string:
				logger.Tracef("transform expression returned %s", v)
				output <- copyEvent(evt, v)
			case []any:
				logger.Tracef("transform expression returned %v", v) // We actually want to log the slice content

				for _, line := range v {
					l, ok := line.(string)
					if !ok {
						logger.Errorf("transform expression returned []interface{}, but cannot assert an element to string")
						output <- evt

						continue
					}

					output <- copyEvent(evt, l)
				}
			case []string:
				logger.Tracef("transform expression returned %v", v)

				for _, line := range v {
					output <- copyEvent(evt, line)
				}
			default:
				logger.Errorf("transform expression returned an invalid type %T, sending event as-is", out)
				output <- evt
			}