Remap per-agent log_level onto the base key in the consumer

comp/core/configstreamconsumer/impl/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     srcs = [
         "bootstrap.go",
         "consumer.go",
+        "overrides.go",
     ],
     importpath = "github.com/DataDog/datadog-agent/comp/core/configstreamconsumer/impl",
     visibility = ["//visibility:public"],

comp/core/configstreamconsumer/impl/consumer.go

@@ -109,6 +109,8 @@ type consumer struct {
 
 	// Layers this stream has written, keyed by setting. Touched only from the stream goroutine.
 	streamedLayers map[string]map[pkgconfigmodel.Source]struct{}
+	// Base keys applyOverrides has written, so the override can be retracted. Same goroutine.
+	appliedOverrides map[string]struct{}
 
 	ready     atomic.Bool
 	readyCh   chan struct{}
@@ -569,21 +571,36 @@ func (c *consumer) connectAndStream() error {
 }
 
 func (c *consumer) handleConfigEvent(event *pb.ConfigEvent) error {
+	snapshotApplied := false
 	switch e := event.Event.(type) {
 	case *pb.ConfigEvent_Snapshot:
-		return c.applySnapshot(e.Snapshot)
+		applied, err := c.applySnapshot(e.Snapshot)
+		if err != nil {
+			return err
+		}
+		snapshotApplied = applied
 	case *pb.ConfigEvent_Update:
-		return c.applyUpdate(e.Update)
+		if err := c.applyUpdate(e.Update); err != nil {
+			return err
+		}
 	default:
 		return fmt.Errorf("unknown event type: %T", event.Event)
 	}
+	// After applySnapshot's retraction loop, so a remapped value is not retracted out from under itself.
+	c.applyOverrides()
+	if snapshotApplied {
+		// Signalled last: waitForReady must not release before the remap has folded in the override.
+		c.markReady()
+	}
+	return nil
 }
 
-func (c *consumer) applySnapshot(snapshot *pb.ConfigSnapshot) error {
+// applySnapshot reports whether the snapshot was applied; a stale one is dropped and reports false.
+func (c *consumer) applySnapshot(snapshot *pb.ConfigSnapshot) (bool, error) {
 	if snapshot.SequenceId <= c.lastSeqID.Load() {
 		c.log.Warnf("Ignoring stale snapshot (seq_id: %d <= %d)", snapshot.SequenceId, c.lastSeqID.Load())
 		c.droppedStaleUpdates.Inc()
-		return nil
+		return false, nil
 	}
 
 	c.log.Infof("Applying config snapshot (seq_id: %d, settings: %d)", snapshot.SequenceId, len(snapshot.Settings))
@@ -622,15 +639,18 @@ func (c *consumer) applySnapshot(snapshot *pb.ConfigSnapshot) error {
 	c.lastSeqID.Store(snapshot.SequenceId)
 	c.lastSeqIDMetric.Set(float64(snapshot.SequenceId))
 
+	return true, nil
+}
+
+// markReady releases waitForReady. Only an applied snapshot signals it; updates never do.
+func (c *consumer) markReady() {
 	c.readyOnce.Do(func() {
 		close(c.readyCh)
 		c.ready.Store(true)
 		duration := time.Since(c.startTime)
 		c.timeToFirstSnapshot.Set(duration.Seconds())
 		c.log.Infof("configstreamconsumer[%s]: first snapshot applied after %v", c.params.ClientName, duration)
 	})
-
-	return nil
 }
 
 // recordLayer notes that the stream put key into source, so a later snapshot can retract the

comp/core/configstreamconsumer/impl/integration_test.go

@@ -631,3 +631,161 @@ remote_agent:
 		t.Fatal("OneShot did not complete")
 	}
 }
+
+// Cleanup matters here: the keys the remap touches live on the process-global config object.
+func overrideTestConfig(t *testing.T, dir, addr string) string {
+	t.Helper()
+	host, port, err := net.SplitHostPort(addr)
+	require.NoError(t, err)
+
+	datadogYaml := fmt.Sprintf(`
+cmd_host: %s
+cmd_port: %s
+auth_token_file_path: %s
+ipc_cert_file_path: %s
+remote_agent:
+  registry:
+    enabled: true
+  configstream:
+    consumer:
+      enabled: true
+`, host, port,
+		filepath.Join(dir, "auth_token"),
+		filepath.Join(dir, "ipc_cert.pem"),
+	)
+	datadogPath := filepath.Join(dir, "datadog.yaml")
+	require.NoError(t, os.WriteFile(datadogPath, []byte(datadogYaml), 0600))
+
+	t.Cleanup(func() {
+		cfg := configstreambootstrap.Config()
+		cfg.UnsetForSource("log_level", model.SourceAgentRuntime)
+		cfg.UnsetForSource("log_level", model.SourceFile)
+		cfg.UnsetForSource("security_agent.log_level", model.SourceFile)
+		cfg.UnsetForSource("apm_config.log_level", model.SourceFile)
+	})
+	return datadogPath
+}
+
+func TestSecurityAgentLogLevelOverridesBaseKey(t *testing.T) {
+	configstreambootstrap.UseDynamicSchema(t)
+	dir := t.TempDir()
+	addr, mock, cleanup := setupFakeCoreAgent(t, dir)
+	defer cleanup()
+
+	datadogPath := overrideTestConfig(t, dir, addr)
+
+	opts := fx.Options(
+		fx.Provide(func() log.Component { return logmock.New(t) }),
+		telemetryfx.Module(),
+		fx.Supply(configstreamconsumer.NewParams("security-agent", datadogPath, configstreamconsumer.WithReadyTimeout(10*time.Second))),
+		configstreamconsumerfx.Module(),
+	)
+
+	testRun := func(_ configstreamconsumer.Component) error {
+		cfg := configstreambootstrap.Config()
+		// Synchronous by contract: readiness is signalled only after the remap has run.
+		require.Equal(t, "debug", cfg.Get("log_level"))
+		require.Equal(t, model.SourceAgentRuntime, cfg.GetSource("log_level"))
+
+		mock.events <- &pb.ConfigEvent{
+			Event: &pb.ConfigEvent_Snapshot{
+				Snapshot: &pb.ConfigSnapshot{
+					SequenceId: 2,
+					Settings: []*pb.ConfigSetting{
+						{Key: "log_level", Value: mustNewValue(t, "info"), Source: string(model.SourceFile)},
+					},
+				},
+			},
+		}
+		require.Eventually(t, func() bool {
+			return cfg.Get("log_level") == "info"
+		}, 10*time.Second, 20*time.Millisecond, "the override outlived the namespaced key that produced it")
+		require.Equal(t, model.SourceFile, cfg.GetSource("log_level"))
+
+		mock.events <- &pb.ConfigEvent{
+			Event: &pb.ConfigEvent_Snapshot{
+				Snapshot: &pb.ConfigSnapshot{
+					SequenceId: 3,
+					Settings: []*pb.ConfigSetting{
+						{Key: "log_level", Value: mustNewValue(t, "warn"), Source: string(model.SourceFile)},
+						{Key: "security_agent.log_level", Value: mustNewValue(t, ""), Source: string(model.SourceFile)},
+					},
+				},
+			},
+		}
+		require.Eventually(t, func() bool {
+			return cfg.Get("log_level") == "warn"
+		}, 10*time.Second, 20*time.Millisecond, "the base value never took effect")
+		require.Equal(t, model.SourceFile, cfg.GetSource("log_level"), "an empty namespaced value must not override")
+		return nil
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- fxutil.OneShot(testRun, opts) }()
+
+	mock.events <- &pb.ConfigEvent{
+		Event: &pb.ConfigEvent_Snapshot{
+			Snapshot: &pb.ConfigSnapshot{
+				SequenceId: 1,
+				Settings: []*pb.ConfigSetting{
+					{Key: "log_level", Value: mustNewValue(t, "info"), Source: string(model.SourceFile)},
+					{Key: "security_agent.log_level", Value: mustNewValue(t, "debug"), Source: string(model.SourceFile)},
+				},
+			},
+		},
+	}
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(30 * time.Second):
+		t.Fatal("OneShot did not complete")
+	}
+}
+
+func TestLogLevelOverrideIsPerClient(t *testing.T) {
+	configstreambootstrap.UseDynamicSchema(t)
+	dir := t.TempDir()
+	addr, mock, cleanup := setupFakeCoreAgent(t, dir)
+	defer cleanup()
+
+	datadogPath := overrideTestConfig(t, dir, addr)
+
+	opts := fx.Options(
+		fx.Provide(func() log.Component { return logmock.New(t) }),
+		telemetryfx.Module(),
+		fx.Supply(configstreamconsumer.NewParams("trace-agent", datadogPath, configstreamconsumer.WithReadyTimeout(10*time.Second))),
+		configstreamconsumerfx.Module(),
+	)
+
+	testRun := func(_ configstreamconsumer.Component) error {
+		cfg := configstreambootstrap.Config()
+		require.Equal(t, "trace", cfg.Get("log_level"))
+		require.Equal(t, model.SourceAgentRuntime, cfg.GetSource("log_level"))
+		require.Equal(t, "error", cfg.Get("security_agent.log_level"), "another agent's key must be left untouched")
+		return nil
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- fxutil.OneShot(testRun, opts) }()
+
+	mock.events <- &pb.ConfigEvent{
+		Event: &pb.ConfigEvent_Snapshot{
+			Snapshot: &pb.ConfigSnapshot{
+				SequenceId: 1,
+				Settings: []*pb.ConfigSetting{
+					{Key: "log_level", Value: mustNewValue(t, "info"), Source: string(model.SourceFile)},
+					{Key: "apm_config.log_level", Value: mustNewValue(t, "trace"), Source: string(model.SourceFile)},
+					{Key: "security_agent.log_level", Value: mustNewValue(t, "error"), Source: string(model.SourceFile)},
+				},
+			},
+		},
+	}
+
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(30 * time.Second):
+		t.Fatal("OneShot did not complete")
+	}
+}

