configstream: add a ConfigUnset event so layer removals cross the wire

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)
 		}
 	}
 

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,13 +83,25 @@ 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
   google.protobuf.Value value = 3; // Typed value (string, int, bool, etc.)
 }
 ```
 
+**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.

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,47 @@ 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 {
+			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}}
+
+		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 +277,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 +306,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.

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)
 
@@ -467,19 +469,72 @@ 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")
+		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)
+	})
+	t.Run("drops the unset when the fallback value cannot be encoded", func(t *testing.T) {
+		provides, configComp := buildComponent(t)
 
-		// 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())
+		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) {

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",

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",