Add modify_fields processor and use it to set logName

confgenerator/confgenerator.go

@@ -209,7 +209,7 @@ func (l *Logging) generateFluentbitComponents(userAgent string, hostInfo *host.I
 					}
 					components = append(components, processor.Components(tag, strconv.Itoa(i))...)
 				}
-				components = append(components, setLogNameComponents(tag, rID)...)
+				components = append(components, setLogNameComponents(tag, rID, receiver.Type())...)
 
 				// Logs ingested using the fluent_forward receiver must add the existing_tag
 				// on the record to the LogName. This is done with a Lua filter.

confgenerator/config.go

@@ -110,9 +110,14 @@ func (ve validationError) Error() string {
 		return fmt.Sprintf("%q must start with %q", ve.Field(), ve.Param())
 	case "url":
 		return fmt.Sprintf("%q must be a URL", ve.Field())
+	case "excluded_with":
+		return fmt.Sprintf("%q cannot be set if one of [%s] is set", ve.Field(), ve.Param())
 	case "filter":
 		_, err := filter.NewFilter(ve.Value().(string))
 		return fmt.Sprintf("%q: %v", ve.Field(), err)
+	case "field":
+		_, err := filter.NewMember(ve.Value().(string))
+		return fmt.Sprintf("%q: %v", ve.Field(), err)
 	}
 
 	return ve.FieldError.Error()
@@ -172,6 +177,12 @@ func newValidator() *validator.Validate {
 		_, err := filter.NewFilter(fl.Field().String())
 		return err == nil
 	})