comp/core/configstreamconsumer/impl/overrides.go

@@ -0,0 +1,45 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-present Datadog, Inc.
+
+package configstreamconsumerimpl
+
+import (
+	pkgconfigmodel "github.com/DataDog/datadog-agent/pkg/config/model"
+	"github.com/DataDog/datadog-agent/pkg/configstreambootstrap"
+)
+
+// Declared entry by entry: most namespaced keys sharing a trailing name mean something different.
+// No system-probe entry: its key is absent from the core schema and config.Adjust already folds it on the sysprobe object.
+var overridesByClient = map[string]map[string]string{
+	"security-agent": {"security_agent.log_level": "log_level"},
+	"process-agent":  {"process_config.log_level": "log_level"},
+	"trace-agent":    {"apm_config.log_level": "log_level"},
+}
+
+// applyOverrides folds this client's namespaced settings onto their base keys, retractably.
+func (c *consumer) applyOverrides() {
+	overrides := overridesByClient[c.params.ClientName]
+	if len(overrides) == 0 {
+		return
+	}
+	cfg := configstreambootstrap.Config()
+	for namespacedKey, baseKey := range overrides {
+		// Non-string values are dropped: pkg/util/log/setup's log_level callback asserts to string unchecked.
+		value, _ := cfg.Get(namespacedKey).(string)
+		if value != "" {
+			// SourceAgentRuntime outranks file/env yet still loses to a streamed RC/CLI value; Set panics on SourceEnvVar.
+			cfg.Set(baseKey, value, pkgconfigmodel.SourceAgentRuntime)
+			if c.appliedOverrides == nil {
+				c.appliedOverrides = make(map[string]struct{}, len(overrides))
+			}
+			c.appliedOverrides[baseKey] = struct{}{}
+			continue
+		}
+		if _, written := c.appliedOverrides[baseKey]; written {
+			cfg.UnsetForSource(baseKey, pkgconfigmodel.SourceAgentRuntime)
+			delete(c.appliedOverrides, baseKey)
+		}
+	}
+}

