From a70f9be97377d0981329ee1946f4302a27728db8 Mon Sep 17 00:00:00 2001 From: "rahul.kaukuntla" Date: Fri, 4 Sep 2026 12:15:11 -0400 Subject: [PATCH] configstreamconsumer: remap per-agent log_level onto the base key Declares a per-ClientName table folding the namespaced log_level settings onto log_level after each applied snapshot or update, so the logger's OnUpdate callback (which matches the literal key) fires. Writes use SourceAgentRuntime and are retracted once the namespaced key stops carrying a value. Readiness moves out of applySnapshot into markReady, called after the remap, so waitForReady cannot release before the base key holds the override. --- .../configstreamconsumer/impl/BUILD.bazel | 1 + .../configstreamconsumer/impl/consumer.go | 32 +++- .../impl/integration_test.go | 158 ++++++++++++++++++ .../configstreamconsumer/impl/overrides.go | 45 +++++ .../configstreamconsumer/impl/session_test.go | 54 +++++- 5 files changed, 279 insertions(+), 11 deletions(-) create mode 100644 comp/core/configstreamconsumer/impl/overrides.go diff --git a/comp/core/configstreamconsumer/impl/BUILD.bazel b/comp/core/configstreamconsumer/impl/BUILD.bazel index a8fc02e0afd5..d451b7690659 100644 --- a/comp/core/configstreamconsumer/impl/BUILD.bazel +++ b/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"], diff --git a/comp/core/configstreamconsumer/impl/consumer.go b/comp/core/configstreamconsumer/impl/consumer.go index ad2d19822804..116e575ce721 100644 --- a/comp/core/configstreamconsumer/impl/consumer.go +++ b/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,6 +639,11 @@ 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) @@ -629,8 +651,6 @@ func (c *consumer) applySnapshot(snapshot *pb.ConfigSnapshot) error { 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 diff --git a/comp/core/configstreamconsumer/impl/integration_test.go b/comp/core/configstreamconsumer/impl/integration_test.go index 7b0ead9dc74e..3f2d767ec3b6 100644 --- a/comp/core/configstreamconsumer/impl/integration_test.go +++ b/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") + } +} diff --git a/comp/core/configstreamconsumer/impl/overrides.go b/comp/core/configstreamconsumer/impl/overrides.go new file mode 100644 index 000000000000..4df2d94a5504 --- /dev/null +++ b/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) + } + } +} diff --git a/comp/core/configstreamconsumer/impl/session_test.go b/comp/core/configstreamconsumer/impl/session_test.go index cb04c1579d7b..4be48b00ca69 100644 --- a/comp/core/configstreamconsumer/impl/session_test.go +++ b/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