Enable OTel Logging for unmodified default pipelines

apps/builtinconf.go

@@ -16,8 +16,8 @@ package apps
 
 import cg "github.com/GoogleCloudPlatform/ops-agent/confgenerator"
 
-var (
-	BuiltInConfStructs = map[string]*cg.UnifiedConfig{
+func init() {
+	cg.BuiltInConfStructs = map[string]*cg.UnifiedConfig{
 		"linux": {
 			Logging: &cg.Logging{
 				Receivers: map[string]cg.LoggingReceiver{
@@ -105,4 +105,4 @@ var (
 			},
 		},
 	}
-)
+}

cmd/agent_wrapper/main.go

@@ -21,7 +21,7 @@ import (
 	"os"
 	"os/exec"
 
-	"github.com/GoogleCloudPlatform/ops-agent/apps"
+	_ "github.com/GoogleCloudPlatform/ops-agent/apps"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
 	"gopkg.in/natefinch/lumberjack.v2"
 )
@@ -34,7 +34,7 @@ func getLogFileRotation(config *confgenerator.UnifiedConfig) confgenerator.LogFi
 }
 
 func run(logFilename, configurationPath string, cmd *exec.Cmd) error {
-	ucConfig, err := confgenerator.MergeConfFiles(context.Background(), configurationPath, apps.BuiltInConfStructs)
+	ucConfig, err := confgenerator.MergeConfFiles(context.Background(), configurationPath)
 	if err != nil {
 		return err
 	}

cmd/google_cloud_ops_agent_engine/main.go

@@ -20,7 +20,7 @@ import (
 	"log"
 	"os"
 
-	"github.com/GoogleCloudPlatform/ops-agent/apps"
+	_ "github.com/GoogleCloudPlatform/ops-agent/apps"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/healthchecks"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/logs"
@@ -55,15 +55,15 @@ func main() {
 func run() error {
 	ctx := context.Background()
 	// TODO(lingshi) Move this to a shared place across Linux and Windows.
-	uc, err := confgenerator.MergeConfFiles(ctx, *input, apps.BuiltInConfStructs)
+	uc, err := confgenerator.MergeConfFiles(ctx, *input)
 	if err != nil {
 		return err
 	}
 
 	// Log the built-in and merged config files to STDOUT. These are then written
 	// by journald to var/log/syslog and so to Cloud Logging once the ops-agent is
 	// running.
-	log.Printf("Built-in config:\n%s", apps.BuiltInConfStructs["linux"])
+	log.Printf("Built-in config:\n%s", confgenerator.BuiltInConfStructs["linux"])
 	log.Printf("Merged config:\n%s", uc)
 
 	switch *service {

cmd/ops_agent_uap_plugin/plugin_test.go

@@ -23,7 +23,7 @@ import (
 
 	"buf.build/go/protoyaml" // Import the protoyaml-go package
 	pb "github.com/GoogleCloudPlatform/google-guest-agent/pkg/proto/plugin_comm"
-	"github.com/GoogleCloudPlatform/ops-agent/apps"
+	_ "github.com/GoogleCloudPlatform/ops-agent/apps"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/platform"
 	spb "google.golang.org/protobuf/types/known/structpb"
@@ -188,7 +188,7 @@ func TestWriteCustomConfigToFile(t *testing.T) {
 				t.Errorf("%v: writeCustomConfigToFile got error: %v, want nil error", tc.name, err)
 			}
 
-			_, err = confgenerator.MergeConfFiles(context.Background(), configPath, apps.BuiltInConfStructs)
+			_, err = confgenerator.MergeConfFiles(context.Background(), configPath)
 			if err != nil {
 				t.Errorf("%v: conf generator fails to validate the output Ops agent yaml: %v", tc.name, err)
 			}

cmd/ops_agent_uap_plugin/service_windows.go

@@ -29,7 +29,7 @@ import (
 	"sync"
 	"unsafe"
 
-	"github.com/GoogleCloudPlatform/ops-agent/apps"
+	_ "github.com/GoogleCloudPlatform/ops-agent/apps"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/healthchecks"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/logs"
@@ -207,12 +207,12 @@ func findPreExistentAgents(mgr serviceManager, agentWindowsServiceNames []string
 }
 
 func generateSubAgentConfigs(ctx context.Context, userConfigPath string, pluginStateDir string) error {
-	uc, err := confgenerator.MergeConfFiles(ctx, userConfigPath, apps.BuiltInConfStructs)
+	uc, err := confgenerator.MergeConfFiles(ctx, userConfigPath)
 	if err != nil {
 		return err
 	}
 
-	log.Printf("Built-in config:\n%s\n", apps.BuiltInConfStructs["windows"])
+	log.Printf("Built-in config:\n%s\n", confgenerator.BuiltInConfStructs["windows"])
 	log.Printf("Merged config:\n%s\n", uc)
 
 	// The generated otlp metric json files are used only by the otel service.

cmd/ops_agent_windows/run_windows.go

@@ -23,7 +23,7 @@ import (
 	"os"
 	"path/filepath"
 
-	"github.com/GoogleCloudPlatform/ops-agent/apps"
+	_ "github.com/GoogleCloudPlatform/ops-agent/apps"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/healthchecks"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/logs"
@@ -158,12 +158,12 @@ func (srv *service) runHealthChecks() {
 
 func (s *service) generateConfigs(ctx context.Context) error {
 	// TODO(lingshi) Move this to a shared place across Linux and Windows.
-	uc, err := confgenerator.MergeConfFiles(ctx, s.userConf, apps.BuiltInConfStructs)
+	uc, err := confgenerator.MergeConfFiles(ctx, s.userConf)
 	if err != nil {
 		return err
 	}
 
-	s.log.Info(EngineEventID, fmt.Sprintf("Built-in config:\n%s\n", apps.BuiltInConfStructs["windows"]))
+	s.log.Info(EngineEventID, fmt.Sprintf("Built-in config:\n%s\n", confgenerator.BuiltInConfStructs["windows"]))
 	s.log.Info(EngineEventID, fmt.Sprintf("Merged config:\n%s\n", uc))
 	if err := s.checkForStandaloneAgents(uc); err != nil {
 		return err

cmd/simulacra/simulacra.go

@@ -30,7 +30,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-collector/integration_test/gce-testing-internal/gce"
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-collector/integration_test/gce-testing-internal/logging"
-	"github.com/GoogleCloudPlatform/ops-agent/apps"
+	_ "github.com/GoogleCloudPlatform/ops-agent/apps"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
 	"github.com/GoogleCloudPlatform/ops-agent/integration_test/agents"
 	"github.com/binxio/gcloudconfig"
@@ -166,7 +166,7 @@ func getReceiversFromConfig(ctx context.Context, vm *gce.VM, logger *logging.Dir
 		return []string{}, nil
 	}
 
-	config, err := confgenerator.MergeConfFiles(ctx, configFilePath, apps.BuiltInConfStructs)
+	config, err := confgenerator.MergeConfFiles(ctx, configFilePath)
 	if err != nil {
 		return nil, err
 	}

confgenerator/agentmetrics.go

@@ -33,7 +33,6 @@ type AgentSelfMetrics struct {
 	FluentBitPort       int
 	OtelPort            int
 	OtelRuntimeDir      string
-	OtelLogging         bool
 }
 
 // Following reference : https://github.com/googleapis/googleapis/blob/master/google/rpc/code.proto

confgenerator/confgenerator.go

@@ -231,7 +231,6 @@ func (uc *UnifiedConfig) GenerateOtelConfig(ctx context.Context, outDir, stateDi
 		FluentBitPort:       fluentbit.MetricsPort,
 		OtelPort:            otel.MetricsPort,
 		OtelRuntimeDir:      outDir,
-		OtelLogging:         uc.Logging.Service.OTelLogging,
 	}
 	agentSelfMetrics.AddSelfMetricsPipelines(receiverPipelines, pipelines, ctx)
 	resource, err := p.GetResource()
@@ -412,6 +411,10 @@ func (p PipelineInstance) OTelComponents(ctx context.Context) (map[string]otel.R
 	if err != nil {
 		return nil, nil, fmt.Errorf("receiver %q has invalid configuration: %w", p.RID, err)
 	}
+	gceMetadataAttributesProcessors, err := addGceMetadataAttributesProcessor(ctx).Processors(ctx)
+	if err != nil {
+		panic("Failed to generate static ModifyFields")
+	}
 	for i, receiverPipeline := range receiverPipelines {
 		receiverPipelineName := strings.ReplaceAll(p.RID, "_", "__")
 		if i > 0 {
@@ -435,6 +438,10 @@ func (p PipelineInstance) OTelComponents(ctx context.Context) (map[string]otel.R
 					otelFluentForwardSetLogNameComponents()...,
 				)
 			}
+			receiverPipeline.Processors["logs"] = append(
+				receiverPipeline.Processors["logs"],
+				gceMetadataAttributesProcessors...,
+			)
 		}
 
 		outR[receiverPipelineName] = receiverPipeline
@@ -524,42 +531,38 @@ const (
 	attributeLabelPrefix string = "compute.googleapis.com/attributes/"
 )
 
-// addGceMetadataAttributesComponents annotates logs with labels corresponding
-// to instance attributes from the GCE metadata server.
-func addGceMetadataAttributesComponents(ctx context.Context, attributes []string, tag, uid string) []fluentbit.Component {
-	processorName := fmt.Sprintf("%s.%s.gce_metadata", tag, uid)
+// addGceMetadataAttributesProcessor annotates logs with labels corresponding
+// to specific instance attributes from the GCE metadata server.
+func addGceMetadataAttributesProcessor(ctx context.Context) LoggingProcessorModifyFields {
+	attributes := []string{
+		"dataproc-cluster-name",
+		"dataproc-cluster-uuid",
+		"dataproc-region",
+	}
+
+	modifications := map[string]*ModifyField{}
+	p := LoggingProcessorModifyFields{
+		Fields: modifications,
+	}
 	resource, err := platform.FromContext(ctx).GetResource()
 	if err != nil {
 		log.Printf("can't get resource metadata: %v", err)
-		return nil
+		return p
 	}
 	gceMetadata, ok := resource.(resourcedetector.GCEResource)
 	if !ok {
 		// Not on GCE; no attributes to detect.
 		log.Printf("ignoring the gce_metadata_attributes processor outside of GCE: %T", resource)
-		return nil
-	}
-	modifications := map[string]*ModifyField{}
-	var attributeKeys []string
-	for k, _ := range gceMetadata.Metadata {
-		attributeKeys = append(attributeKeys, k)
+		return p
 	}
-	sort.Strings(attributeKeys)
-	for _, k := range attributeKeys {
-		if !sliceContains(attributes, k) {
-			continue
-		}
-		v := gceMetadata.Metadata[k]
-		modifications[fmt.Sprintf(`labels."%s%s"`, attributeLabelPrefix, k)] = &ModifyField{
-			StaticValue: &v,
+	for _, k := range attributes {
+		if v, ok := gceMetadata.Metadata[k]; ok {
+			modifications[fmt.Sprintf(`labels."%s%s"`, attributeLabelPrefix, k)] = &ModifyField{
+				StaticValue: &v,
+			}
 		}
 	}
-	if len(modifications) == 0 {
-		return nil
-	}
-	return LoggingProcessorModifyFields{
-		Fields: modifications,
-	}.Components(ctx, tag, processorName)
+	return p
 }
 
 type fbSource struct {
@@ -578,7 +581,7 @@ func (uc *UnifiedConfig) generateFluentbitComponents(ctx context.Context, userAg
 	out = append(out, service.Component())
 	out = append(out, fluentbit.MetricsInputComponent())
 
-	if l != nil && l.Service != nil && !l.Service.OTelLogging {
+	if l != nil && l.Service != nil && (l.Service.OTelLogging == nil || !*l.Service.OTelLogging) {
 		// Type for sorting.
 		var sources []fbSource
 		var tags []string
@@ -606,11 +609,7 @@ func (uc *UnifiedConfig) generateFluentbitComponents(ctx context.Context, userAg
 		if len(tags) > 0 {
 			out = append(out, stackdriverOutputComponent(ctx, strings.Join(tags, "|"), userAgent, "2G", l.Service.Compress))
 		}
-		out = append(out, addGceMetadataAttributesComponents(ctx, []string{
-			"dataproc-cluster-name",
-			"dataproc-cluster-uuid",
-			"dataproc-region",
-		}, "*", "default-dataproc")...)
+		out = append(out, addGceMetadataAttributesProcessor(ctx).Components(ctx, "*", "*.default-data-proc.gce_metadata")...)
 	}
 	out = append(out, uc.generateSelfLogsComponents(ctx, userAgent)...)
 	out = append(out, fluentbit.MetricsOutputComponent())

confgenerator/confgenerator_test.go

@@ -25,7 +25,7 @@ import (
 	"testing"
 
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp"
-	"github.com/GoogleCloudPlatform/ops-agent/apps"
+	_ "github.com/GoogleCloudPlatform/ops-agent/apps"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator"
 	"github.com/GoogleCloudPlatform/ops-agent/confgenerator/resourcedetector"
 	"github.com/GoogleCloudPlatform/ops-agent/internal/platform"
@@ -261,12 +261,11 @@ func generateConfigs(pc platformConfig, testDir string) (got map[string]string,
 	mergedUc, err := confgenerator.MergeConfFiles(
 		ctx,
 		filepath.Join("testdata", testDir, inputFileName),
-		apps.BuiltInConfStructs,
 	)
 	if err != nil {
 		return
 	}
-	got[builtinConfigFileName] = apps.BuiltInConfStructs[pc.platform.Name()].String()
+	got[builtinConfigFileName] = confgenerator.BuiltInConfStructs[pc.platform.Name()].String()
 
 	// Fluent Bit configs
 	flbGeneratedConfigs, err := mergedUc.GenerateFluentBitConfigs(ctx,
@@ -351,7 +350,6 @@ func generateOtelConfigWithOtlpExporterEnabled(got map[string]string, experiment
 	mergedUcOtlp, err := confgenerator.MergeConfFiles(
 		ctxOtlp,
 		filepath.Join("testdata", testDir, inputFileName),
-		apps.BuiltInConfStructs,
 	)
 	if err == nil {
 		otelGeneratedConfigOtlp, err := mergedUcOtlp.GenerateOtelConfig(ctxOtlp, "", "")

confgenerator/config.go

@@ -21,6 +21,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"runtime"
+	"slices"
 	"sort"
 	"strings"
 	"time"
@@ -38,6 +39,10 @@ import (
 	"go.uber.org/multierr"
 )
 
+// BuiltInConfStructs contains the default configuration for each platform.
+// It is populated by an initializer in the `apps` package.
+var BuiltInConfStructs map[string]*UnifiedConfig
+
 // Ops Agent config.
 type UnifiedConfig struct {
 	Combined *Combined `yaml:"combined,omitempty"`
@@ -583,7 +588,7 @@ type LoggingService struct {
 	Compress    string               `yaml:"compress,omitempty" validate:"omitempty,oneof=gzip,experimental=log_compression"`
 	LogLevel    string               `yaml:"log_level,omitempty" validate:"omitempty,oneof=error warn info debug trace"`
 	Pipelines   map[string]*Pipeline `validate:"dive,keys,startsnotwith=lib:"`
-	OTelLogging bool                 `yaml:"experimental_otel_logging,omitempty" validate:"omitempty,experimental=otel_logging"`
+	OTelLogging *bool                `yaml:"experimental_otel_logging,omitempty" validate:"omitempty"`
 }
 
 type Pipeline struct {
@@ -1084,16 +1089,25 @@ func (uc *UnifiedConfig) loggingPipelines(ctx context.Context) ([]PipelineInstan
 	if err != nil {
 		return nil, err
 	}
+	platformDefaultConfig := BuiltInConfStructs[platform.FromContext(ctx).Name()].Logging
 	exp_otlp := experimentsFromContext(ctx)["otlp_logging"]
-	exp_otel := l.Service.OTelLogging
+	// N.B. Temporarily gate the "auto" otel logging behind an experiment flag.
+	exp_otel := experimentsFromContext(ctx)["otel_logging"]
+	force_otel := l.Service.OTelLogging
 	var out []PipelineInstance
 	for _, pID := range otel.SortedKeys(l.Service.Pipelines) {
 		p := l.Service.Pipelines[pID]
+		defaultP, ok := platformDefaultConfig.Service.Pipelines[pID]
+		isDefaultPipeline := ok && slices.Equal(p.ReceiverIDs, defaultP.ReceiverIDs) && slices.Equal(p.ProcessorIDs, defaultP.ProcessorIDs)
 		for _, rID := range p.ReceiverIDs {
 			receiver, ok := receivers[rID]
 			if !ok {
 				return nil, fmt.Errorf("logging receiver %q not found", rID)
 			}
+			defaultReceiver, ok := platformDefaultConfig.Receivers[rID]
+			if !ok || !reflect.DeepEqual(receiver, defaultReceiver) {
+				isDefaultPipeline = false
+			}
 			var processors []struct {
 				ID string
 				Component
@@ -1120,7 +1134,9 @@ func (uc *UnifiedConfig) loggingPipelines(ctx context.Context) ([]PipelineInstan
 				Receiver:     receiver,
 				Processors:   processors,
 			}
-			if exp_otel || (receiver.Type() == "otlp" && exp_otlp) {
+			if (force_otel != nil && *force_otel) || // User asked for OTel logging
+				(exp_otel && force_otel == nil && isDefaultPipeline) || // Unmodified default pipeline
+				(receiver.Type() == "otlp" && exp_otlp) { // OTLP receiver
 				instance.Backend = BackendOTel
 			}
 			out = append(out, instance)
@@ -1195,7 +1211,8 @@ func (uc *UnifiedConfig) OTelLoggingSupported(ctx context.Context) bool {
 	if ucLoggingCopy.Logging.Service == nil {
 		ucLoggingCopy.Logging.Service = &LoggingService{}
 	}
-	ucLoggingCopy.Logging.Service.OTelLogging = true
+	t := true
+	ucLoggingCopy.Logging.Service.OTelLogging = &t
 	_, err = ucLoggingCopy.GenerateOtelConfig(ctx, "", "")
 	return err == nil
 }

confgenerator/confmerger.go

@@ -23,8 +23,8 @@ import (
 )
 
 // MergeConfFiles merges the user provided config with the built-in config struct for the platform.
-func MergeConfFiles(ctx context.Context, userConfPath string, builtInConfStructs map[string]*UnifiedConfig) (*UnifiedConfig, error) {
-	builtInStruct := builtInConfStructs[platform.FromContext(ctx).Name()]
+func MergeConfFiles(ctx context.Context, userConfPath string) (*UnifiedConfig, error) {
+	builtInStruct := BuiltInConfStructs[platform.FromContext(ctx).Name()]
 
 	// Start with the built-in config.
 	result, err := builtInStruct.DeepCopy(ctx)