feat: Update Mongodb receiver to use LoggingReceiverMacro

apps/mongodb.go

@@ -16,12 +16,9 @@ package apps
 
 import (
 	"context"
-	"fmt"
 	"strings"
 
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
-	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/fluentbit"
-	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/fluentbit/modify"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/otel"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/secret"
 )
@@ -95,120 +92,81 @@ func init() {
 	confgenerator.MetricsReceiverTypes.RegisterType(func() confgenerator.MetricsReceiver { return &MetricsReceiverMongoDB{} })
 }
 
-type LoggingProcessorMongodb struct {
-	confgenerator.ConfigComponent `yaml:",inline"`
+type LoggingProcessorMacroMongodb struct {
 }
 
-func (*LoggingProcessorMongodb) Type() string {
+func (LoggingProcessorMacroMongodb) Type() string {
 	return "mongodb"
 }
 
-func (p *LoggingProcessorMongodb) Components(ctx context.Context, tag, uid string) []fluentbit.Component {
-	c := []fluentbit.Component{}
+func (p LoggingProcessorMacroMongodb) Expand(ctx context.Context) []confgenerator.InternalLoggingProcessor {
+	c := []confgenerator.InternalLoggingProcessor{}
 
-	c = append(c, p.JsonLogComponents(ctx, tag, uid)...)
-	c = append(c, p.RegexLogComponents(tag, uid)...)
-	c = append(c, p.severityParser(ctx, tag, uid)...)
+	c = append(c, p.JsonLogComponents(ctx)...)
+	c = append(c, p.RegexLogComponents()...)
+	c = append(c, p.severityParser()...)
 
 	return c
 }
 
 // JsonLogComponents are the fluentbit components for parsing log messages that are json formatted.
 // these are generally messages from mongo with versions greater than or equal to 4.4
 // documentation: https://docs.mongodb.com/v4.4/reference/log-messages/#log-message-format
-func (p *LoggingProcessorMongodb) JsonLogComponents(ctx context.Context, tag, uid string) []fluentbit.Component {
-	c := p.jsonParserWithTimeKey(ctx, tag, uid)
+func (p LoggingProcessorMacroMongodb) JsonLogComponents(ctx context.Context) []confgenerator.InternalLoggingProcessor {
+	c := p.jsonParserWithTimeKey()
 
-	c = append(c, p.promoteWiredTiger(tag, uid)...)
-	c = append(c, p.renames(tag, uid)...)
+	c = append(c, p.promoteWiredTiger()...)
+	c = append(c, p.renames()...)
 
 	return c
 }
 
 // jsonParserWithTimeKey requires promotion of the nested timekey for the json parser so we must
 // first promote the $date field from the "t" field before declaring the parser
-func (p *LoggingProcessorMongodb) jsonParserWithTimeKey(ctx context.Context, tag, uid string) []fluentbit.Component {
-	c := []fluentbit.Component{}
+func (p LoggingProcessorMacroMongodb) jsonParserWithTimeKey() []confgenerator.InternalLoggingProcessor {
+	c := []confgenerator.InternalLoggingProcessor{}
 
-	jsonParser := &confgenerator.LoggingProcessorParseJson{
+	c = append(c, &confgenerator.LoggingProcessorParseJson{
 		ParserShared: confgenerator.ParserShared{
-			TimeKey:    "time",
-			TimeFormat: "%Y-%m-%dT%H:%M:%S.%L%z",
 			Types: map[string]string{
 				"id":      "integer",
 				"message": "string",
 			},
 		},
-	}
-	jpComponents := jsonParser.Components(ctx, tag, uid)
-
-	// The parserFilterComponent is the actual filter component that configures and defines
-	// which parser to use. We need the component to determine which parser to use when
-	// re-parsing below. Each time a parser filter is used, there are 2 filter components right
-	// before it to account for the nest lua script (see confgenerator/fluentbit/parse_deduplication.go).
-	// Therefore, the parse filter component is actually the third component in the list.
-	parserFilterComponent := jpComponents[2]
-	c = append(c, jpComponents...)
-
-	tempPrefix := "temp_ts_"
-	timeKey := "time"
+	})
+
 	// have to bring $date to top level in order for it to be parsed as timeKey
 	// see https://github.com/fluent/fluent-bit/issues/1013
-	liftTs := fluentbit.Component{
-		Kind: "FILTER",
-		Config: map[string]string{
-			"Name":         "nest",
-			"Match":        tag,
-			"Operation":    "lift",
-			"Nested_under": "t",
-			"Add_prefix":   tempPrefix,
+	c = append(c, &confgenerator.LoggingProcessorModifyFields{
+		Fields: map[string]*confgenerator.ModifyField{
+			"jsonPayload.time": {
+				MoveFrom: "jsonPayload.t.$date",
+			},
 		},
-	}
-
-	renameTsOption := modify.NewHardRenameOptions(fmt.Sprintf("%s$date", tempPrefix), timeKey)
-	renameTs := renameTsOption.Component(tag)
+	})
 
-	c = append(c, liftTs, renameTs)
+	c = append(c, &confgenerator.LoggingProcessorRemoveField{
+		Field: "t",
+	})
 
 	// IMPORTANT: now that we have lifted the json to top level
 	// we need to re-parse in order to properly set time at the
 	// parser level
-	nestFilters := fluentbit.LuaFilterComponents(tag, fluentbit.ParserNestLuaFunction, fmt.Sprintf(fluentbit.ParserNestLuaScriptContents, "message"))
-	parserFilter := fluentbit.Component{
-		Kind: "FILTER",
-		Config: map[string]string{
-			"Name":         "parser",
-			"Match":        tag,
-			"Key_Name":     "message",
-			"Reserve_Data": "True",
-			"Parser":       parserFilterComponent.OrderedConfig[0][1],
-		},
-	}
-	mergeFilters := fluentbit.LuaFilterComponents(tag, fluentbit.ParserMergeLuaFunction, fluentbit.ParserMergeLuaScriptContents)
-	c = append(c, nestFilters...)
-	c = append(c, parserFilter)
-	c = append(c, mergeFilters...)
-
-	removeTimestamp := fluentbit.Component{
-		Kind: "FILTER",
-		Config: map[string]string{
-			"Name":   "modify",
-			"Match":  tag,
-			"Remove": timeKey,
+	c = append(c, &confgenerator.LoggingProcessorParseTimestamp{
+		ParserShared: confgenerator.ParserShared{
+			TimeKey:    "time",
+			TimeFormat: "%Y-%m-%dT%H:%M:%S.%L%z",
 		},
-	}
-	c = append(c, removeTimestamp)
+	})
 
 	return c
 }
 
 // severityParser is used by both regex and json parser to ensure an "s" field on the entry gets translated
 // to a valid logging.googleapis.com/seveirty field
-func (p *LoggingProcessorMongodb) severityParser(ctx context.Context, tag, uid string) []fluentbit.Component {
-	severityComponents := []fluentbit.Component{}
-
-	severityComponents = append(severityComponents,
-		confgenerator.LoggingProcessorModifyFields{
+func (p LoggingProcessorMacroMongodb) severityParser() []confgenerator.InternalLoggingProcessor {
+	return []confgenerator.InternalLoggingProcessor{
+		&confgenerator.LoggingProcessorModifyFields{
 			Fields: map[string]*confgenerator.ModifyField{
 				"jsonPayload.severity": {
 					MoveFrom: "jsonPayload.s",
@@ -231,115 +189,85 @@ func (p *LoggingProcessorMongodb) severityParser(ctx context.Context, tag, uid s
 				},
 				InstrumentationSourceLabel: instrumentationSourceValue(p.Type()),
 			},
-		}.Components(ctx, tag, uid)...,
-	)
-
-	return severityComponents
+		},
+	}
 }
 
-func (p *LoggingProcessorMongodb) renames(tag, uid string) []fluentbit.Component {
-	r := []fluentbit.Component{}
+func (p LoggingProcessorMacroMongodb) renames() []confgenerator.InternalLoggingProcessor {
+	r := []confgenerator.InternalLoggingProcessor{}
 	renames := []struct {
 		src  string
 		dest string
 	}{
-		{"c", "component"},
-		{"ctx", "context"},
-		{"msg", "message"},
+		{"jsonPayload.c", "jsonPayload.component"},
+		{"jsonPayload.ctx", "jsonPayload.context"},
+		{"jsonPayload.msg", "jsonPayload.message"},
+		{"jsonPayload.attr", "jsonPayload.attributes"},
 	}
 
 	for _, rename := range renames {
-		rename := modify.NewRenameOptions(rename.src, rename.dest)
-		r = append(r, rename.Component(tag))
+		r = append(r, &confgenerator.LoggingProcessorModifyFields{
+			Fields: map[string]*confgenerator.ModifyField{
+				rename.dest: {
+					MoveFrom: rename.src,
+				},
+			},
+		})
 	}
 
 	return r
 }
 
-func (p *LoggingProcessorMongodb) promoteWiredTiger(tag, uid string) []fluentbit.Component {
+func (p LoggingProcessorMacroMongodb) promoteWiredTiger() []confgenerator.InternalLoggingProcessor {
 	// promote messages that are WiredTiger messages and are nested in attr.message
+	c := []confgenerator.InternalLoggingProcessor{}
+
 	addPrefix := "temp_attributes_"
-	upNest := fluentbit.Component{
-		Kind: "FILTER",
-		Config: map[string]string{
-			"Name":         "nest",
-			"Match":        tag,
-			"Operation":    "lift",
-			"Nested_under": "attr",
-			"Add_prefix":   addPrefix,
-		},
-	}
 
-	hardRenameMessage := modify.NewHardRenameOptions(fmt.Sprintf("%smessage", addPrefix), "msg")
-	wiredTigerRename := hardRenameMessage.Component(tag)
-
-	renameRemainingAttributes := fluentbit.Component{
-		Kind: "FILTER",
-		Config: map[string]string{
-			"Name":          "nest",
-			"Wildcard":      fmt.Sprintf("%s*", addPrefix),
-			"Match":         tag,
-			"Operation":     "nest",
-			"Nest_under":    "attributes",
-			"Remove_prefix": addPrefix,
+	c = append(c, &confgenerator.LoggingProcessorModifyFields{
+		Fields: map[string]*confgenerator.ModifyField{
+			"jsonPayload.temp_attributes_message": {
+				MoveFrom: "jsonPayload.attr.message",
+			},
 		},
-	}
+	})
+
+	c = append(c, &confgenerator.LoggingProcessorRenameIfExists{
+		Field:   addPrefix + "message",
+		NewName: "msg",
+	})
 
-	return []fluentbit.Component{upNest, wiredTigerRename, renameRemainingAttributes}
+	return c
 }
 
-func (p *LoggingProcessorMongodb) RegexLogComponents(tag, uid string) []fluentbit.Component {
-	c := []fluentbit.Component{}
-	parseKey := "message"
-	parser, parserName := fluentbit.ParserComponentBase("%Y-%m-%dT%H:%M:%S.%L%z", "timestamp", map[string]string{
-		"message":   "string",
-		"id":        "integer",
-		"s":         "string",
-		"component": "string",
-		"context":   "string",
-	}, fmt.Sprintf("%s_regex", tag), uid)
-	parser.Config["Format"] = "regex"
-	parser.Config["Regex"] = `^(?<timestamp>[^ ]*)\s+(?<s>\w)\s+(?<component>[^ ]+)\s+\[(?<context>[^\]]+)]\s+(?<message>.*?) *(?<ms>(\d+))?(:?ms)?$`
-	parser.Config["Key_Name"] = parseKey
-
-	nestFilters := fluentbit.LuaFilterComponents(tag, fluentbit.ParserNestLuaFunction, fmt.Sprintf(fluentbit.ParserNestLuaScriptContents, parseKey))
-	parserFilter := fluentbit.Component{
-		Kind: "FILTER",
-		Config: map[string]string{
-			"Match":        tag,
-			"Name":         "parser",
-			"Parser":       parserName,
-			"Reserve_Data": "True",
-			"Key_Name":     parseKey,
-		},
-	}
-	mergeFilters := fluentbit.LuaFilterComponents(tag, fluentbit.ParserMergeLuaFunction, fluentbit.ParserMergeLuaScriptContents)
+func (p LoggingProcessorMacroMongodb) RegexLogComponents() []confgenerator.InternalLoggingProcessor {
+	c := []confgenerator.InternalLoggingProcessor{}
 
-	c = append(c, parser)
-	c = append(c, nestFilters...)
-	c = append(c, parserFilter)
-	c = append(c, mergeFilters...)
+	c = append(c, &confgenerator.LoggingProcessorParseRegex{
+		ParserShared: confgenerator.ParserShared{
+			TimeKey:    "timestamp",
+			TimeFormat: "%Y-%m-%dT%H:%M:%S.%L%z",
+			Types: map[string]string{
+				"message":   "string",
+				"id":        "integer",
+				"s":         "string",
+				"component": "string",
+				"context":   "string",
+			},
+		},
+		Regex: `^(?<timestamp>[^ ]*)\s+(?<s>\w)\s+(?<component>[^ ]+)\s+\[(?<context>[^\]]+)]\s+(?<message>.*?) *(?<ms>(\d+))?(:?ms)?$`,
+		Field: "message",
+	})
 
 	return c
 }
 
-type LoggingReceiverMongodb struct {
-	LoggingProcessorMongodb `yaml:",inline"`
-	ReceiverMixin           confgenerator.LoggingReceiverFilesMixin `yaml:",inline" validate:"structonly"`
-}
-
-func (r *LoggingReceiverMongodb) Components(ctx context.Context, tag string) []fluentbit.Component {
-	if len(r.ReceiverMixin.IncludePaths) == 0 {
-		r.ReceiverMixin.IncludePaths = []string{
-			// default logging location
-			"/var/log/mongodb/mongod.log*",
-		}
+func loggingReceiverFilesMixinMongodb() confgenerator.LoggingReceiverFilesMixin {
+	return confgenerator.LoggingReceiverFilesMixin{
+		IncludePaths: []string{"/var/log/mongodb/mongod.log*"},
 	}
-	c := r.ReceiverMixin.Components(ctx, tag)
-	c = append(c, r.LoggingProcessorMongodb.Components(ctx, tag, "mongodb")...)
-	return c
 }
 
 func init() {
-	confgenerator.LoggingReceiverTypes.RegisterType(func() confgenerator.LoggingReceiver { return &LoggingReceiverMongodb{} })
+	confgenerator.RegisterLoggingFilesProcessorMacro[LoggingProcessorMacroMongodb](loggingReceiverFilesMixinMongodb)
 }