comp/core/configstreamconsumer/impl/session_test.go

@@ -15,9 +15,12 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/structpb"
 
 	configstreamconsumer "github.com/DataDog/datadog-agent/comp/core/configstreamconsumer/def"
 	telemetrymock "github.com/DataDog/datadog-agent/comp/core/telemetry/mock"
+	pkgconfigmodel "github.com/DataDog/datadog-agent/pkg/config/model"
+	"github.com/DataDog/datadog-agent/pkg/configstreambootstrap"
 	pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/core"
 	pkglog "github.com/DataDog/datadog-agent/pkg/util/log"
 )
@@ -68,35 +71,76 @@ func TestApplySnapshotAfterStreamReset(t *testing.T) {
 	snapshot := func(seqID int32) *pb.ConfigSnapshot {
 		return &pb.ConfigSnapshot{SequenceId: seqID}
 	}
+	apply := func(t *testing.T, c *consumer, seqID int32) bool {
+		t.Helper()
+		applied, err := c.applySnapshot(snapshot(seqID))
+		require.NoError(t, err)
+		return applied
+	}
 
 	t.Run("lower sequence id is accepted on a new stream", func(t *testing.T) {
 		c := newTestConsumer(t)
-		require.NoError(t, c.applySnapshot(snapshot(100)))
+		require.True(t, apply(t, c, 100))
 		require.Equal(t, int32(100), c.lastSeqID.Load())
 
 		// A restarted core agent counts from zero again. connectAndStream resets the
 		// sequence ID before every stream, so the fresh snapshot must win.
 		c.lastSeqID.Store(seqIDUnset)
-		require.NoError(t, c.applySnapshot(snapshot(3)))
+		require.True(t, apply(t, c, 3))
 		require.Equal(t, int32(3), c.lastSeqID.Load())
 	})
 
 	t.Run("sequence id zero is accepted on a new stream", func(t *testing.T) {
 		c := newTestConsumer(t)
 		c.lastSeqID.Store(seqIDUnset)
-		require.NoError(t, c.applySnapshot(snapshot(0)))
+		require.NoError(t, c.handleConfigEvent(&pb.ConfigEvent{
+			Event: &pb.ConfigEvent_Snapshot{Snapshot: snapshot(0)},
+		}))
 		require.Equal(t, int32(0), c.lastSeqID.Load())
 		require.True(t, c.IsActive())
 	})
 
 	t.Run("stale snapshot within a stream is dropped", func(t *testing.T) {
 		c := newTestConsumer(t)
-		require.NoError(t, c.applySnapshot(snapshot(10)))
-		require.NoError(t, c.applySnapshot(snapshot(4)))
+		require.True(t, apply(t, c, 10))
+		require.False(t, apply(t, c, 4))
 		require.Equal(t, int32(10), c.lastSeqID.Load())
 	})
 }
 
