Add helper methods to support kafka output in beatreceivers (#49768) (#49872)

* Add helper methods to support kafka output in beatreceivers

(cherry picked from commit 91d75e4)

Co-authored-by: Khushi Jain <khushi.jain@elastic.co>

Author: mergify[bot]

libbeat/common/fmtstr/formatevents.go

@@ -315,7 +315,7 @@ func (e *eventFieldCompiler) compileEventField(
 		defaultValue = op.param
 	}
 
-	path, err := parseEventPath(field)
+	path, err := ParseEventPath(field)
 	if err != nil {
 		return nil, err
 	}
@@ -382,7 +382,7 @@ func (e *eventTimestampEvaler) Eval(c interface{}, out *bytes.Buffer) error {
 	return err
 }
 
-func parseEventPath(field string) (string, error) {
+func ParseEventPath(field string) (string, error) {
 	field = strings.Trim(field, " \n\r\t")
 	fields := []string{}
 

libbeat/common/fmtstr/formatstring.go

@@ -132,7 +132,7 @@ func Compile(in string, vc VariableCompiler) (StringFormatter, error) {
 }
 
 func compile(ctx *compileCtx, in string) (StringFormatter, error) {
-	lexer := makeLexer(in)
+	lexer := MakeLexer(in)
 	defer lexer.Finish()
 
 	// parse format string
@@ -265,33 +265,44 @@ func (e variableElement) compile(ctx *compileCtx) (FormatEvaler, error) {
 	return ctx.compileVariable(e.field, e.ops)
 }
 
-func parse(lex lexer) ([]formatElement, error) {
-	var elems []formatElement
+// parseFormatTokens runs the shared lexer-driven loop
+func parseFormatTokens[T any](
+	lex lexer,
+	appendLiteral func(*[]T, string),
+	parseVar func(lexer) (T, error),
+) ([]T, error) {
+	var elems []T
 
 	for token := range lex.Tokens() {
 		switch token.typ {
 		case tokErr:
 			return nil, errors.New(token.val)
 
 		case tokString:
-			elems = append(elems, StringElement{token.val})
+			appendLiteral(&elems, token.val)
 
 		case tokOpen:
-			elem, err := parseVariable(lex)
+			elem, err := parseVar(lex)
 			if err != nil {
 				return nil, err
 			}
 			elems = append(elems, elem)
 
 		case tokClose, tokOperator:
 			// should not happen, but let's return error just in case
-			return nil, fmt.Errorf("Token '%v'(%v) not allowed", token.val, token.typ)
+			return nil, fmt.Errorf("token '%v'(%v) not allowed", token.val, token.typ)
 		}
 	}
 
 	return elems, nil
 }
 
+func parse(lex lexer) ([]formatElement, error) {
+	return parseFormatTokens(lex, func(elems *[]formatElement, s string) {
+		*elems = append(*elems, StringElement{s})
+	}, parseVariable)
+}
+
 func parseVariable(lex lexer) (formatElement, error) {
 	var strings []string
 	var ops []string
@@ -312,7 +323,7 @@ func parseVariable(lex lexer) (formatElement, error) {
 
 		case tokString:
 			if len(strings) != len(ops) {
-				return nil, fmt.Errorf("Unexpected string token %v, expected operator", token.val)
+				return nil, fmt.Errorf("unexpected string token %v, expected operator", token.val)
 			}
 			strings = append(strings, token.val)
 
@@ -322,18 +333,82 @@ func parseVariable(lex lexer) (formatElement, error) {
 			}
 			ops = append(ops, token.val)
 			if len(ops) > len(strings) {
-				return nil, fmt.Errorf("Consecutive operator tokens '%v'", token.val)
+				return nil, fmt.Errorf("consecutive operator tokens '%v'", token.val)
 			}
 
 		default:
-			return nil, fmt.Errorf("Unexpected token '%v' (%v)", token.val, token.typ)
+			return nil, fmt.Errorf("unexpected token '%v' (%v)", token.val, token.typ)
 		}
 	}
 
 	return nil, errMissingClose
 }
 
-func makeLexer(in string) lexer {
+type VariableToken string
+
+// ParseRawTokens returns a slice of tokens as they occur.
+// variable tokens are stored as typed value
+func ParseRawTokens(lex lexer) ([]any, error) {
+	return parseFormatTokens(lex, func(elems *[]any, s string) {
+		*elems = append(*elems, s)
+	}, func(lex lexer) (any, error) {
+		s, err := parseVariableToken(lex)
+		if err != nil {
+			return nil, err
+		}
+		return VariableToken(s), nil
+	})
+}
+
+// parseVariableToken consumes lexer tokens inside %{...} and returns the full inner
+// expression as one string  e.g. "unknown:default".
+// Callers must parse operators if needed
+func parseVariableToken(lex lexer) (string, error) {
+	// finalValue appends each string chunk and operator as they appear.
+	var finalValue string
+	var strings []string
+	var ops []string
+
+	for token := range lex.Tokens() {
+		switch token.typ {
+		case tokErr:
+			return "", errors.New(token.val)
+
+		case tokOpen:
+			return "", errNestedVar
+
+		case tokClose:
+			if len(strings) == 0 {
+				return "", errEmptyFormat
+			}
+			return finalValue, nil
+
+		case tokString:
+			if len(strings) != len(ops) {
+				return "", fmt.Errorf("unexpected string token %v, expected operator", token.val)
+			}
+			strings = append(strings, token.val)
+			finalValue += token.val
+
+		case tokOperator:
+			if len(strings) == 0 {
+				return "", errUnexpectedOperator
+			}
+			ops = append(ops, token.val)
+			if len(ops) > len(strings) {
+				return "", fmt.Errorf("consecutive operator tokens '%v'", token.val)
+			}
+			finalValue += token.val
+
+		default:
+			return "", fmt.Errorf("unexpected token '%v' (%v)", token.val, token.typ)
+		}
+	}
+
+	return "", errMissingClose
+}
+
+func MakeLexer(in string) lexer {
 	lex := make(chan token, 1)
 
 	go func() {
@@ -359,7 +434,10 @@ func makeLexer(in string) lexer {
 
 		varcount := 0
 		for len(content) > 0 {
-			idx := -1
+			if off > len(content) {
+				return
+			}
+			var idx int
 			if varcount == 0 {
 				idx = strings.IndexAny(content[off:], `%\`)
 			} else {
@@ -387,7 +465,8 @@ func makeLexer(in string) lexer {
 				op := ":"
 				if strings.ContainsRune("!@#&*=+<>?", rune(content[off])) {
 					off++
-					op = content[idx : off+1]
+					// Two-byte op e.g. ":!"
+					op = content[idx:off]
 				}
 				lex <- opToken(op)
 

libbeat/common/fmtstr/formatstring_test.go

@@ -20,9 +20,11 @@ package fmtstr
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestFormatString(t *testing.T) {
@@ -171,3 +173,145 @@ func TestFormatStringErrors(t *testing.T) {
 		assert.Error(t, err)
 	}
 }
+
+func TestParseRawTokens(t *testing.T) {
+
+	testCases := []struct {
+		name         string
+		input        string
+		expectedList []any
+		err          error
+	}{
+
+		{
+			name:         "empty string",
+			input:        "",
+			expectedList: nil,
+		},
+		{
+			name:         `when two %%`,
+			input:        `%%`,
+			expectedList: []any{"%%"},
+		},
+		{
+			name:  `when input is %%{}`,
+			input: `%%{}`,
+			err:   fmt.Errorf("empty format expansion"),
+		},
+		{
+			name:         `when input is %\{}`,
+			input:        `%\{}`,
+			expectedList: []any{"%{}"},
+		},
+		{
+			name:  `when input is %{}`,
+			input: `%{}`,
+			err:   fmt.Errorf("empty format expansion"),
+		},
+		{
+			name:         `when input is %{key}\\`,
+			input:        `%{key}\\`,
+			expectedList: []any{VariableToken("key"), `\`},
+		},
+		{
+			name:         `when input is %{a:b:c}`,
+			input:        `%{a:b:c}`,
+			expectedList: []any{VariableToken("a:b:c")},
+		},
+		{
+			name:  `when input is %{a`,
+			input: `%{a`,
+			err:   fmt.Errorf(`missing closing '}'`),
+		},
+		{
+			name:         "simple lookup start of string",
+			input:        "%{k} test",
+			expectedList: []any{VariableToken("k"), " test"},
+		},
+		{
+			name:         "simple lookup end of string",
+			input:        "test %{k}",
+			expectedList: []any{"test ", VariableToken("k")},
+		},
+		{
+			name:         "simple lookup middle of string",
+			input:        "pre %{k} post",
+			expectedList: []any{"pre ", VariableToken("k"), " post"},
+		},
+		{
+			name:         "compile lookup default",
+			input:        "%{unknown:default}",
+			expectedList: []any{VariableToken("unknown:default")},
+		},
+		{
+			name:         "with escaped % symbol",
+			input:        `\%{abc}`,
+			expectedList: []any{`%{abc}`},
+		},
+	}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			lexer := MakeLexer(test.input)
+			defer lexer.Finish()
+			got, err := ParseRawTokens(lexer)
+			if test.err != nil {
+				require.Equal(t, test.err, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, test.expectedList, got)
+		})
+	}
+}
+
+func FuzzParseRawTokens(f *testing.F) {
+	var cases = []string{
+		"",
+		"%{k} test",
+		"pre %{k} post",
+		"%{unknown:default}",
+		"100% literal",
+		"%%",
+		"%%{}",
+		"%\\{}",
+		"%{}",
+		"%{a:b:c}",
+		"%{a:b:?c}",
+		"%{a",
+	}
+
+	for _, c := range cases {
+		f.Add(c)
+	}
+
+	f.Fuzz(func(t *testing.T, a string) {
+		lex := MakeLexer(a)
+		defer lex.Finish()
+		output, err := ParseRawTokens(lex)
+		if err != nil {
+			t.Logf("skipping input %s with error: %v", a, err)
+			return // invalid input
+		}
+
+		// stringify output and match it with original input
+		var finalOutput string
+		for _, out := range output {
+			switch tok := out.(type) {
+			case string:
+				finalOutput += tok
+			case VariableToken:
+				finalOutput += "%{" + string(tok) + "}"
+			default:
+				assert.Fail(t, fmt.Sprintf("unexpected type %T", tok))
+			}
+		}
+
+		// We cannot accurately reconstruct with escaped characters, so we remove them before comparing.
+		assert.Equalf(t, removeEscapes(a), removeEscapes(finalOutput), "unexpected output: %s != %s", a, finalOutput)
+	})
+}
+
+func removeEscapes(str string) string {
+	return strings.ReplaceAll(str, `\`, "")
+}