feat(confgenerator): migrate legacy receivers to OTel and remove Fluent Bit components

Author: rafaelwestphal

apps/common_logging_processors.go

@@ -31,7 +31,7 @@ func instrumentationSourceValue(processorType string) *confgenerator.ModifyField
 }
 
 // genericAccessLogParser is an internal logging processor that parses access logs.
-func genericAccessLogParser(ctx context.Context, processorType string) []confgenerator.InternalLoggingProcessor {
+func genericAccessLogParser(ctx context.Context, processorType string) []confgenerator.InternalOTelProcessor {
 	c := confgenerator.LoggingProcessorParseRegex{
 		// Documentation:
 		// https://httpd.apache.org/docs/current/logs.html#accesslog
@@ -85,7 +85,7 @@ func genericAccessLogParser(ctx context.Context, processorType string) []confgen
 		}
 	}
 
-	return []confgenerator.InternalLoggingProcessor{
+	return []confgenerator.InternalOTelProcessor{
 		c,
 		mf,
 	}

apps/iis.go

@@ -19,7 +19,6 @@ import (
 	"fmt"
 
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
-	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/fluentbit"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/otel"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/otel/ottl"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/platform"
@@ -162,39 +161,6 @@ func (LoggingProcessorMacroIisAccess) Type() string {
 // IISConcatFields handles field concatenation for IIS logs
 type IISConcatFields struct{}
 
-// Components implements the Fluent-bit concatenation using Lua
-func (IISConcatFields) Components(ctx context.Context, tag, uid string) []fluentbit.Component {
-	iisMergeRecordFieldsLuaFunction := `iis_merge_fields`
-	iisMergeRecordFieldsLuaScriptContents := `
-	function iis_merge_fields(tag, timestamp, record)
-
-	  if (record["cs_uri_query"] == "-") then
-	    record["cs_uri_query"] = nil
-	  end
-	  if (record["http_request_referer"] == "-") then
-	    record["http_request_referer"] = nil
-	  end
-	  if (record["user"] == "-") then
-	    record["user"] = nil
-	  end
-
-	  record["http_request_serverIp"] = table.concat({record["http_request_serverIp"], ":", record["s_port"]})
-	  if (record["cs_uri_query"] == nil or record["cs_uri_query"] == '') then
-		record["http_request_requestUrl"] = record["cs_uri_stem"]
-	  else
-		record["http_request_requestUrl"] = table.concat({record["cs_uri_stem"], "?", record["cs_uri_query"]})
-	  end
-
-	  record["cs_uri_query"] = nil
-	  record["cs_uri_stem"] = nil
-	  record["s_port"] = nil
-	  return 2, timestamp, record
-	end
-	`
-
-	return fluentbit.LuaFilterComponents(tag, iisMergeRecordFieldsLuaFunction, iisMergeRecordFieldsLuaScriptContents)
-}
-
 // Processors implements the OTEL concatenation using ModifyFields + CustomConvertFunc
 func (IISConcatFields) Processors(ctx context.Context) ([]otel.Component, error) {
 	// Required OTTL fields
@@ -259,7 +225,7 @@ func (IISConcatFields) Processors(ctx context.Context) ([]otel.Component, error)
 	return modifyFields.Processors(ctx)
 }
 
-func (p LoggingProcessorMacroIisAccess) Expand(ctx context.Context) []confgenerator.InternalLoggingProcessor {
+func (p LoggingProcessorMacroIisAccess) Expand(ctx context.Context) []confgenerator.InternalOTelProcessor {
 	parseRegex := confgenerator.LoggingProcessorParseRegex{
 		// Microsoft updated the default format in Feb 2026.
 		// The new format now has fields sc_bytes and cs_bytes added right before time_taken
@@ -316,7 +282,7 @@ func (p LoggingProcessorMacroIisAccess) Expand(ctx context.Context) []confgenera
 		Fields: fields,
 	}
 
-	return []confgenerator.InternalLoggingProcessor{
+	return []confgenerator.InternalOTelProcessor{
 		parseRegex,
 		concatFields,
 		excludeLogs,

confgenerator/confgenerator.go

@@ -17,18 +17,12 @@ package confgenerator
 
 import (
 	"context"
-	"crypto/md5"
-	"encoding/hex"
 	"fmt"
 	"log"
 	"maps"
 	"path"
-	"regexp"
-	"sort"
-	"strconv"
 	"strings"
 
-	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/fluentbit"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/otel"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/resourcedetector"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/experiments"
@@ -281,114 +275,6 @@ func (uc *UnifiedConfig) GenerateOtelConfig(ctx context.Context, outDir, stateDi
 	return otelConfig, nil
 }
 
-func (p PipelineInstance) simplifiedLoggingComponents(ctx context.Context) (InternalLoggingReceiver, []InternalLoggingProcessor, error) {
-	receiver, ok := p.Receiver.(InternalLoggingReceiver)
-	if !ok {
-		return nil, nil, fmt.Errorf("%q is not a logging receiver", p.RID)
-	}
-	// Expand receiver and processors
-	// TODO: What if they can be recursively expanded?
-	var processors []InternalLoggingProcessor
-	if r, ok := p.Receiver.(LoggingReceiverMacro); ok {
-		receiver, processors = r.Expand(ctx)
-	}
-	for _, processorItem := range p.Processors {
-		processor, ok := processorItem.Component.(InternalLoggingProcessor)
-		if !ok {
-			return nil, nil, fmt.Errorf("logging processor %q is incompatible with a receiver of type %q", processorItem.ID, p.Receiver.Type())
-		}
-		if p, ok := processor.(LoggingProcessorMacro); ok {
-			processors = append(processors, p.Expand(ctx)...)
-			continue
-		}
-		processors = append(processors, processor)
-	}
-	// Now that receiver and processors are all expanded, try merging them.
-	for len(processors) > 0 {
-		// Check if current receiver can merge processors.
-		// This needs to happen every iteration because the receiver might be different after a previous merge.
-		mr, ok := receiver.(InternalLoggingProcessorMerger)
-		if !ok {
-			return receiver, processors, nil
-		}
-
-		// Attempt processor merge.
-		receiver, processors[0] = mr.MergeInternalLoggingProcessor(processors[0])
-		if processors[0] != nil {
-			break
-		}
-		processors = processors[1:]
-	}
-	// Now receiver has been merged as much as possible.
-	return receiver, processors, nil
-}
-
-func (p PipelineInstance) FluentBitComponents(ctx context.Context) (fbSource, error) {
-	tag := fmt.Sprintf("%s.%s", p.PID, p.RID)
-
-	// For fluent_forward we create the tag in the following format:
-	// <hash_string>.<pipeline_id>.<receiver_id>.<existing_tag>
-	//
-	// hash_string: Deterministic unique identifier for the pipeline_id + receiver_id.
-	//   This is needed to prevent collisions between receivers in the same
-	//   pipeline when using the glob syntax for matching (using wildcards).
-	// pipeline_id: User defined pipeline_id but with the "." replaced with "_"
-	//   since the "." character is reserved to be used as a delimiter in the
-	//   Lua script.
-	// receiver_id: User defined receiver_id but with the "." replaced with "_"
-	//   since the "." character is reserved to be used as a delimiter in the
-	//   Lua script.
-	//  existing_tag: Tag associated with the record prior to ingesting.
-	//
-	// For an example testing collisions in receiver_ids, see:
-	//
-	// testdata/valid/linux/logging-receiver_forward_multiple_receivers_conflicting_id
-	if p.Receiver.Type() == "fluent_forward" {
-		hashString := getMD5Hash(tag)
-
-		// Note that we only update the tag for the tag. The LogName will still
-		// use the user defined receiver_id without this replacement.
-		pipelineIdCleaned := strings.ReplaceAll(p.PID, ".", "_")
-		receiverIdCleaned := strings.ReplaceAll(p.RID, ".", "_")
-		tag = fmt.Sprintf("%s.%s.%s", hashString, pipelineIdCleaned, receiverIdCleaned)
-	}
-	receiver, processors, err := p.simplifiedLoggingComponents(ctx)
-	if err != nil {
-		return fbSource{}, err
-	}
-	var components []fluentbit.Component
-	receiverComponents := receiver.Components(ctx, tag)
-	components = append(components, receiverComponents...)
-
-	// To match on fluent_forward records, we need to account for the addition
-	// of the existing tag (unknown during config generation) as the suffix
-	// of the tag.
-	globSuffix := ""
-	regexSuffix := ""
-	if p.Receiver.Type() == "fluent_forward" {
-		regexSuffix = `\..*`
-		globSuffix = `.*`
-	}
-	tagRegex := regexp.QuoteMeta(tag) + regexSuffix
-	tag = tag + globSuffix
-
-	for i, processor := range processors {
-		processorComponents := processor.Components(ctx, tag, strconv.Itoa(i))
-		components = append(components, processorComponents...)
-	}
-	components = append(components, setLogNameComponents(ctx, tag, p.RID, p.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.
-	if p.Receiver.Type() == "fluent_forward" {
-		components = append(components, fluentbit.LuaFilterComponents(tag, addLogNameLuaFunction, addLogNameLuaScriptContents)...)
-	}
-	return fbSource{
-		TagRegex:   tagRegex,
-		Components: components,
-	}, nil
-}
-
 func (p PipelineInstance) OTelComponents(ctx context.Context) (map[string]otel.ReceiverPipeline, map[string]otel.Pipeline, error) {
 	outR := make(map[string]otel.ReceiverPipeline)
 	outP := make(map[string]otel.Pipeline)
@@ -479,25 +365,6 @@ func (uc *UnifiedConfig) generateOtelPipelines(ctx context.Context) (map[string]
 	return outR, outP, nil
 }
 
-// GenerateFluentBitConfigs generates configuration file(s) for Fluent Bit.
-// It returns a map of filenames to file contents.
-func (uc *UnifiedConfig) GenerateFluentBitConfigs(ctx context.Context, logsDir string, stateDir string) (map[string]string, error) {
-	userAgent, _ := platform.FromContext(ctx).UserAgent("Google-Cloud-Ops-Agent-Logging")
-	components, err := uc.generateFluentbitComponents(ctx, userAgent)
-	if err != nil {
-		return nil, err
-	}
-
-	c := fluentbit.ModularConfig{
-		Variables: map[string]string{
-			"buffers_dir": path.Join(stateDir, "buffers"),
-			"logs_dir":    logsDir,
-		},
-		Components: components,
-	}
-	return c.Generate()
-}
-
 func contains(s []string, str string) bool {
 	for _, v := range s {
 		if v == str {
@@ -554,61 +421,3 @@ func addGceMetadataAttributesProcessor(ctx context.Context) LoggingProcessorModi
 	}
 	return p
 }
-
-type fbSource struct {
-	TagRegex   string
-	Components []fluentbit.Component
-}
-
-// generateFluentbitComponents generates a slice of fluentbit config sections to represent l.
-func (uc *UnifiedConfig) generateFluentbitComponents(ctx context.Context, userAgent string) ([]fluentbit.Component, error) {
-	l := uc.Logging
-	var out []fluentbit.Component
-	if l.Service.LogLevel == "" {
-		l.Service.LogLevel = "info"
-	}
-	service := fluentbit.Service{LogLevel: l.Service.LogLevel}
-	out = append(out, service.Component())
-	out = append(out, fluentbit.MetricsInputComponent())
-
-	if l != nil && l.Service != nil && (l.Service.OTelLogging == nil || !*l.Service.OTelLogging) {
-		// Type for sorting.
-		var sources []fbSource
-		var tags []string
-		pipelines, err := uc.Pipelines(ctx)
-		if err != nil {
-			return nil, err
-		}
-		for _, pipeline := range pipelines {
-			if pipeline.Backend != BackendFluentBit {
-				continue
-			}
-			source, err := pipeline.FluentBitComponents(ctx)
-			if err != nil {
-				return nil, err
-			}
-			sources = append(sources, source)
-			tags = append(tags, source.TagRegex)
-		}
-		sort.Slice(sources, func(i, j int) bool { return sources[i].TagRegex < sources[j].TagRegex })
-		sort.Strings(tags)
-
-		for _, s := range sources {
-			out = append(out, s.Components...)
-		}
-		if len(tags) > 0 {
-			out = append(out, stackdriverOutputComponent(ctx, strings.Join(tags, "|"), userAgent, "2G", l.Service.Compress))
-		}
-		out = append(out, addGceMetadataAttributesProcessor(ctx).Components(ctx, "*", "*.default-data-proc.gce_metadata")...)
-	}
-	out = append(out, uc.generateSelfLogsComponents(ctx, userAgent)...)
-	out = append(out, fluentbit.MetricsOutputComponent(int(uc.GetFluentBitMetricsPort())))
-
-	return out, nil
-}
-
-func getMD5Hash(text string) string {
-	hasher := md5.New()
-	hasher.Write([]byte(text))
-	return hex.EncodeToString(hasher.Sum(nil))
-}

confgenerator/confgenerator_test.go

@@ -300,32 +300,10 @@ func generateConfigs(pc platformConfig, testDir string) (got map[string]string,
 	}
 	got[builtinConfigFileName] = confgenerator.BuiltInConfStructs[pc.platform.Name()].String()
 
-	pipelines, err := mergedUc.Pipelines(ctx)
+	_, err = mergedUc.Pipelines(ctx)
 	if err != nil {
 		return
 	}
-	hasFluentBit := false
-	for _, p := range pipelines {
-		if p.Backend == confgenerator.BackendFluentBit {
-			hasFluentBit = true
-			break
-		}
-	}
-
-	if hasFluentBit {
-		// Fluent Bit configs
-		var flbGeneratedConfigs map[string]string
-		flbGeneratedConfigs, err = mergedUc.GenerateFluentBitConfigs(ctx,
-			pc.defaultLogsDir,
-			pc.defaultStateDir,
-		)
-		for k, v := range flbGeneratedConfigs {
-			got[k] = v
-		}
-		if err != nil {
-			return
-		}
-	}
 
 	// Otel configs
 	otelGeneratedConfig, err := mergedUc.GenerateOtelConfig(ctx, "", "")