+func TestOnlySnapshotsSignalReadiness(t *testing.T) {
+	event := func(seqID int32) *pb.ConfigEvent {
+		return &pb.ConfigEvent{Event: &pb.ConfigEvent_Update{Update: &pb.ConfigUpdate{
+			SequenceId: seqID,
+			Setting:    &pb.ConfigSetting{Key: "log_level", Value: structpb.NewStringValue("info"), Source: string(pkgconfigmodel.SourceFile)},
+		}}}
+	}
+	t.Cleanup(func() {
+		configstreambootstrap.Config().UnsetForSource("log_level", pkgconfigmodel.SourceFile)
+	})
+
+	c := newTestConsumer(t)
+	c.lastSeqID.Store(seqIDUnset)
+	require.NoError(t, c.handleConfigEvent(event(0)))
+	require.NoError(t, c.handleConfigEvent(event(1)))
+	require.False(t, c.IsActive())
+
+	require.NoError(t, c.handleConfigEvent(&pb.ConfigEvent{
+		Event: &pb.ConfigEvent_Snapshot{Snapshot: &pb.ConfigSnapshot{SequenceId: 2}},
+	}))
+	require.True(t, c.IsActive())
+}
+
+// A stale snapshot is not an apply, so it must not release waitForReady.
+func TestStaleSnapshotDoesNotSignalReadiness(t *testing.T) {
+	c := newTestConsumer(t)
+	c.lastSeqID.Store(10)
+	require.NoError(t, c.handleConfigEvent(&pb.ConfigEvent{
+		Event: &pb.ConfigEvent_Snapshot{Snapshot: &pb.ConfigSnapshot{SequenceId: 4}},
+	}))
+	require.False(t, c.IsActive())
+}
+
 func TestSessionRejected(t *testing.T) {
 	tests := []struct {
 		err  error