From 820f1332ca5c4dd7f203ea6bb70645edcd3c0289 Mon Sep 17 00:00:00 2001 From: "rahul.kaukuntla" Date: Fri, 28 Aug 2026 14:02:57 -0400 Subject: [PATCH 1/4] configstream: add a ConfigUnset event so removals cross the wire Snapshots only ever carry the core agent's merged view -- one entry per key, tagged with the winning source -- so a subscriber has no lower layer of its own to fall back to when a layer is cleared. A ConfigUpdate cannot express a removal: it wrote the shadowed value back into the layer being cleared, leaving the subscriber diverged from the core agent. Reporting the resolved source instead would not help either, since the stale higher-priority entry would keep winning the merge. Add a ConfigUnset event carrying the cleared layer plus what the key resolves to without it, and a model.Config OnUnset hook to observe layer removals, which NotificationReceiver cannot describe. The consumer seeds the fallback value before applying the removal so local receivers see the final value once rather than transiently observing the default. --- comp/core/configstream/impl/configstream.go | 61 ++++- .../configstream/impl/configstream_test.go | 19 +- .../configstreamconsumer/impl/BUILD.bazel | 1 + .../configstreamconsumer/impl/consumer.go | 32 +++ .../impl/integration_test.go | 92 +++++++ pkg/config/model/types.go | 13 + pkg/config/nodetreemodel/config.go | 98 ++++---- pkg/config/nodetreemodel/config_test.go | 113 +++++++++ pkg/proto/datadog/model/v1/model.proto | 15 ++ pkg/proto/pbgo/core/model.pb.go | 224 +++++++++++++----- 10 files changed, 561 insertions(+), 107 deletions(-) diff --git a/comp/core/configstream/impl/configstream.go b/comp/core/configstream/impl/configstream.go index 84052ef93537..48a233c8ceab 100644 --- a/comp/core/configstream/impl/configstream.go +++ b/comp/core/configstream/impl/configstream.go @@ -55,6 +55,10 @@ type configStream struct { // Cached origin (set once at initialization to avoid lock contention) origin string + // lastUnsetSeqID suppresses the OnUpdate that UnsetForSource fires under the same sequence ID: + // it reports the shadowed value under the cleared source, which the ConfigUnset stands in for. + lastUnsetSeqID atomic.Uint64 + subscribersGauge telemetry.Gauge snapshotsSent telemetry.Counter updatesSent telemetry.Counter @@ -141,10 +145,44 @@ func (cs *configStream) Subscribe(req *pb.ConfigStreamRequest) (<-chan *pb.Confi func (cs *configStream) run() { updatesChan := make(chan *pb.ConfigEvent, 100) + cs.config.OnUnset(func(setting string, clearedSource model.Source, resolvedValue interface{}, resolvedSource model.Source, sequenceID uint64) { + if cs.stopped.Load() { + return + } + cs.lastUnsetSeqID.Store(sequenceID) + + unset := &pb.ConfigUnset{ + SequenceId: int32(sequenceID), + Origin: cs.origin, + Key: setting, + Source: clearedSource.String(), + } + + // Left unset when the key resolves to nothing, which tells the subscriber to only drop it. + if resolvedSource != model.SourceUnknown { + if resolved, err := newConfigSetting(setting, resolvedValue, resolvedSource); err != nil { + cs.log.Errorf("Failed to encode post-unset value of '%s': %v", setting, err) + } else { + unset.Resolved = resolved + } + } + + configUnset := &pb.ConfigEvent{Event: &pb.ConfigEvent_Unset{Unset: unset}} + + select { + case updatesChan <- configUnset: + default: + cs.log.Warn("Config update channel is full, dropping unset.") + } + }) + cs.config.OnUpdate(func(setting string, source model.Source, _, newValue interface{}, sequenceID uint64) { if cs.stopped.Load() { return } + if cs.lastUnsetSeqID.Load() == sequenceID { + return + } sanitizedValue, err := sanitizeValue(newValue) if err != nil { cs.log.Warnf("Failed to sanitize setting '%s': %v", setting, err) @@ -236,6 +274,27 @@ func (cs *configStream) removeSubscriber(id string) { } } +// newConfigSetting encodes one resolved setting for the wire. +func newConfigSetting(key string, value interface{}, source model.Source) (*pb.ConfigSetting, error) { + sanitizedValue, err := sanitizeValue(value) + if err != nil { + return nil, err + } + pbValue, err := structpb.NewValue(sanitizedValue) + if err != nil { + return nil, err + } + return &pb.ConfigSetting{Key: key, Source: source.String(), Value: pbValue}, nil +} + +// incrementalSequenceID reads the sequence ID off either incremental event kind; both share a counter. +func incrementalSequenceID(event *pb.ConfigEvent) uint64 { + if unset := event.GetUnset(); unset != nil { + return uint64(unset.SequenceId) + } + return uint64(event.GetUpdate().SequenceId) +} + func (cs *configStream) handleConfigUpdate(event *pb.ConfigEvent) { cs.m.Lock() defer cs.m.Unlock() @@ -244,7 +303,7 @@ func (cs *configStream) handleConfigUpdate(event *pb.ConfigEvent) { var snapshotSeqID uint64 var snapshotErr error - currentSequenceID := uint64(event.GetUpdate().SequenceId) + currentSequenceID := incrementalSequenceID(event) for id, sub := range cs.subscribers { // Skip updates that are older than the last one we sent to this subscriber. diff --git a/comp/core/configstream/impl/configstream_test.go b/comp/core/configstream/impl/configstream_test.go index 2f1161eb676d..a52aaa445b7b 100644 --- a/comp/core/configstream/impl/configstream_test.go +++ b/comp/core/configstream/impl/configstream_test.go @@ -467,19 +467,22 @@ func TestConfigStream(t *testing.T) { require.Equal(t, "new_value", update.Update.Setting.Value.GetStringValue()) configComp.UnsetForSource("my.new.setting", model.SourceCLI) - // verify we receive the update for the unset. + select { case event = <-eventsCh: case <-time.After(2 * time.Second): - t.Fatal("timed out waiting for config update") + t.Fatal("timed out waiting for config unset") } require.NotNil(t, event) - update, isUpdate = event.GetEvent().(*pb.ConfigEvent_Update) - require.True(t, isUpdate, "unset event must be an update") - - // verify that the value has been unset and back to the original value. - require.Equal(t, "my.new.setting", update.Update.Setting.Key) - require.Equal(t, "original_value", update.Update.Setting.Value.GetStringValue()) + unset, isUnset := event.GetEvent().(*pb.ConfigEvent_Unset) + require.True(t, isUnset, "unset must be its own event kind, got %T", event.GetEvent()) + require.Equal(t, "my.new.setting", unset.Unset.Key) + require.Equal(t, string(model.SourceCLI), unset.Unset.Source, "the cleared layer, not the fallback") + + // Subscribers mirror the merged view, so the removal has to say what the key resolves to now. + require.NotNil(t, unset.Unset.Resolved, "unset must carry the post-unset resolution") + require.Equal(t, "original_value", unset.Unset.Resolved.Value.GetStringValue()) + require.Equal(t, string(model.SourceAgentRuntime), unset.Unset.Resolved.Source) }) resyncsWithSnapshotOnDiscontinuity := func(t *testing.T) { diff --git a/comp/core/configstreamconsumer/impl/BUILD.bazel b/comp/core/configstreamconsumer/impl/BUILD.bazel index b5eb932b44bf..9de33b16c238 100644 --- a/comp/core/configstreamconsumer/impl/BUILD.bazel +++ b/comp/core/configstreamconsumer/impl/BUILD.bazel @@ -46,6 +46,7 @@ dd_agent_go_test( "//comp/core/log/mock", "//comp/core/telemetry/fx", "//pkg/api/security/cert", + "//pkg/config/model", "//pkg/configstreambootstrap", "//pkg/proto/pbgo/core", "//pkg/util/fxutil", diff --git a/comp/core/configstreamconsumer/impl/consumer.go b/comp/core/configstreamconsumer/impl/consumer.go index c0645c14d409..dd5e0bfc7894 100644 --- a/comp/core/configstreamconsumer/impl/consumer.go +++ b/comp/core/configstreamconsumer/impl/consumer.go @@ -318,6 +318,8 @@ func (c *consumer) handleConfigEvent(event *pb.ConfigEvent) error { return c.applySnapshot(e.Snapshot) case *pb.ConfigEvent_Update: return c.applyUpdate(e.Update) + case *pb.ConfigEvent_Unset: + return c.applyUnset(e.Unset) default: return fmt.Errorf("unknown event type: %T", event.Event) } @@ -393,6 +395,36 @@ func (c *consumer) applyUpdate(update *pb.ConfigUpdate) error { return nil } +// applyUnset mirrors an UnsetForSource performed on the core agent. Seeding the fallback layer is not +// optional: snapshots only carry the merged view, so this config has no lower layer of its own to +// fall back to. It is also seeded before the unset rather than after, so the unset resolves onto it +// and notifies local receivers once with the final value instead of transiently with the default. +func (c *consumer) applyUnset(unset *pb.ConfigUnset) error { + if unset.SequenceId <= c.lastSeqID.Load() { + c.log.Warnf("Ignoring stale unset (seq_id: %d <= %d)", unset.SequenceId, c.lastSeqID.Load()) + c.droppedStaleUpdates.Inc() + return nil + } + + if unset.SequenceId != c.lastSeqID.Load()+1 { + return fmt.Errorf("seq_id discontinuity: expected %d, got %d", c.lastSeqID.Load()+1, unset.SequenceId) + } + + c.log.Debugf("Applying config unset (seq_id: %d, key: %s, source: %s)", unset.SequenceId, unset.Key, unset.Source) + + cfg := configstreambootstrap.Config() + // Absent when the key resolves to nothing on the core agent, leaving only the removal to mirror. + if resolved := unset.GetResolved(); resolved != nil { + cfg.DirectBulkSet([]pkgconfigmodel.DirectSetting{toDirectSetting(resolved)}) + } + cfg.UnsetForSource(unset.Key, pkgconfigmodel.Source(unset.Source)) + + c.lastSeqID.Store(unset.SequenceId) + c.lastSeqIDMetric.Set(float64(unset.SequenceId)) + + return nil +} + func (c *consumer) initMetrics() { c.timeToFirstSnapshot = c.telemetry.NewGauge( "configstream_consumer", diff --git a/comp/core/configstreamconsumer/impl/integration_test.go b/comp/core/configstreamconsumer/impl/integration_test.go index 7e388b444022..d6ba51e6cb59 100644 --- a/comp/core/configstreamconsumer/impl/integration_test.go +++ b/comp/core/configstreamconsumer/impl/integration_test.go @@ -40,6 +40,7 @@ import ( logmock "github.com/DataDog/datadog-agent/comp/core/log/mock" telemetryfx "github.com/DataDog/datadog-agent/comp/core/telemetry/fx" "github.com/DataDog/datadog-agent/pkg/api/security/cert" + "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" "github.com/DataDog/datadog-agent/pkg/util/fxutil" @@ -245,3 +246,94 @@ func TestRunNoopWhenConfigstreamDisabled(t *testing.T) { t.Fatal("OneShot blocked unexpectedly when configstream is disabled") } } + +// TestUnsetEventRestoresShadowedValue covers the scenario from the review thread: a value supplied by +// an env var, shadowed by an agent-runtime write, then unset. Snapshots only carry the core agent's +// merged view, and the consumer's own env layer is deliberately disabled, so it has no lower layer to +// fall back to -- the unset event has to supply the resolution or the key silently reverts to default. +func TestUnsetEventRestoresShadowedValue(t *testing.T) { + configstreambootstrap.UseDynamicSchema(t) + dir := t.TempDir() + addr, mock, cleanup := setupFakeCoreAgent(t, dir) + defer cleanup() + + 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)) + + 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, "from-runtime", cfg.Get("test.key")) + require.Equal(t, model.SourceAgentRuntime, cfg.GetSource("test.key")) + + // SourceEnvVar is why this cannot be applied as an ordinary update: Set rejects writes to + // the env layer, and the consumer's local one is cleared at startup. + mock.events <- &pb.ConfigEvent{ + Event: &pb.ConfigEvent_Unset{ + Unset: &pb.ConfigUnset{ + SequenceId: 2, + Key: "test.key", + Source: string(model.SourceAgentRuntime), + Resolved: &pb.ConfigSetting{ + Key: "test.key", + Value: mustNewValue(t, "from-env"), + Source: string(model.SourceEnvVar), + }, + }, + }, + } + + require.Eventually(t, func() bool { + return cfg.Get("test.key") == "from-env" + }, 10*time.Second, 20*time.Millisecond, "key never fell back to the shadowed env value") + require.Equal(t, model.SourceEnvVar, cfg.GetSource("test.key"), "the layer the core agent resolves from") + return nil + } + + done := make(chan error, 1) + go func() { done <- fxutil.OneShot(testRun, opts) }() + + // One entry per key, as createConfigSnapshot sends: the resolved value tagged with the winning + // layer. The shadowed env value is never transmitted. + mock.events <- &pb.ConfigEvent{ + Event: &pb.ConfigEvent_Snapshot{ + Snapshot: &pb.ConfigSnapshot{ + SequenceId: 1, + Settings: []*pb.ConfigSetting{ + {Key: "test.key", Value: mustNewValue(t, "from-runtime"), Source: string(model.SourceAgentRuntime)}, + }, + }, + }, + } + + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(30 * time.Second): + t.Fatal("OneShot did not complete") + } +} diff --git a/pkg/config/model/types.go b/pkg/config/model/types.go index 36cf8be9caf3..8e4bbc4019b1 100644 --- a/pkg/config/model/types.go +++ b/pkg/config/model/types.go @@ -139,8 +139,18 @@ type Proxy struct { // NotificationReceiver represents the callback type to receive notifications each time the `Set` method is called. The // configuration will call each NotificationReceiver registered through the 'OnUpdate' method, therefore // 'NotificationReceiver' should not be blocking. +// +// source is the layer 'newValue' resolves from, which for an unset is the layer being fallen back to +// rather than the one cleared. Use 'OnUnset' to observe which layer changed. type NotificationReceiver func(setting string, source Source, oldValue, newValue any, sequenceID uint64) +// UnsetNotificationReceiver receives the removal of a setting from one source layer, which +// NotificationReceiver cannot express: it names the cleared layer and reports what the setting +// resolves to without it, so a config mirroring only another one's merged view can reproduce the +// result. resolvedSource is SourceUnknown when no layer is left to fall back to. Fires even when the +// resolved value is unchanged, before the NotificationReceiver call sharing its sequence ID. +type UnsetNotificationReceiver func(setting string, clearedSource Source, resolvedValue any, resolvedSource Source, sequenceID uint64) + // Reader is a subset of Config that only allows reading of configuration type Reader interface { Get(key string) interface{} @@ -216,6 +226,9 @@ type Reader interface { // by a call to the 'Set' method. The configuration will sequentially call each receiver. OnUpdate(callback NotificationReceiver) + // OnUnset adds a callback called each time 'UnsetForSource' removes a setting from a source layer. + OnUnset(callback UnsetNotificationReceiver) + // Stringify stringifies the config, only available if "test" build tag is enabled Stringify(source Source, opts ...StringifyOption) string } diff --git a/pkg/config/nodetreemodel/config.go b/pkg/config/nodetreemodel/config.go index c392de5b1a76..46a20b01996d 100644 --- a/pkg/config/nodetreemodel/config.go +++ b/pkg/config/nodetreemodel/config.go @@ -111,6 +111,7 @@ type ntmConfig struct { envTransform map[string]func(string) interface{} notificationReceivers []model.NotificationReceiver + unsetReceivers []model.UnsetNotificationReceiver sequenceID uint64 // Proxy settings @@ -177,6 +178,13 @@ func (c *ntmConfig) OnUpdate(callback model.NotificationReceiver) { c.notificationReceivers = append(c.notificationReceivers, callback) } +// OnUnset adds a receiver called on every layer removal, including ones that leave the resolved value unchanged. +func (c *ntmConfig) OnUnset(callback model.UnsetNotificationReceiver) { + c.Lock() + defer c.Unlock() + c.unsetReceivers = append(c.unsetReceivers, callback) +} + func (c *ntmConfig) getTreeBySource(source model.Source) (*nodeImpl, error) { switch source { case "root": @@ -413,13 +421,16 @@ func (c *ntmConfig) UnsetForSource(key string, source model.Source) { c.maybeRebuild() var ( - previousValue interface{} - newValue interface{} - receivers []model.NotificationReceiver - sequenceID uint64 + previousValue interface{} + resolvedValue interface{} + resolvedSource model.Source + receivers []model.NotificationReceiver + unsetReceivers []model.UnsetNotificationReceiver + sequenceID uint64 + notifyUpdate bool ) - ok := func() bool { + func() { c.Lock() defer c.Unlock() @@ -430,69 +441,74 @@ func (c *ntmConfig) UnsetForSource(key string, source model.Source) { tree, err := c.getTreeBySource(source) if err != nil { log.Errorf("%s", err) - return false + return } parentNode, childName, err := c.parentOfNode(tree, key) if err != nil { - return false + return } // Only remove if the setting is a leaf + removed := false if child, err := parentNode.GetChild(childName); err == nil { if child.IsLeafNode() { parentNode.RemoveChild(childName) + removed = true } else { log.Errorf("cannot remove setting %q, not a leaf", key) - return false + return } } - - // If the node in the merged tree doesn't match the source we expect, we're done - if c.leafAtPathFromNode(key, c.root).Source() != source { - return false - } - - // Find what the previous value used to be, based upon the previous source - prevNode, findPreviousSourceError := c.findPreviousSourceNode(key, source) - - // Get the parent node of the leaf we're unsetting - parentNode, childName, err = c.parentOfNode(c.root, key) - if err != nil { - return false + // Nothing left the layer, so root cannot name this source as its winner either. + if !removed { + return } - // If there was no previous source with a node of this name, simply remove it from the parent - if findPreviousSourceError != nil { - parentNode.RemoveChild(childName) - return false + // Allocated on removal, not on value change: a mirror drops the entry either way. + c.sequenceID++ + sequenceID = c.sequenceID + unsetReceivers = slices.Clone(c.unsetReceivers) + + // The merged tree only needs mending when the layer we cleared was the one winning in it. + if c.leafAtPathFromNode(key, c.root).Source() == source { + prevNode, findPreviousSourceError := c.findPreviousSourceNode(key, source) + if rootParent, rootChild, err := c.parentOfNode(c.root, key); err == nil { + if findPreviousSourceError != nil { + // No lower layer holds this key, so it leaves the merged tree entirely. + rootParent.RemoveChild(rootChild) + } else { + rootParent.InsertChildNode(rootChild, prevNode) + } + } } - // Replace the child with the node from the previous layer - parentNode.InsertChildNode(childName, prevNode) - - newValue = c.leafAtPathFromNode(key, c.root).Get() + // Read once the merged tree has settled, so every removal path reports the same thing: what + // the key resolves to now. missingLeaf reports SourceUnknown, meaning nothing is left. + resolved := c.leafAtPathFromNode(key, c.root) + resolvedValue = resolved.Get() + resolvedSource = resolved.Source() // Value has not changed, do not notify - if reflect.DeepEqual(previousValue, newValue) { - return false + if reflect.DeepEqual(previousValue, resolvedValue) { + return } - c.sequenceID++ receivers = slices.Clone(c.notificationReceivers) - // Capture the sequenceID here whilst locked to send to the receivers - // after unlocking. - sequenceID = c.sequenceID - return true + notifyUpdate = true }() - if !ok { - return - } - // Notify receivers outside the lock. Subscribers commonly read the // config from within their callback, and doing so while the write // lock is still held deadlocks them against this goroutine. + // Unset receivers run first so a mirror drops the entry before the value change is reported. + for _, receiver := range unsetReceivers { + receiver(key, source, resolvedValue, resolvedSource, sequenceID) + } + + if !notifyUpdate { + return + } for _, receiver := range receivers { - receiver(key, source, previousValue, newValue, sequenceID) + receiver(key, resolvedSource, previousValue, resolvedValue, sequenceID) } } diff --git a/pkg/config/nodetreemodel/config_test.go b/pkg/config/nodetreemodel/config_test.go index c21a4a201bb7..6e8651a4aec0 100644 --- a/pkg/config/nodetreemodel/config_test.go +++ b/pkg/config/nodetreemodel/config_test.go @@ -2417,3 +2417,116 @@ b: }, res) } + +func TestOnUnsetReportsEveryLayerRemoval(t *testing.T) { + type unsetEvent struct { + key string + clearedSource model.Source + resolvedValue any + resolvedSource model.Source + seqID uint64 + } + type updateEvent struct { + source model.Source + value any + seqID uint64 + } + + newCfg := func() model.Config { + cfg := NewNodeTreeConfig("test", "TEST", nil) + cfg.SetDefault("shadowed", "default") + cfg.BuildSchema() + return cfg + } + + // Attached after setup writes so only the unset under test is recorded. + watch := func(cfg model.Config) (*[]unsetEvent, *[]updateEvent) { + unsets := &[]unsetEvent{} + updates := &[]updateEvent{} + cfg.OnUnset(func(key string, clearedSource model.Source, resolvedValue any, resolvedSource model.Source, seqID uint64) { + *unsets = append(*unsets, unsetEvent{key, clearedSource, resolvedValue, resolvedSource, seqID}) + }) + cfg.OnUpdate(func(_ string, source model.Source, _, newValue any, seqID uint64) { + *updates = append(*updates, updateEvent{source, newValue, seqID}) + }) + return unsets, updates + } + + t.Run("falls back to a lower layer", func(t *testing.T) { + cfg := newCfg() + cfg.Set("shadowed", "from_file", model.SourceFile) + cfg.Set("shadowed", "from_cli", model.SourceCLI) + unsets, updates := watch(cfg) + + cfg.UnsetForSource("shadowed", model.SourceCLI) + + require.Len(t, *unsets, 1) + assert.Equal(t, "shadowed", (*unsets)[0].key) + assert.Equal(t, model.SourceCLI, (*unsets)[0].clearedSource, "the cleared layer, not the fallback") + // The pair a mirror needs: without it, dropping the CLI entry leaves it on the default. + assert.Equal(t, "from_file", (*unsets)[0].resolvedValue) + assert.Equal(t, model.SourceFile, (*unsets)[0].resolvedSource) + assert.Equal(t, "from_file", cfg.Get("shadowed")) + + // One mutation, one sequence ID, so a subscriber tracking continuity sees no gap. + require.Len(t, *updates, 1) + assert.Equal(t, (*unsets)[0].seqID, (*updates)[0].seqID) + // The update names where the value came from, not the layer that was cleared. + assert.Equal(t, model.SourceFile, (*updates)[0].source) + assert.Equal(t, "from_file", (*updates)[0].value) + }) + + t.Run("resolved value unchanged because a higher layer still wins", func(t *testing.T) { + cfg := newCfg() + cfg.Set("shadowed", "from_file", model.SourceFile) + cfg.Set("shadowed", "from_cli", model.SourceCLI) + unsets, updates := watch(cfg) + + // File is outranked by CLI, so a mirror must still drop the entry or it resurfaces later. + cfg.UnsetForSource("shadowed", model.SourceFile) + + require.Len(t, *unsets, 1) + assert.Equal(t, model.SourceFile, (*unsets)[0].clearedSource) + assert.Equal(t, "from_cli", (*unsets)[0].resolvedValue, "still the winning layer") + assert.Equal(t, model.SourceCLI, (*unsets)[0].resolvedSource) + assert.Empty(t, *updates, "no value change, so no update notification") + }) + + t.Run("falls back to the default layer", func(t *testing.T) { + cfg := newCfg() + cfg.Set("shadowed", "from_cli", model.SourceCLI) + unsets, _ := watch(cfg) + + cfg.UnsetForSource("shadowed", model.SourceCLI) + + require.Len(t, *unsets, 1) + assert.Equal(t, "default", (*unsets)[0].resolvedValue) + assert.Equal(t, model.SourceDefault, (*unsets)[0].resolvedSource) + }) + + t.Run("nothing left to fall back to", func(t *testing.T) { + cfg := NewNodeTreeConfig("test", "TEST", nil) + cfg.BuildSchema() + cfg.SetTestOnlyDynamicSchema(true) + cfg.Set("undeclared", "from_cli", model.SourceCLI) + unsets, _ := watch(cfg) + + cfg.UnsetForSource("undeclared", model.SourceCLI) + + require.Len(t, *unsets, 1) + // SourceUnknown tells a mirror to drop the key outright rather than seed a fallback. + assert.Equal(t, model.SourceUnknown, (*unsets)[0].resolvedSource) + assert.Nil(t, (*unsets)[0].resolvedValue) + }) + + t.Run("nothing in the layer to remove", func(t *testing.T) { + cfg := newCfg() + cfg.Set("shadowed", "from_file", model.SourceFile) + unsets, updates := watch(cfg) + + cfg.UnsetForSource("shadowed", model.SourceCLI) + + assert.Empty(t, *unsets, "no removal happened, so nothing to report") + assert.Empty(t, *updates) + }) +} diff --git a/pkg/proto/datadog/model/v1/model.proto b/pkg/proto/datadog/model/v1/model.proto index 9f12dba89608..a5dbf5906d7c 100644 --- a/pkg/proto/datadog/model/v1/model.proto +++ b/pkg/proto/datadog/model/v1/model.proto @@ -168,9 +168,24 @@ message ConfigUpdate { ConfigSetting setting = 3; } +// ConfigUnset clears a setting from one source layer. A ConfigUpdate cannot express this: it would +// write the shadowed value into the layer being cleared. +message ConfigUnset { + string origin = 1; + int32 sequence_id = 2; + string key = 3; + // The source layer to clear, not the layer the value falls back to. + string source = 4; + // What the key resolves to on the sender once the layer is cleared. Receivers mirror the sender's + // merged view rather than its layers, so they have no lower layer of their own to fall back to. + // Unset when nothing is left to fall back to. + ConfigSetting resolved = 5; +} + message ConfigEvent { oneof event { ConfigSnapshot snapshot = 1; ConfigUpdate update = 2; + ConfigUnset unset = 3; } } diff --git a/pkg/proto/pbgo/core/model.pb.go b/pkg/proto/pbgo/core/model.pb.go index cf669445d1ce..4cdb572bbba9 100644 --- a/pkg/proto/pbgo/core/model.pb.go +++ b/pkg/proto/pbgo/core/model.pb.go @@ -1393,12 +1393,95 @@ func (x *ConfigUpdate) GetSetting() *ConfigSetting { return nil } +// ConfigUnset clears a setting from one source layer. A ConfigUpdate cannot express this: it would +// write the shadowed value into the layer being cleared. +type ConfigUnset struct { + state protoimpl.MessageState `protogen:"open.v1"` + Origin string `protobuf:"bytes,1,opt,name=origin,proto3" json:"origin,omitempty"` + SequenceId int32 `protobuf:"varint,2,opt,name=sequence_id,json=sequenceId,proto3" json:"sequence_id,omitempty"` + Key string `protobuf:"bytes,3,opt,name=key,proto3" json:"key,omitempty"` + // The source layer to clear, not the layer the value falls back to. + Source string `protobuf:"bytes,4,opt,name=source,proto3" json:"source,omitempty"` + // What the key resolves to on the sender once the layer is cleared. Receivers mirror the sender's + // merged view rather than its layers, so they have no lower layer of their own to fall back to. + // Unset when nothing is left to fall back to. + Resolved *ConfigSetting `protobuf:"bytes,5,opt,name=resolved,proto3" json:"resolved,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ConfigUnset) Reset() { + *x = ConfigUnset{} + mi := &file_datadog_model_v1_model_proto_msgTypes[23] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ConfigUnset) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ConfigUnset) ProtoMessage() {} + +func (x *ConfigUnset) ProtoReflect() protoreflect.Message { + mi := &file_datadog_model_v1_model_proto_msgTypes[23] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ConfigUnset.ProtoReflect.Descriptor instead. +func (*ConfigUnset) Descriptor() ([]byte, []int) { + return file_datadog_model_v1_model_proto_rawDescGZIP(), []int{23} +} + +func (x *ConfigUnset) GetOrigin() string { + if x != nil { + return x.Origin + } + return "" +} + +func (x *ConfigUnset) GetSequenceId() int32 { + if x != nil { + return x.SequenceId + } + return 0 +} + +func (x *ConfigUnset) GetKey() string { + if x != nil { + return x.Key + } + return "" +} + +func (x *ConfigUnset) GetSource() string { + if x != nil { + return x.Source + } + return "" +} + +func (x *ConfigUnset) GetResolved() *ConfigSetting { + if x != nil { + return x.Resolved + } + return nil +} + type ConfigEvent struct { state protoimpl.MessageState `protogen:"open.v1"` // Types that are valid to be assigned to Event: // // *ConfigEvent_Snapshot // *ConfigEvent_Update + // *ConfigEvent_Unset Event isConfigEvent_Event `protobuf_oneof:"event"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -1406,7 +1489,7 @@ type ConfigEvent struct { func (x *ConfigEvent) Reset() { *x = ConfigEvent{} - mi := &file_datadog_model_v1_model_proto_msgTypes[23] + mi := &file_datadog_model_v1_model_proto_msgTypes[24] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1418,7 +1501,7 @@ func (x *ConfigEvent) String() string { func (*ConfigEvent) ProtoMessage() {} func (x *ConfigEvent) ProtoReflect() protoreflect.Message { - mi := &file_datadog_model_v1_model_proto_msgTypes[23] + mi := &file_datadog_model_v1_model_proto_msgTypes[24] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1431,7 +1514,7 @@ func (x *ConfigEvent) ProtoReflect() protoreflect.Message { // Deprecated: Use ConfigEvent.ProtoReflect.Descriptor instead. func (*ConfigEvent) Descriptor() ([]byte, []int) { - return file_datadog_model_v1_model_proto_rawDescGZIP(), []int{23} + return file_datadog_model_v1_model_proto_rawDescGZIP(), []int{24} } func (x *ConfigEvent) GetEvent() isConfigEvent_Event { @@ -1459,6 +1542,15 @@ func (x *ConfigEvent) GetUpdate() *ConfigUpdate { return nil } +func (x *ConfigEvent) GetUnset() *ConfigUnset { + if x != nil { + if x, ok := x.Event.(*ConfigEvent_Unset); ok { + return x.Unset + } + } + return nil +} + type isConfigEvent_Event interface { isConfigEvent_Event() } @@ -1471,10 +1563,16 @@ type ConfigEvent_Update struct { Update *ConfigUpdate `protobuf:"bytes,2,opt,name=update,proto3,oneof"` } +type ConfigEvent_Unset struct { + Unset *ConfigUnset `protobuf:"bytes,3,opt,name=unset,proto3,oneof"` +} + func (*ConfigEvent_Snapshot) isConfigEvent_Event() {} func (*ConfigEvent_Update) isConfigEvent_Event() {} +func (*ConfigEvent_Unset) isConfigEvent_Event() {} + // Nested message for the local data type GenerateContainerIDFromOriginInfoRequest_LocalData struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -1488,7 +1586,7 @@ type GenerateContainerIDFromOriginInfoRequest_LocalData struct { func (x *GenerateContainerIDFromOriginInfoRequest_LocalData) Reset() { *x = GenerateContainerIDFromOriginInfoRequest_LocalData{} - mi := &file_datadog_model_v1_model_proto_msgTypes[24] + mi := &file_datadog_model_v1_model_proto_msgTypes[25] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1500,7 +1598,7 @@ func (x *GenerateContainerIDFromOriginInfoRequest_LocalData) String() string { func (*GenerateContainerIDFromOriginInfoRequest_LocalData) ProtoMessage() {} func (x *GenerateContainerIDFromOriginInfoRequest_LocalData) ProtoReflect() protoreflect.Message { - mi := &file_datadog_model_v1_model_proto_msgTypes[24] + mi := &file_datadog_model_v1_model_proto_msgTypes[25] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1556,7 +1654,7 @@ type GenerateContainerIDFromOriginInfoRequest_ExternalData struct { func (x *GenerateContainerIDFromOriginInfoRequest_ExternalData) Reset() { *x = GenerateContainerIDFromOriginInfoRequest_ExternalData{} - mi := &file_datadog_model_v1_model_proto_msgTypes[25] + mi := &file_datadog_model_v1_model_proto_msgTypes[26] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1568,7 +1666,7 @@ func (x *GenerateContainerIDFromOriginInfoRequest_ExternalData) String() string func (*GenerateContainerIDFromOriginInfoRequest_ExternalData) ProtoMessage() {} func (x *GenerateContainerIDFromOriginInfoRequest_ExternalData) ProtoReflect() protoreflect.Message { - mi := &file_datadog_model_v1_model_proto_msgTypes[25] + mi := &file_datadog_model_v1_model_proto_msgTypes[26] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1720,10 +1818,18 @@ const file_datadog_model_v1_model_proto_rawDesc = "" + "\x06origin\x18\x01 \x01(\tR\x06origin\x12\x1f\n" + "\vsequence_id\x18\x02 \x01(\x05R\n" + "sequenceId\x129\n" + - "\asetting\x18\x03 \x01(\v2\x1f.datadog.model.v1.ConfigSettingR\asetting\"\x90\x01\n" + + "\asetting\x18\x03 \x01(\v2\x1f.datadog.model.v1.ConfigSettingR\asetting\"\xad\x01\n" + + "\vConfigUnset\x12\x16\n" + + "\x06origin\x18\x01 \x01(\tR\x06origin\x12\x1f\n" + + "\vsequence_id\x18\x02 \x01(\x05R\n" + + "sequenceId\x12\x10\n" + + "\x03key\x18\x03 \x01(\tR\x03key\x12\x16\n" + + "\x06source\x18\x04 \x01(\tR\x06source\x12;\n" + + "\bresolved\x18\x05 \x01(\v2\x1f.datadog.model.v1.ConfigSettingR\bresolved\"\xc7\x01\n" + "\vConfigEvent\x12>\n" + "\bsnapshot\x18\x01 \x01(\v2 .datadog.model.v1.ConfigSnapshotH\x00R\bsnapshot\x128\n" + - "\x06update\x18\x02 \x01(\v2\x1e.datadog.model.v1.ConfigUpdateH\x00R\x06updateB\a\n" + + "\x06update\x18\x02 \x01(\v2\x1e.datadog.model.v1.ConfigUpdateH\x00R\x06update\x125\n" + + "\x05unset\x18\x03 \x01(\v2\x1d.datadog.model.v1.ConfigUnsetH\x00R\x05unsetB\a\n" + "\x05event*1\n" + "\tEventType\x12\t\n" + "\x05ADDED\x10\x00\x12\f\n" + @@ -1747,39 +1853,40 @@ func file_datadog_model_v1_model_proto_rawDescGZIP() []byte { } var file_datadog_model_v1_model_proto_enumTypes = make([]protoimpl.EnumInfo, 2) -var file_datadog_model_v1_model_proto_msgTypes = make([]protoimpl.MessageInfo, 28) +var file_datadog_model_v1_model_proto_msgTypes = make([]protoimpl.MessageInfo, 29) var file_datadog_model_v1_model_proto_goTypes = []any{ - (EventType)(0), // 0: datadog.model.v1.EventType - (TagCardinality)(0), // 1: datadog.model.v1.TagCardinality - (*HostnameRequest)(nil), // 2: datadog.model.v1.HostnameRequest - (*HostnameReply)(nil), // 3: datadog.model.v1.HostnameReply - (*HostTagRequest)(nil), // 4: datadog.model.v1.HostTagRequest - (*HostTagReply)(nil), // 5: datadog.model.v1.HostTagReply - (*CaptureTriggerRequest)(nil), // 6: datadog.model.v1.CaptureTriggerRequest - (*CaptureTriggerResponse)(nil), // 7: datadog.model.v1.CaptureTriggerResponse - (*StreamTagsRequest)(nil), // 8: datadog.model.v1.StreamTagsRequest - (*StreamTagsResponse)(nil), // 9: datadog.model.v1.StreamTagsResponse - (*StreamTagsEvent)(nil), // 10: datadog.model.v1.StreamTagsEvent - (*DeprecatedFilter)(nil), // 11: datadog.model.v1.DeprecatedFilter - (*Entity)(nil), // 12: datadog.model.v1.Entity - (*GenerateContainerIDFromOriginInfoRequest)(nil), // 13: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest - (*GenerateContainerIDFromOriginInfoResponse)(nil), // 14: datadog.model.v1.GenerateContainerIDFromOriginInfoResponse - (*FetchEntityRequest)(nil), // 15: datadog.model.v1.FetchEntityRequest - (*FetchEntityResponse)(nil), // 16: datadog.model.v1.FetchEntityResponse - (*EntityId)(nil), // 17: datadog.model.v1.EntityId - (*UnixDogstatsdMsg)(nil), // 18: datadog.model.v1.UnixDogstatsdMsg - (*TaggerState)(nil), // 19: datadog.model.v1.TaggerState - (*TaggerStateResponse)(nil), // 20: datadog.model.v1.TaggerStateResponse - (*ConfigStreamRequest)(nil), // 21: datadog.model.v1.ConfigStreamRequest - (*ConfigSetting)(nil), // 22: datadog.model.v1.ConfigSetting - (*ConfigSnapshot)(nil), // 23: datadog.model.v1.ConfigSnapshot - (*ConfigUpdate)(nil), // 24: datadog.model.v1.ConfigUpdate - (*ConfigEvent)(nil), // 25: datadog.model.v1.ConfigEvent - (*GenerateContainerIDFromOriginInfoRequest_LocalData)(nil), // 26: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.LocalData - (*GenerateContainerIDFromOriginInfoRequest_ExternalData)(nil), // 27: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.ExternalData - nil, // 28: datadog.model.v1.TaggerState.StateEntry - nil, // 29: datadog.model.v1.TaggerState.PidMapEntry - (*structpb.Value)(nil), // 30: google.protobuf.Value + (EventType)(0), // 0: datadog.model.v1.EventType + (TagCardinality)(0), // 1: datadog.model.v1.TagCardinality + (*HostnameRequest)(nil), // 2: datadog.model.v1.HostnameRequest + (*HostnameReply)(nil), // 3: datadog.model.v1.HostnameReply + (*HostTagRequest)(nil), // 4: datadog.model.v1.HostTagRequest + (*HostTagReply)(nil), // 5: datadog.model.v1.HostTagReply + (*CaptureTriggerRequest)(nil), // 6: datadog.model.v1.CaptureTriggerRequest + (*CaptureTriggerResponse)(nil), // 7: datadog.model.v1.CaptureTriggerResponse + (*StreamTagsRequest)(nil), // 8: datadog.model.v1.StreamTagsRequest + (*StreamTagsResponse)(nil), // 9: datadog.model.v1.StreamTagsResponse + (*StreamTagsEvent)(nil), // 10: datadog.model.v1.StreamTagsEvent + (*DeprecatedFilter)(nil), // 11: datadog.model.v1.DeprecatedFilter + (*Entity)(nil), // 12: datadog.model.v1.Entity + (*GenerateContainerIDFromOriginInfoRequest)(nil), // 13: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest + (*GenerateContainerIDFromOriginInfoResponse)(nil), // 14: datadog.model.v1.GenerateContainerIDFromOriginInfoResponse + (*FetchEntityRequest)(nil), // 15: datadog.model.v1.FetchEntityRequest + (*FetchEntityResponse)(nil), // 16: datadog.model.v1.FetchEntityResponse + (*EntityId)(nil), // 17: datadog.model.v1.EntityId + (*UnixDogstatsdMsg)(nil), // 18: datadog.model.v1.UnixDogstatsdMsg + (*TaggerState)(nil), // 19: datadog.model.v1.TaggerState + (*TaggerStateResponse)(nil), // 20: datadog.model.v1.TaggerStateResponse + (*ConfigStreamRequest)(nil), // 21: datadog.model.v1.ConfigStreamRequest + (*ConfigSetting)(nil), // 22: datadog.model.v1.ConfigSetting + (*ConfigSnapshot)(nil), // 23: datadog.model.v1.ConfigSnapshot + (*ConfigUpdate)(nil), // 24: datadog.model.v1.ConfigUpdate + (*ConfigUnset)(nil), // 25: datadog.model.v1.ConfigUnset + (*ConfigEvent)(nil), // 26: datadog.model.v1.ConfigEvent + (*GenerateContainerIDFromOriginInfoRequest_LocalData)(nil), // 27: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.LocalData + (*GenerateContainerIDFromOriginInfoRequest_ExternalData)(nil), // 28: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.ExternalData + nil, // 29: datadog.model.v1.TaggerState.StateEntry + nil, // 30: datadog.model.v1.TaggerState.PidMapEntry + (*structpb.Value)(nil), // 31: google.protobuf.Value } var file_datadog_model_v1_model_proto_depIdxs = []int32{ 1, // 0: datadog.model.v1.StreamTagsRequest.cardinality:type_name -> datadog.model.v1.TagCardinality @@ -1789,25 +1896,27 @@ var file_datadog_model_v1_model_proto_depIdxs = []int32{ 0, // 4: datadog.model.v1.StreamTagsEvent.type:type_name -> datadog.model.v1.EventType 12, // 5: datadog.model.v1.StreamTagsEvent.entity:type_name -> datadog.model.v1.Entity 17, // 6: datadog.model.v1.Entity.id:type_name -> datadog.model.v1.EntityId - 26, // 7: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.localData:type_name -> datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.LocalData - 27, // 8: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.externalData:type_name -> datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.ExternalData + 27, // 7: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.localData:type_name -> datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.LocalData + 28, // 8: datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.externalData:type_name -> datadog.model.v1.GenerateContainerIDFromOriginInfoRequest.ExternalData 17, // 9: datadog.model.v1.FetchEntityRequest.id:type_name -> datadog.model.v1.EntityId 1, // 10: datadog.model.v1.FetchEntityRequest.cardinality:type_name -> datadog.model.v1.TagCardinality 17, // 11: datadog.model.v1.FetchEntityResponse.id:type_name -> datadog.model.v1.EntityId 1, // 12: datadog.model.v1.FetchEntityResponse.cardinality:type_name -> datadog.model.v1.TagCardinality - 28, // 13: datadog.model.v1.TaggerState.state:type_name -> datadog.model.v1.TaggerState.StateEntry - 29, // 14: datadog.model.v1.TaggerState.pidMap:type_name -> datadog.model.v1.TaggerState.PidMapEntry - 30, // 15: datadog.model.v1.ConfigSetting.value:type_name -> google.protobuf.Value + 29, // 13: datadog.model.v1.TaggerState.state:type_name -> datadog.model.v1.TaggerState.StateEntry + 30, // 14: datadog.model.v1.TaggerState.pidMap:type_name -> datadog.model.v1.TaggerState.PidMapEntry + 31, // 15: datadog.model.v1.ConfigSetting.value:type_name -> google.protobuf.Value 22, // 16: datadog.model.v1.ConfigSnapshot.settings:type_name -> datadog.model.v1.ConfigSetting 22, // 17: datadog.model.v1.ConfigUpdate.setting:type_name -> datadog.model.v1.ConfigSetting - 23, // 18: datadog.model.v1.ConfigEvent.snapshot:type_name -> datadog.model.v1.ConfigSnapshot - 24, // 19: datadog.model.v1.ConfigEvent.update:type_name -> datadog.model.v1.ConfigUpdate - 12, // 20: datadog.model.v1.TaggerState.StateEntry.value:type_name -> datadog.model.v1.Entity - 21, // [21:21] is the sub-list for method output_type - 21, // [21:21] is the sub-list for method input_type - 21, // [21:21] is the sub-list for extension type_name - 21, // [21:21] is the sub-list for extension extendee - 0, // [0:21] is the sub-list for field type_name + 22, // 18: datadog.model.v1.ConfigUnset.resolved:type_name -> datadog.model.v1.ConfigSetting + 23, // 19: datadog.model.v1.ConfigEvent.snapshot:type_name -> datadog.model.v1.ConfigSnapshot + 24, // 20: datadog.model.v1.ConfigEvent.update:type_name -> datadog.model.v1.ConfigUpdate + 25, // 21: datadog.model.v1.ConfigEvent.unset:type_name -> datadog.model.v1.ConfigUnset + 12, // 22: datadog.model.v1.TaggerState.StateEntry.value:type_name -> datadog.model.v1.Entity + 23, // [23:23] is the sub-list for method output_type + 23, // [23:23] is the sub-list for method input_type + 23, // [23:23] is the sub-list for extension type_name + 23, // [23:23] is the sub-list for extension extendee + 0, // [0:23] is the sub-list for field type_name } func init() { file_datadog_model_v1_model_proto_init() } @@ -1816,19 +1925,20 @@ func file_datadog_model_v1_model_proto_init() { return } file_datadog_model_v1_model_proto_msgTypes[11].OneofWrappers = []any{} - file_datadog_model_v1_model_proto_msgTypes[23].OneofWrappers = []any{ + file_datadog_model_v1_model_proto_msgTypes[24].OneofWrappers = []any{ (*ConfigEvent_Snapshot)(nil), (*ConfigEvent_Update)(nil), + (*ConfigEvent_Unset)(nil), } - file_datadog_model_v1_model_proto_msgTypes[24].OneofWrappers = []any{} file_datadog_model_v1_model_proto_msgTypes[25].OneofWrappers = []any{} + file_datadog_model_v1_model_proto_msgTypes[26].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_datadog_model_v1_model_proto_rawDesc), len(file_datadog_model_v1_model_proto_rawDesc)), NumEnums: 2, - NumMessages: 28, + NumMessages: 29, NumExtensions: 0, NumServices: 0, }, From 8bfd1c6b5f29e3b3a0f7e3fcf0cace73699eb27e Mon Sep 17 00:00:00 2001 From: "rahul.kaukuntla" Date: Fri, 28 Aug 2026 14:03:24 -0400 Subject: [PATCH 2/4] config: notify Set receivers with the resolved value, not the written one Set compared previousValue against its own newValue argument and passed that argument to receivers. A write to a layer that loses the merge therefore notified with a value that Get never returns, and named the layer written to rather than the one the setting resolves from. Read back from the merged tree instead, so a losing write notifies nobody and a winning one reports what the setting now is. The write is still recorded, and surfaces correctly if the layer above it is later cleared. --- pkg/config/nodetreemodel/config.go | 10 +++++++-- pkg/config/nodetreemodel/config_test.go | 28 +++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/pkg/config/nodetreemodel/config.go b/pkg/config/nodetreemodel/config.go index 46a20b01996d..35a9bb5ea680 100644 --- a/pkg/config/nodetreemodel/config.go +++ b/pkg/config/nodetreemodel/config.go @@ -295,8 +295,14 @@ func (c *ntmConfig) Set(key string, newValue interface{}, source model.Source) { receivers := slices.Clone(c.notificationReceivers) + // Read back rather than trusting newValue: a write to a layer that loses the merge changes + // nothing that resolves, and receivers are told what the setting is, not what was stored. + resolved := c.leafAtPathFromNode(key, c.root) + resolvedValue := resolved.Get() + resolvedSource := resolved.Source() + // if no value has changed we don't notify - if reflect.DeepEqual(previousValue, newValue) { + if reflect.DeepEqual(previousValue, resolvedValue) { c.Unlock() return } @@ -309,7 +315,7 @@ func (c *ntmConfig) Set(key string, newValue interface{}, source model.Source) { // notifying all receiver about the updated setting for _, receiver := range receivers { - receiver(key, source, previousValue, newValue, sequenceID) + receiver(key, resolvedSource, previousValue, resolvedValue, sequenceID) } } diff --git a/pkg/config/nodetreemodel/config_test.go b/pkg/config/nodetreemodel/config_test.go index 6e8651a4aec0..51393f30d283 100644 --- a/pkg/config/nodetreemodel/config_test.go +++ b/pkg/config/nodetreemodel/config_test.go @@ -2530,3 +2530,31 @@ func TestOnUnsetReportsEveryLayerRemoval(t *testing.T) { assert.Empty(t, *updates) }) } + +func TestSetNotifiesOnlyWhenTheResolvedValueChanges(t *testing.T) { + type notification struct { + source model.Source + value any + } + + cfg := NewNodeTreeConfig("test", "TEST", nil) + cfg.SetDefault("shadowed", "default") + cfg.BuildSchema() + cfg.Set("shadowed", "from_cli", model.SourceCLI) + + var got []notification + cfg.OnUpdate(func(_ string, source model.Source, _, newValue any, _ uint64) { + got = append(got, notification{source, newValue}) + }) + + // CLI outranks file, so nothing a receiver can observe has changed. + cfg.Set("shadowed", "from_file", model.SourceFile) + assert.Empty(t, got, "a write that loses the merge is not a change") + assert.Equal(t, "from_cli", cfg.Get("shadowed")) + + // The write was still recorded, so clearing CLI surfaces it, named by the layer it came from. + cfg.UnsetForSource("shadowed", model.SourceCLI) + require.Len(t, got, 1) + assert.Equal(t, model.SourceFile, got[0].source) + assert.Equal(t, "from_file", got[0].value) +} From 9d5f26bea6d3018ee004242fec3cdbce3939360f Mon Sep 17 00:00:00 2001 From: "rahul.kaukuntla" Date: Fri, 28 Aug 2026 14:25:24 -0400 Subject: [PATCH 3/4] configstream: drop the unset when its fallback value cannot be encoded Emitting the unset without Resolved after a failed encode reads as "nothing remains" on the subscriber, which drops the key and advances its sequence ID -- diverging permanently with nothing left to signal it. Drop the event instead, matching what the update path already does. The consumed sequence ID leaves a gap, and handleConfigUpdate resynchronizes the affected subscriber with a fresh snapshot. --- comp/core/configstream/impl/configstream.go | 11 ++-- .../configstream/impl/configstream_test.go | 52 +++++++++++++++++++ 2 files changed, 59 insertions(+), 4 deletions(-) diff --git a/comp/core/configstream/impl/configstream.go b/comp/core/configstream/impl/configstream.go index 48a233c8ceab..e81c11025e51 100644 --- a/comp/core/configstream/impl/configstream.go +++ b/comp/core/configstream/impl/configstream.go @@ -160,11 +160,14 @@ func (cs *configStream) run() { // Left unset when the key resolves to nothing, which tells the subscriber to only drop it. if resolvedSource != model.SourceUnknown { - if resolved, err := newConfigSetting(setting, resolvedValue, resolvedSource); err != nil { - cs.log.Errorf("Failed to encode post-unset value of '%s': %v", setting, err) - } else { - unset.Resolved = resolved + resolved, err := newConfigSetting(setting, resolvedValue, resolvedSource) + if err != nil { + // Sending it anyway would read as "nothing remains" and diverge the subscriber for + // good. Dropping the event leaves a sequence gap, which makes it resync instead. + cs.log.Errorf("Failed to encode post-unset value of '%s', dropping unset to force a resync: %v", setting, err) + return } + unset.Resolved = resolved } configUnset := &pb.ConfigEvent{Event: &pb.ConfigEvent_Unset{Unset: unset}} diff --git a/comp/core/configstream/impl/configstream_test.go b/comp/core/configstream/impl/configstream_test.go index a52aaa445b7b..b3bfbd5d2867 100644 --- a/comp/core/configstream/impl/configstream_test.go +++ b/comp/core/configstream/impl/configstream_test.go @@ -7,6 +7,7 @@ package configstreamimpl import ( "context" + "math" "testing" "testing/synctest" "time" @@ -365,6 +366,7 @@ func buildComponent(t *testing.T) (Provides, *configInterceptor) { cfg.BindEnvAndSetDefault("my.new.setting", "") cfg.BindEnvAndSetDefault("dropped.setting", "") cfg.BindEnvAndSetDefault("another.setting", 0) + cfg.BindEnvAndSetDefault("complex.setting", map[string]interface{}{}) cfg.BindEnvAndSetDefault("logs_config.auto_multi_line_detection", true) cfg.BindEnvAndSetDefault("logs_config.use_compression", false) @@ -484,6 +486,56 @@ func TestConfigStream(t *testing.T) { require.Equal(t, "original_value", unset.Unset.Resolved.Value.GetStringValue()) require.Equal(t, string(model.SourceAgentRuntime), unset.Unset.Resolved.Source) }) + t.Run("drops the unset when the fallback value cannot be encoded", func(t *testing.T) { + provides, configComp := buildComponent(t) + + eventsCh, unsubscribe := provides.Comp.Subscribe(&pb.ConfigStreamRequest{Name: "test-client-unencodable"}) + defer unsubscribe() + + var event *pb.ConfigEvent + select { + case event = <-eventsCh: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting initial snapshot") + } + _, isSnapshot := event.GetEvent().(*pb.ConfigEvent_Snapshot) + require.True(t, isSnapshot, "first event must be snapshot") + + configComp.Set("complex.setting", map[string]interface{}{"n": 1.0}, model.SourceCLI) + select { + case event = <-eventsCh: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting config update") + } + update, isUpdate := event.GetEvent().(*pb.ConfigEvent_Update) + require.True(t, isUpdate) + require.Equal(t, "complex.setting", update.Update.Setting.Key) + + // Scalar keys are coerced to their declared type, so an unencodable value only survives under + // a complex one. This write is shadowed by CLI, so it notifies nobody and stays off the wire. + configComp.Set("complex.setting", map[string]interface{}{"n": math.Inf(1)}, model.SourceAgentRuntime) + + // Clearing CLI falls back to the +Inf map, which has no JSON representation. An unset without + // Resolved would read as "nothing remains" and diverge the subscriber for good, so nothing is + // sent at all. + configComp.UnsetForSource("complex.setting", model.SourceCLI) + select { + case event = <-eventsCh: + t.Fatalf("expected no event, got %v", event.GetEvent()) + case <-time.After(500 * time.Millisecond): + } + + // The dropped event still consumed a sequence ID, so the next change lands out of order and + // the subscriber is resynchronized with a snapshot instead of carrying a stale value. + configComp.Set("complex.setting", map[string]interface{}{"n": 2.0}, model.SourceCLI) + select { + case event = <-eventsCh: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting resynchronization snapshot") + } + _, isSnapshot = event.GetEvent().(*pb.ConfigEvent_Snapshot) + require.True(t, isSnapshot, "sequence gap must resynchronize the subscriber") + }) resyncsWithSnapshotOnDiscontinuity := func(t *testing.T) { provides, configComp := buildComponent(t) From 7d3ffecc1447233d66049cd7890df807a144713b Mon Sep 17 00:00:00 2001 From: "rahul.kaukuntla" Date: Fri, 28 Aug 2026 14:51:11 -0400 Subject: [PATCH 4/4] configstream: document ConfigUnset and handle it in the other stream clients The README described a ConfigEvent with only snapshots and updates. Add the new arm along with the removal semantics: why the fallback value has to travel with the removal, and why an unrecognized event has to resynchronize rather than be skipped. config-stream-client and internal/remote-agent switch on the event kind with no default case, so both silently swallowed unsets -- the former being the tool reached for to debug exactly this. Both now report them, and the debug client reports unknown kinds instead of hiding them. --- cmd/config-stream-client/main.go | 19 +++++++++++++++++++ comp/core/configstream/README.md | 13 +++++++++++++ internal/remote-agent/main.go | 2 ++ 3 files changed, 34 insertions(+) diff --git a/cmd/config-stream-client/main.go b/cmd/config-stream-client/main.go index a78173dc2f28..cd767be08415 100644 --- a/cmd/config-stream-client/main.go +++ b/cmd/config-stream-client/main.go @@ -203,6 +203,25 @@ func main() { fmt.Printf(" Value: %v\n", formatValue(setting.Value)) fmt.Printf(" Source: %s\n", setting.Source) fmt.Println() + + case *pb.ConfigEvent_Unset: + currentSeqID := e.Unset.SequenceId + fmt.Printf("UNSET received (seq_id=%d)\n", currentSeqID) + if snapshotReceived && currentSeqID > maxSeqID { + maxSeqID = currentSeqID + } + fmt.Printf(" Key: %s\n", e.Unset.Key) + fmt.Printf(" Cleared source: %s\n", e.Unset.Source) + if resolved := e.Unset.GetResolved(); resolved != nil { + fmt.Printf(" Now resolves to: %v (source: %s)\n", formatValue(resolved.Value), resolved.Source) + } else { + fmt.Printf(" Now resolves to: nothing\n") + } + fmt.Println() + + default: + // Resynchronizing beats ignoring it: skipping an event diverges from the sender. + fmt.Printf("Unknown event type %T, a newer core agent may be sending events this client cannot read\n", event.Event) } } diff --git a/comp/core/configstream/README.md b/comp/core/configstream/README.md index 2546750caf56..c0fa0e5f93fd 100644 --- a/comp/core/configstream/README.md +++ b/comp/core/configstream/README.md @@ -67,6 +67,7 @@ message ConfigEvent { oneof event { ConfigSnapshot snapshot = 1; // Sent first, then on resync ConfigUpdate update = 2; // Incremental changes + ConfigUnset unset = 3; // A source layer was cleared } } @@ -82,6 +83,14 @@ message ConfigUpdate { ConfigSetting setting = 3; // Single changed setting } +message ConfigUnset { + string origin = 1; // Config file (e.g., "datadog.yaml") + int32 sequence_id = 2; // Monotonic sequence ID, shared with updates + string key = 3; // Setting name + string source = 4; // The layer cleared, not the one fallen back to + ConfigSetting resolved = 5; // What the key resolves to now; unset if nothing remains +} + message ConfigSetting { string source = 1; // "file", "env-var", "remote-config", etc. string key = 2; // Setting name @@ -89,6 +98,10 @@ message ConfigSetting { } ``` +**Removals:** snapshots carry only the merged view, one entry per key tagged with the winning source, so a subscriber has no lower layer of its own to fall back to. `ConfigUnset` therefore names the layer that was cleared and carries the value the key resolves to without it, which the subscriber writes before dropping the cleared entry. `resolved` is absent when nothing remains, and an unset shares its sequence ID with the update it replaces. + +**Unrecognized events** must trigger a resynchronization, not be ignored: skipping an event silently diverges from the sender, whereas erroring out of the stream reconnects and receives a fresh snapshot. + ## Configuration The config stream component always runs. Individual connections are RAR-gated: the caller must be a registered remote agent. diff --git a/internal/remote-agent/main.go b/internal/remote-agent/main.go index b92787b49370..09b62c559c7c 100644 --- a/internal/remote-agent/main.go +++ b/internal/remote-agent/main.go @@ -211,6 +211,8 @@ func streamConfigEvents(ctx context.Context, agentIpcAddress, agentAuthToken, ag log.Printf("Config snapshot received: %d settings (seq=%d)", len(e.Snapshot.GetSettings()), e.Snapshot.GetSequenceId()) case *pbcore.ConfigEvent_Update: log.Printf("Config update received: seq=%d", e.Update.GetSequenceId()) + case *pbcore.ConfigEvent_Unset: + log.Printf("Config unset received: seq=%d key=%s cleared_source=%s", e.Unset.GetSequenceId(), e.Unset.GetKey(), e.Unset.GetSource()) } } }