Function Parse

pkg/parser/runtime.go:233–343

(ctx UnixParserCtx, event pipeline.Event, nodes []Node, collector *StageParseCollector)

}

func Parse(ctx UnixParserCtx, event pipeline.Event, nodes []Node, collector *StageParseCollector) (pipeline.Event, error) {
	/* the stage is undefined, probably line is freshly acquired, set to first stage !*/
	if event.Stage == "" && len(ctx.Stages) > 0 {
		event.Stage = ctx.Stages[0]
		log.Tracef("no stage, set to : %s", event.Stage)
	}

	event.Process = false
	if event.Time.IsZero() {
		event.Time = time.Now().UTC()
	}

	if event.Parsed == nil {
		event.Parsed = make(map[string]string)
	}

	if event.Enriched == nil {
		event.Enriched = make(map[string]string)
	}

	if event.Meta == nil {
		event.Meta = make(map[string]string)
	}

	if event.Unmarshaled == nil {
		event.Unmarshaled = make(map[string]any)
	}

	if event.Type == pipeline.LOG {
		log.Tracef("INPUT '%s'", event.Line.Raw)
	}

	exprEnv := map[string]any{"evt": &event}

	for _, stage := range ctx.Stages {
		/* if the node is forward in stages, seek to this stage */
		/* this is for example used by testing system to inject logs in post-syslog-parsing phase*/
		if slices.Index(ctx.Stages, event.Stage) > slices.Index(ctx.Stages, stage) {
			log.Tracef("skipping stage, we are already at [%s] expecting [%s]", event.Stage, stage)
			continue
		}

		log.Tracef("node stage : %s, current stage : %s", event.Stage, stage)

		/* if the stage is wrong, it means that the log didn't manage "pass" a stage with a onsuccess: next_stage tag */
		if event.Stage != stage {
			log.Debugf("Event not parsed, expected stage '%s' got '%s', abort", stage, event.Stage)
			event.Process = false

			return event, nil
		}

		isStageOK := false

		for idx := range nodes {
			// Only process current stage's nodes
			if event.Stage != nodes[idx].Stage {
				continue