configstream: let clients ingest streamed env-var-sourced settings (#55310)

comp/core/configstreamconsumer/impl/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "//comp/def",
         "//pkg/api/security",
         "//pkg/api/security/cert",
+        "//pkg/config/model",
         "//pkg/configstreambootstrap",
         "//pkg/proto/pbgo/core",
         "//pkg/util/defaultpaths",
@@ -26,6 +27,7 @@ go_library(
         "@in_yaml_go_yaml_v3//:yaml",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//metadata",
+        "@org_golang_google_protobuf//types/known/structpb",
     ],
 )
 

comp/core/configstreamconsumer/impl/consumer.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cenkalti/backoff/v7"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/protobuf/types/known/structpb"
 
 	configstreamconsumer "github.com/DataDog/datadog-agent/comp/core/configstreamconsumer/def"
 	log "github.com/DataDog/datadog-agent/comp/core/log/def"
@@ -34,6 +35,7 @@ import (
 	compdef "github.com/DataDog/datadog-agent/comp/def"
 	pkgtoken "github.com/DataDog/datadog-agent/pkg/api/security"
 	"github.com/DataDog/datadog-agent/pkg/api/security/cert"
+	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"
 	"github.com/DataDog/datadog-agent/pkg/util/flavor"
@@ -331,9 +333,11 @@ func (c *consumer) applySnapshot(snapshot *pb.ConfigSnapshot) error {
 
 	c.log.Infof("Applying config snapshot (seq_id: %d, settings: %d)", snapshot.SequenceId, len(snapshot.Settings))
 
+	settings := make([]pkgconfigmodel.DirectSetting, 0, len(snapshot.Settings))
 	for _, setting := range snapshot.Settings {
-		configstreambootstrap.ApplySetting(setting.Key, setting.Value, setting.Source)
+		settings = append(settings, toDirectSetting(setting))
 	}
+	configstreambootstrap.Config().DirectBulkSet(settings)
 	c.lastSeqID.Store(snapshot.SequenceId)
 	c.lastSeqIDMetric.Set(float64(snapshot.SequenceId))
 
@@ -348,6 +352,24 @@ func (c *consumer) applySnapshot(snapshot *pb.ConfigSnapshot) error {
 	return nil
 }
 
+// toDirectSetting decodes a streamed setting into the form the config builder takes.
+func toDirectSetting(setting *pb.ConfigSetting) pkgconfigmodel.DirectSetting {
+	return pkgconfigmodel.DirectSetting{
+		Key:    setting.Key,
+		Value:  pbValueToGo(setting.Value),
+		Source: pkgconfigmodel.Source(setting.Source),
+	}
+}
+
+// pbValueToGo converts a protobuf Value to a Go value. structpb has no integer type, so numbers
+// arrive as float64; narrowing is left to the declared default type.
+func pbValueToGo(v *structpb.Value) any {
+	if v == nil {
+		return nil
+	}
+	return v.AsInterface()
+}
+
 func (c *consumer) applyUpdate(update *pb.ConfigUpdate) error {
 	if update.SequenceId <= c.lastSeqID.Load() {
 		c.log.Warnf("Ignoring stale update (seq_id: %d <= %d)", update.SequenceId, c.lastSeqID.Load())
@@ -361,7 +383,10 @@ func (c *consumer) applyUpdate(update *pb.ConfigUpdate) error {
 
 	c.log.Debugf("Applying config update (seq_id: %d, key: %s)", update.SequenceId, update.Setting.Key)
 
-	configstreambootstrap.ApplySetting(update.Setting.Key, update.Setting.Value, update.Setting.Source)
+	setting := toDirectSetting(update.Setting)
+	// Updates never carry env-var-sourced settings, so Set's guardrail is not in the way and
+	// registered receivers still get notified.
+	configstreambootstrap.Config().Set(setting.Key, setting.Value, setting.Source)
 	c.lastSeqID.Store(update.SequenceId)
 	c.lastSeqIDMetric.Set(float64(update.SequenceId))
 

comp/core/configstreamconsumer/impl/consumer_test.go

@@ -13,6 +13,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/DataDog/datadog-agent/pkg/configstreambootstrap"
 )
@@ -128,3 +129,12 @@ remote_agent:
 		require.Equal(t, "vsock:2:5001", got.VSockAddr)
 	})
 }
+
+func TestPbValueToGoKeepsWireType(t *testing.T) {
+	// Narrowing is the declared default type's job, not this function's.
+	require.Nil(t, pbValueToGo(nil))
+	require.Equal(t, float64(5), pbValueToGo(structpb.NewNumberValue(5)))
+	require.Equal(t, float64(5.5), pbValueToGo(structpb.NewNumberValue(5.5)))
+	require.Equal(t, "s", pbValueToGo(structpb.NewStringValue("s")))
+	require.Equal(t, true, pbValueToGo(structpb.NewBoolValue(true)))
+}

pkg/config/model/types.go

@@ -94,6 +94,13 @@ var sourcesPriority = map[Source]int{
 	SourceCLI:                11,
 }
 
+// DirectSetting is one key/value/source assignment for nodetreemodel's DirectBulkSet.
+type DirectSetting struct {
+	Key    string
+	Value  interface{}
+	Source Source
+}
+
 // ValueWithSource is a tuple for a source and a value, not necessarily the applied value in the main config
 type ValueWithSource struct {
 	Source Source
@@ -223,6 +230,11 @@ type Writer interface {
 	Set(key string, value interface{}, source Source)
 	SetInTest(key string, value interface{})
 	UnsetForSource(key string, source Source)
+	// DirectBulkSet writes settings already resolved by another config, keeping each one in the
+	// source layer it came from so the result mirrors the sender. It exists for config streaming
+	// and nothing else should call it: unlike Set it accepts SourceEnvVar and skips notifications,
+	// which makes it unfit for applying a live change.
+	DirectBulkSet(settings []DirectSetting)
 }
 
 // ReaderWriter is a subset of Config that allows reading and writing the configuration

pkg/config/nodetreemodel/config.go

@@ -221,7 +221,6 @@ func (c *ntmConfig) Set(key string, newValue interface{}, source model.Source) {
 	if source == model.SourceEnvVar {
 		panicInTest("Writing to env var layers is not allowed, use SourceAgentRuntime instead.")
 	}
-
 	c.maybeRebuild()
 
 	c.Lock()
@@ -308,6 +307,47 @@ func (c *ntmConfig) insertValueIntoTree(key string, value interface{}, source mo
 	return tree, err
 }
 
+// DirectBulkSet implements model.Writer. Keys are assumed already lowercased, which holds for
+// anything enumerated from another config.
+func (c *ntmConfig) DirectBulkSet(settings []model.DirectSetting) {
+	c.Lock()
+	defer c.Unlock()
+
+	for _, setting := range settings {
+		key := setting.Key
+		// Stored anyway, as the YAML loader does, so the client mirrors the sender. Reconnects
+		// resend the whole snapshot, hence warn once.
+		if !c.isKnownKey(key) {
+			if _, alreadySeen := c.unknownKeys.LoadOrStore(key, struct{}{}); !alreadySeen {
+				log.Warnf("unknown key from config stream: %s", key)
+			}
+		}
+
+		declaredNode := c.nodeAtPathFromNode(key, c.defaults)
+		if declaredNode.IsInnerNode() {
+			log.Errorf("could not set '%s': partial path of a setting", key)
+			continue
+		}
+
+		// structpb collapses every number to float64, so a value still needs coercing back to the
+		// declared type. Unknown keys have no default, for which this is a no-op.
+		value := setting.Value
+		if converted, err := basic.ConvertToDefaultType(value, declaredNode.Get(), false); err == nil {
+			value = converted
+		}
+
+		if _, err := c.insertValueIntoTree(key, value, setting.Source); err != nil {
+			log.Errorf("could not insert value for '%s': %s", key, err)
+		}
+	}
+
+	// Set merges per write; rebuilding the root once at the end is equivalent because Merge
+	// ranks conflicting leaves by source, not by merge order.
+	if err := c.mergeAllLayers(); err != nil {
+		log.Errorf("could not merge config layers: %s", err)
+	}
+}
+
 // SetInTest assigns the value to the given key using source Unknown, may only be called from tests
 func (c *ntmConfig) SetInTest(key string, value interface{}) {
 	c.assertIsTest("SetInTest")

pkg/config/nodetreemodel/config_test.go

@@ -2068,3 +2068,48 @@ func BenchmarkMaybeRebuildUnchangedEnv(b *testing.B) {
 		cfg.Get("key")
 	}
 }
+
+func TestDirectBulkSet(t *testing.T) {
+	cfg := NewNodeTreeConfig("test", "TEST", nil)
+	cfg.SetDefault("from_env", 0)
+	cfg.SetDefault("from_file", 0)
+	cfg.SetDefault("coerced", "")
+	cfg.SetDefault("outranked", 0)
+	cfg.SetDefault("a_float", 0.0)
+	cfg.BuildSchema()
+
+	cfg.Set("outranked", 9, model.SourceAgentRuntime)
+
+	var notified int
+	cfg.OnUpdate(func(_ string, _ model.Source, _, _ any, _ uint64) { notified++ })
+
+	// Set would reject the env var layer outright.
+	cfg.DirectBulkSet([]model.DirectSetting{
+		{Key: "from_env", Value: 1, Source: model.SourceEnvVar},
+		{Key: "from_file", Value: 2, Source: model.SourceFile},
+		{Key: "coerced", Value: 3, Source: model.SourceEnvVar},
+		{Key: "outranked", Value: 4, Source: model.SourceFile},
+		{Key: "undeclared", Value: 5, Source: model.SourceEnvVar},
+		{Key: "a_float", Value: float64(5), Source: model.SourceEnvVar},
+	})
+
+	assert.Equal(t, 1, cfg.Get("from_env"))
+	assert.Equal(t, model.SourceEnvVar, cfg.GetSource("from_env"))
+	assert.Equal(t, 2, cfg.Get("from_file"))
+	assert.Equal(t, model.SourceFile, cfg.GetSource("from_file"))
+
+	assert.Equal(t, "3", cfg.Get("coerced"), "values are coerced to the declared type, as Set does")
+
+	// A lower-priority layer written in bulk must not overtake a higher-priority one.
+	assert.Equal(t, 9, cfg.Get("outranked"))
+	assert.Equal(t, model.SourceAgentRuntime, cfg.GetSource("outranked"))
+
+	// A key absent from this process's schema is still stored, so the config mirrors the sender.
+	assert.Equal(t, 5, cfg.Get("undeclared"))
+
+	// An integral float64 must not collapse to int.
+	assert.Equal(t, float64(5), cfg.Get("a_float"))
+
+	assert.Zero(t, notified, "notifications should not fire")
+}
+

pkg/configstreambootstrap/BUILD.bazel

@@ -16,7 +16,6 @@ go_library(
         "//pkg/config/model",
         "//pkg/config/setup",
         "//pkg/util/log",
-        "@org_golang_google_protobuf//types/known/structpb",
     ],
 )
 

pkg/configstreambootstrap/bootstrap.go

@@ -9,10 +9,6 @@
 package configstreambootstrap
 
 import (
-	"math"
-
-	"google.golang.org/protobuf/types/known/structpb"
-
 	pkgtoken "github.com/DataDog/datadog-agent/pkg/api/security"
 	"github.com/DataDog/datadog-agent/pkg/api/security/cert"
 	"github.com/DataDog/datadog-agent/pkg/config/create"
@@ -96,23 +92,7 @@ func IPCCertFilepath() string {
 	return pkgconfigsetup.Datadog().GetString("ipc_cert_file_path")
 }
 
-// ApplySetting writes one streamed setting to the global config, preserving the source.
-func ApplySetting(key string, value *structpb.Value, source string) {
-	pkgconfigsetup.Datadog().Set(key, pbValueToGo(value), pkgconfigmodel.Source(source))
-}
-
-// pbValueToGo converts a protobuf Value to a Go value. It preserves integer types that structpb widens to float64.
-// Bounded to |x| <= 2^53 — beyond that float64 loses integer precision.
-func pbValueToGo(v *structpb.Value) any {
-	if v == nil {
-		return nil
-	}
-	result := v.AsInterface()
-	if f, ok := result.(float64); ok {
-		const maxExactInt = 1 << 53
-		if !math.IsNaN(f) && !math.IsInf(f, 0) && f >= -maxExactInt && f <= maxExactInt && f == math.Trunc(f) {
-			return int64(f)
-		}
-	}
-	return result
+// Config returns the global config builder the streamed settings are written to.
+func Config() pkgconfigmodel.Config {
+	return pkgconfigsetup.Datadog()
 }