+	// field validates that a Cloud Logging field expression is valid
+	v.RegisterValidation("field", func(fl validator.FieldLevel) bool {
+		_, err := filter.NewMember(fl.Field().String())
+		// TODO: Disallow specific target fields?
+		return err == nil
+	})
 	// multipleof_time validates that the value duration is a multiple of the parameter
 	v.RegisterValidation("multipleof_time", func(fl validator.FieldLevel) bool {
 		t, ok := fl.Field().Interface().(time.Duration)

confgenerator/filter/filter.go

@@ -27,6 +27,26 @@ import (
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/fluentbit"
 )
 
+type Member struct {
+	ast.Target
+}
+
+func NewMember(m string) (*Member, error) {
+	lex := lexer.NewLexer([]byte(m))
+	p := parser.NewParser()
+	out, err := p.Parse(lex)
+	if err != nil {
+		return nil, err
+	}
+	r, ok := out.(ast.Restriction)
+	if !ok || r.Operator != "GLOBAL" {
+		return nil, fmt.Errorf("not a field: %#v", out)
+	}
+	return &Member{r.LHS}, nil
+}
+
+var LuaQuote = ast.LuaQuote
+
 type Filter struct {
 	expr ast.Expression
 }
@@ -44,31 +64,45 @@ func NewFilter(f string) (*Filter, error) {
 	return nil, fmt.Errorf("not an expression: %+v", out)
 }
 
-// innerComponents returns only the logical modify filters that are intended to be
-// positioned between corresponding nest/grep/lift filters.
-func (f *Filter) innerComponents(tag string) []fluentbit.Component {
-	match := fmt.Sprintf("__match_%s", strings.ReplaceAll(tag, ".", "_"))
-	return f.expr.Components(tag, match)
+// innerFluentConfig returns components that are intended to be positioned between corresponding nest/lift filters and a Lua expression to evaluate.
+func (f *Filter) innerFluentConfig(tag, prefix string) ([]fluentbit.Component, string) {
+	return f.expr.FluentConfig(tag, prefix)
 }
 
-func (f *Filter) Components(tag string, isExclusionFilter bool) []fluentbit.Component {
-	return AllComponents(tag, []*Filter{f}, isExclusionFilter)
+func (f Filter) String() string {
+	return f.expr.String()
 }
 
-// AllComponents returns a list of FluentBit components for a list of filters.
-// As an optimization, only a single set of nest/grep/lift components is
-// emitted in total.
-func AllComponents(tag string, filters []*Filter, isExclusionFilter bool) []fluentbit.Component {
-	var parity string
-	if isExclusionFilter {
-		parity = "Exclude"
-	} else {
-		parity = "Regex"
+// MatchesAny returns a single Filter that matches if any of the child filters match.
+func MatchesAny(filters []*Filter) *Filter {
+	d := ast.Disjunction{}
+	for _, f := range filters {
+		d = append(d, f.expr)
 	}
-	// TODO: Re-implement using Lua once regex is supported. Lua has been shown to perform better
-	// than the next/modify/grep/lift pattern used here, but we are unable to use Lua for now since
-	// it does not yet support regex.
-	c := []fluentbit.Component{{
+	return &Filter{expr: d}
+}
+
+// AllFluentConfig returns components (if any) and Lua code that sets a Boolean local variable for each filter to indicate if that filter matched.
+func AllFluentConfig(tag string, filters map[string]*Filter) ([]fluentbit.Component, string) {
+	var c []fluentbit.Component
+	var lua strings.Builder
+	var vars []string
+	for k := range filters {
+		vars = append(vars, k)
+	}
+
+	for i, k := range vars {
+		prefix := fmt.Sprintf("__match_%d", i)
+		filter := filters[k]
+		filterComponents, filterExpr := filter.innerFluentConfig(tag, prefix)
+		c = append(c, filterComponents...)
+		lua.WriteString(fmt.Sprintf("local %s = %s\n", k, filterExpr))
+	}
+	if len(c) == 0 {
+		// If we didn't need any filters, just return the Lua code.
+		return nil, lua.String()
+	}
+	out := []fluentbit.Component{{
 		Kind: "FILTER",
 		Config: map[string]string{
 			"Name":       "nest",
@@ -78,36 +112,23 @@ func AllComponents(tag string, filters []*Filter, isExclusionFilter bool) []flue
 			"Wildcard":   "*",
 		},
 	}}
-	match := fmt.Sprintf("__match_%s", strings.ReplaceAll(tag, ".", "_"))
-	for _, filter := range filters {
-		c = append(c, filter.innerComponents(tag)...)
-	}
-	c = append(c,
-		fluentbit.Component{
-			Kind: "FILTER",
-			Config: map[string]string{
-				"Name":  "grep",
-				"Match": tag,
-				parity:  fmt.Sprintf("%s 1", match),
-			},
-		},
-		fluentbit.Component{
-			Kind: "FILTER",
-			Config: map[string]string{
-				"Name":            "modify",
-				"Match":           tag,
-				"Remove_wildcard": match,
-			},
-		},
-		fluentbit.Component{
-			Kind: "FILTER",
-			Config: map[string]string{
-				"Name":         "nest",
-				"Match":        tag,
-				"Operation":    "lift",
-				"Nested_under": "record",
-			},
+	out = append(out, c...)
+	out = append(out, fluentbit.Component{
+		Kind: "FILTER",
+		Config: map[string]string{
+			"Name":         "nest",
+			"Match":        tag,
+			"Operation":    "lift",
+			"Nested_under": "record",
 		},
-	)
-	return c
+	})
+	// Remove match keys
+	lua.WriteString(`
+for k, v in pairs(record) do
+  if string.match(k, "^__match_.+") then
+    record[k] = nil
+  end
+end
+`)
+	return out, lua.String()
 }

confgenerator/filter/filter_test.go

@@ -20,20 +20,28 @@ import (
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/filter/internal/generated/lexer"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/filter/internal/generated/token"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/fluentbit"
+	"github.com/google/go-cmp/cmp"
 )
 
 var validFilters = []string{
 	`severity = "hello"`,
 	`jsonPayload."bar.baz" = "hello"`,
 	`jsonPayload.b.c=~"b.*c"`,
+	`"jsonPayload"."foo" = "bar"`,
 	`-severity = 1`,
 	`NOT severity = 3`,
 	`(jsonPayload.bar = "one" OR jsonPayload.bar = "two") jsonPayload.baz = "three"`,
 	`jsonPayload.one = 1 jsonPayload.two = 2 AND jsonPayload.three = 3`,
 	`jsonPayload.int_field:0 OR jsonPayload.int_field:0 AND jsonPayload.int_field:0`,
 	`jsonPayload.compound.string_field : wal\"rus`,
-	`severity =~ ERROR AND jsonPayload.message =~ foo AND httpRequest.requestMethod =~ GET`,
+	`severity =~ "ERROR" AND jsonPayload.message =~ "foo" AND httpRequest.requestMethod =~ "GET"`,
 	`severity = "AND"`,
+	`severity = AND`,
+	`severity = OR`,
+	`severity = NOT`,
+	`"json\u0050ayload".foo = bar`,
+	`jsonPayload.\= = bar`,
+	`jsonPayload."\=" = bar`,
 }
 
 func TestShouldLex(t *testing.T) {
@@ -59,13 +67,88 @@ func TestShouldParse(t *testing.T) {
 			if err != nil {
 				t.Error(err)
 			}
-			components := filter.Components("logname", false)
+			t.Logf("parsed filter = %s", filter)
+			if filter == nil {
+				t.Fatal("got nil filter")
+			}
+			components, expr := AllFluentConfig("logname", map[string]*Filter{"filter": filter})
 			t.Logf("components = %+v", components)
-			files, err := fluentbit.ModularConfig{Components: components}.Generate()
+			t.Logf("expression =\n%s", expr)
+			if components != nil {
+				files, err := fluentbit.ModularConfig{Components: components}.Generate()
+				if err != nil {
+					t.Error(err)
+				}
+				t.Logf("generated config:\n%v", files)
+			}
+		})
+	}
+}
+
+func TestFilterRoundTrip(t *testing.T) {
+	for _, test := range validFilters {
+		test := test
+		t.Run(test, func(t *testing.T) {
+			filter, err := NewFilter(test)
 			if err != nil {
-				t.Error(err)
+				t.Fatal(err)
+			}
+			first := filter.String()
+			filter2, err := NewFilter(first)
+			if err != nil {
+				t.Fatalf("failed to re-parse %q: %v", first, err)
+			}
+			second := filter2.String()
+			if diff := cmp.Diff(second, first); diff != "" {
+				t.Errorf("filter did not round-trip (second -, first+):\n%s", diff)
+			}
+		})
+	}
+}
+
+func TestInvalidFilters(t *testing.T) {
+	for _, test := range []string{
+		`"missing operator"`,
+		`invalid/characters*here`,
+		`jsonPayload.foo =~ bareword`,
+		`json\u0050ayload.foo = bar`,
+	} {
+		test := test
+		t.Run(test, func(t *testing.T) {
+			filter, err := NewFilter(test)
+			if err != nil {
+				t.Logf("got expected error %v", err)
+				return
+			}
+			t.Errorf("invalid filter %q unexpectedly parsed: %v", test, filter)
+		})
+	}
+}
+
+func TestValidMembers(t *testing.T) {
+	for _, test := range []struct {
+		in   string
+		want []string
+	}{
+		{`jsonPayload.foo`, []string{"jsonPayload", "foo"}},
+		{`labels."logging.googleapis.com/foo"`, []string{"labels", "logging.googleapis.com/foo"}},
+		{`severity`, []string{"severity"}},
+		{`jsonPayload.\=`, []string{"jsonPayload", `\=`}},
+		{`jsonPayload."\="`, []string{"jsonPayload", `=`}},
+	} {
+		test := test
+		t.Run(test.in, func(t *testing.T) {
+			member, err := NewMember(test.in)
+			if err != nil {
+				t.Fatal(err)
+			}
+			got, err := member.Unquote()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if diff := cmp.Diff(got, test.want); diff != "" {
+				t.Errorf("incorrect parse (got -/want +):\n%s", diff)
 			}
-			t.Logf("generated config:\n%v", files)
 		})
 	}
 }