fix(ssi): wait for RC before inject-all

pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies.go

@@ -11,6 +11,7 @@ import (
 	"regexp"
 	"sort"
 	"strconv"
+	"time"
 
 	rcclient "github.com/DataDog/datadog-agent/pkg/config/remote/client"
 	"github.com/DataDog/datadog-agent/pkg/remoteconfig/state"
@@ -63,21 +64,54 @@ func remotePolicyPathOrder(path string) int {
 // format is the dd-wls policies document; targets do not appear on this path.
 func (m *TargetMutator) subscribeRemoteConfig(client *rcclient.Client) {
 	if client == nil {
+		m.allowInjectAll.Store(true)
 		return
 	}
 
 	log.Infof("auto-instrumentation: subscribing to remote config product %q for SSI policies", state.ProductApmPolicies)
-	// Apply the latest snapshot already held by the client, then subscribe for
-	// future updates.
+
+	if client.HasSynced() {
+		m.applySnapshotAndSubscribe(client)
+		return
+	}
+
+	log.Infof("auto-instrumentation: waiting for the first remote config snapshot of %q before applying SSI inject-all", state.ProductApmPolicies)
+	go m.waitForFirstRemoteConfigUpdate(client)
+}
+
+// applySnapshotAndSubscribe applies the current APM_POLICIES snapshot then
+// subscribes for subsequent updates.
+func (m *TargetMutator) applySnapshotAndSubscribe(client *rcclient.Client) {
 	m.onRemoteConfigUpdate(client.GetConfigs(state.ProductApmPolicies), client.UpdateApplyStatus)
 	client.Subscribe(state.ProductApmPolicies, m.onRemoteConfigUpdate)
 }
 
+// waitForFirstRemoteConfigUpdate waits until the RC client has completed a
+// poll, then applies the snapshot and subscribes. This is local to the mutator:
+// the RC client must not fan that signal out to other product listeners.
+func (m *TargetMutator) waitForFirstRemoteConfigUpdate(client *rcclient.Client) {
+	ticker := time.NewTicker(200 * time.Millisecond)
+	defer ticker.Stop()
+	for range ticker.C {
+		if client.HasSynced() {
+			m.applySnapshotAndSubscribe(client)
+			return
+		}
+	}
+}
+
+func (m *TargetMutator) enableInjectAll() {
+	if m.allowInjectAll.CompareAndSwap(false, true) {
+		log.Infof("auto-instrumentation: first remote config snapshot for SSI policies received")
+	}
+}
+
 func (m *TargetMutator) onRemoteConfigUpdate(updates map[string]state.RawConfig, applyStateCallback func(string, state.ApplyStatus)) {
 	log.Debugf("auto-instrumentation: remote config update for SSI policies: %d config(s)", len(updates))
 
 	if len(updates) == 0 {
 		m.ClearRemotePolicies()
+		m.enableInjectAll()
 		return
 	}
 
@@ -116,4 +150,5 @@ func (m *TargetMutator) onRemoteConfigUpdate(updates map[string]state.RawConfig,
 	for path := range updates {
 		applyStateCallback(path, state.ApplyStatus{State: state.ApplyStateAcknowledged})
 	}
+	m.enableInjectAll()
 }

pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go

@@ -276,3 +276,77 @@ func TestOnRemoteConfigUpdate_InvalidPayloadKeepsBaseline(t *testing.T) {
 	require.Equal(t, "config-default", name)
 	require.False(t, fromPolicy)
 }
+
+// TestRemotePolicies_InjectAllWaitsForFirstSnapshot verifies that SSI-on with no
+// static targets does not inject-all until a remote-config round-trip has been
+// observed. An empty snapshot then unblocks inject-all; a policy snapshot
+// evaluates RC instead.
+func TestRemotePolicies_InjectAllWaitsForFirstSnapshot(t *testing.T) {
+	wmeta := newMatchTestWmeta(t)
+	m := newMatchMutator(t, rcSSIOnNoTargets, wmeta)
+	pod := rcPod("ns", map[string]string{"app": "db"})
+
+	name, fromPolicy := matchedTarget(t, m, pod)
+	require.Equal(t, "default", name)
+	require.False(t, fromPolicy)
+
+	m.allowInjectAll.Store(false)
+	require.Nil(t, m.getMatchingTarget(pod))
+
+	m.onRemoteConfigUpdate(map[string]state.RawConfig{}, func(string, state.ApplyStatus) {})
+	require.True(t, m.allowInjectAll.Load())
+	name, fromPolicy = matchedTarget(t, m, pod)
+	require.Equal(t, "default", name)
+	require.False(t, fromPolicy)
+}
+
+func TestRemotePolicies_FirstSnapshotAppliesPolicies(t *testing.T) {
+	wmeta := newMatchTestWmeta(t)
+	m := newMatchMutator(t, rcSSIOnNoTargets, wmeta)
+	pod := rcPod("ns", map[string]string{"app": "db-user"})
+
+	m.allowInjectAll.Store(false)
+	require.Nil(t, m.getMatchingTarget(pod))
+
+	const raw = `{
+      "policies": [{
+        "description": "java for db-user",
+        "rules": {
+          "node_type": "EvaluatorNode",
+          "node": {
+            "eval_type": "StrEvaluator",
+            "eval": {"id": "POD_LABEL", "cmp": "CMP_EXACT", "value": "app=db-user"}
+          }
+        },
+        "actions": [
+          {"action": "INJECT_ALLOW"},
+          {"action": "ENABLE_SDK", "values": ["java=latest"]}
+        ]
+      }]
+    }`
+	m.onRemoteConfigUpdate(map[string]state.RawConfig{
+		"datadog/2/APM_POLICIES/policy-1/config": {Config: []byte(raw)},
+	}, func(string, state.ApplyStatus) {})
+
+	require.True(t, m.allowInjectAll.Load())
+	name, fromPolicy := matchedTarget(t, m, pod)
+	require.Equal(t, "java for db-user", name)
+	require.True(t, fromPolicy)
+	require.Nil(t, m.getMatchingTarget(rcPod("ns", map[string]string{"app": "other"})))
+}
+
+func TestOnRemoteConfigUpdate_InvalidFirstSnapshotStaysPending(t *testing.T) {
+	wmeta := newMatchTestWmeta(t)
+	m := newMatchMutator(t, rcSSIOnNoTargets, wmeta)
+	pod := rcPod("ns", map[string]string{"app": "db"})
+
+	m.allowInjectAll.Store(false)
+	applied := 0
+	m.onRemoteConfigUpdate(map[string]state.RawConfig{
+		"datadog/2/APM_POLICIES/1.bad/config": {Config: []byte("{")},
+	}, func(string, state.ApplyStatus) { applied++ })
+
+	require.Equal(t, 1, applied)
+	require.False(t, m.allowInjectAll.Load())
+	require.Nil(t, m.getMatchingTarget(pod))
+}

pkg/clusteragent/admission/mutate/autoinstrumentation/target_matching_test.go

@@ -124,6 +124,7 @@ apm_config:
 //	SSI | static targets | RC          | Decision
 //	off | —              | none        | nothing
 //	off | —              | policies    | last matching policy, else nothing
+//	on  | none           | awaiting    | nothing (no inject-all before first RC snapshot)
 //	on  | none           | none        | everything
 //	on  | none           | policies    | last matching policy, else nothing
 //	on  | present        | none        | first matching target, else nothing
@@ -210,6 +211,13 @@ apm_config:
 		assertMatch(t, m, "ns", map[string]string{"app": "other"}, helm("default"))
 	})
 
+	t.Run("ssi on / no targets / awaiting first RC snapshot / nothing", func(t *testing.T) {
+		m := newMatchMutator(t, ssiOnNoTargets, newMatchTestWmeta(t))
+		m.allowInjectAll.Store(false)
+		assertMatch(t, m, "ns", map[string]string{"app": "db"}, nothing)
+		assertMatch(t, m, "ns", map[string]string{"app": "other"}, nothing)
+	})
+
 	t.Run("ssi on / no targets / RC / last matching policy, else nothing", func(t *testing.T) {
 		m := newMatchMutator(t, ssiOnNoTargets, newMatchTestWmeta(t))
 		require.NoError(t, m.SetRemotePolicies(rcPolicies))
@@ -239,6 +247,13 @@ apm_config:
 		assertMatch(t, m, "ns", map[string]string{"app": "db"}, nothing)
 	})
 
+	t.Run("ssi on / targets / awaiting first RC snapshot / static still matches", func(t *testing.T) {
+		m := newMatchMutator(t, ssiOnTargets, newMatchTestWmeta(t))
+		m.allowInjectAll.Store(false)
+		assertMatch(t, m, "ns", map[string]string{"language": "python"}, helm("helm-python"))
+		assertMatch(t, m, "ns", map[string]string{"app": "db"}, nothing)
+	})
+
 	t.Run("ssi on / targets / RC / first matching target, else last matching policy, else nothing", func(t *testing.T) {
 		m := newMatchMutator(t, ssiOnTargets, newMatchTestWmeta(t))
 		require.NoError(t, m.SetRemotePolicies(rcPolicies))

pkg/clusteragent/admission/mutate/autoinstrumentation/target_mutator.go

@@ -18,6 +18,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/dynamic"
 
+	"github.com/DataDog/dd-policy-engine/go/policies"
+
 	"github.com/DataDog/datadog-agent/comp/core/workloadmeta/collectors/util"
 	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/common"
@@ -28,7 +30,6 @@ import (
 	mutatecommon "github.com/DataDog/datadog-agent/pkg/clusteragent/admission/mutate/common"
 	rcclient "github.com/DataDog/datadog-agent/pkg/config/remote/client"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
-	"github.com/DataDog/dd-policy-engine/go/policies"
 )
 
 const (
@@ -61,10 +62,13 @@ type TargetMutator struct {
 	// as a namespace target (Helm, Operator, or datadog.yaml). Empty when SSI
 	// is off or when SSI is on with no targeting.
 	staticPolicies policySet
-	// injectAll is the SSI-on fallback when there is no static targeting and no RC.
-	injectAll *targetInternal
 	// remotePolicies is the current RC policy set. Nil when none are installed.
 	remotePolicies atomic.Pointer[policySet]
+	// allowInjectAll gates the SSI-on fallback when there is no static
+	// targeting and no remote policies. The zero value is false (fail-closed).
+	allowInjectAll atomic.Bool
+	// injectAll is the SSI-on fallback when there is no static targeting and no RC.
+	injectAll *targetInternal
 }
 
 // NewTargetMutator creates a new mutator for target based workload selection. We convert the targets to a more
@@ -106,7 +110,8 @@ func NewTargetMutator(config *Config, wmeta workloadmeta.Component, imageResolve
 		ssiEnabled:                    ssiEnabled,
 		staticPolicies:                staticPolicies,
 	}
-	// SSI on and no static targeting: prepare inject-all. Applied only when RC is also absent.
+	// SSI on and no static targeting: prepare inject-all. Applied only when RC
+	// policies are also absents (after sync is complete).
 	if ssiEnabled && len(targets) == 0 {
 		fallback, err := buildInternalTargets(config, []Target{createDefaultTarget(nil, config.Instrumentation.LibVersions)}, defaultLibVersions)
 		if err != nil {
@@ -120,9 +125,11 @@ func NewTargetMutator(config *Config, wmeta workloadmeta.Component, imageResolve
 
 	// On-demand instrumentation is the local gate for remote-config SSI
 	// policies. subscribeRemoteConfig is a no-op when rcClient is nil (e.g. in
-	// tests or when remote config is disabled).
+	// tests or when remote config is disabled) and enables inject-all immediately.
 	if config.Instrumentation.OnDemand {
 		m.subscribeRemoteConfig(rcClient)
+	} else {
+		m.allowInjectAll.Store(true)
 	}
 
 	return m, nil
@@ -220,7 +227,8 @@ func (m *TargetMutator) SetRemotePolicies(ps []policies.Policy) error {
 }
 
 // ClearRemotePolicies drops remote-config policies. Matching falls back to
-// static targets, then the SSI inject-all default if there is no static targeting.
+// static targets, then the SSI inject-all default if there is no static
+// targeting and inject-all is allowed.
 func (m *TargetMutator) ClearRemotePolicies() {
 	m.remotePolicies.Store(nil)
 }
@@ -480,7 +488,8 @@ func (m *TargetMutator) getTargetFromAnnotation(pod *corev1.Pod) *annotationResu
 }
 
 // getMatchingTarget: static targets first, then RC, then SSI inject-all if both
-// are absent. A matched deny returns nil and does not fall through.
+// are absent. Inject-all is withheld until the first RC snapshot when on-demand
+// RC is subscribed. A matched deny returns nil and does not fall through.
 func (m *TargetMutator) getMatchingTarget(pod *corev1.Pod) *targetInternal {
 	if _, ok := m.disabledNamespaces[pod.Namespace]; ok {
 		return nil
@@ -494,6 +503,10 @@ func (m *TargetMutator) getMatchingTarget(pod *corev1.Pod) *targetInternal {
 		return t
 	}
 	if m.ssiEnabled && !hasTargets(&m.staticPolicies) && remotePolicies == nil {
+		if !m.allowInjectAll.Load() {
+			log.Debugf("Pod %q skipped SSI inject-all while waiting for the first remote config snapshot", mutatecommon.PodString(pod))
+			return nil
+		}
 		return m.injectAll
 	}
 	return nil

pkg/config/remote/client/client.go

@@ -122,6 +122,10 @@ type Client struct {
 	backoffPolicy     backoff.Policy
 	backoffErrorCount int
 
+	// synced is set after the first successful ClientGetConfigs round-trip,
+	// including an empty snapshot. It is not cleared on later poll failures.
+	synced atomic.Bool
+
 	configFetcher ConfigFetcher
 
 	state *state.Repository
@@ -401,6 +405,14 @@ func (c *Client) GetConfigs(product string) map[string]state.RawConfig {
 	return c.state.GetConfigs(product)
 }
 
+// HasSynced reports whether this client has completed at least one successful
+// config poll against the remote-config service. An empty snapshot counts: the
+// round-trip happened, there were simply no configs for the subscribed
+// products. Later poll failures do not clear this.
+func (c *Client) HasSynced() bool {
+	return c.synced.Load()
+}
+
 // SetCWSWorkloads updates the list of workloads that needs cws profiles
 func (c *Client) SetCWSWorkloads(workloads []string) {
 	c.cwsWorkloads.Store(workloads)
@@ -525,6 +537,7 @@ func (c *Client) update() error {
 	if err != nil {
 		return err
 	}
+	c.synced.Store(true)
 	// We don't want to force the products to reload config if nothing changed
 	// in the latest update.
 	if len(changedProducts) == 0 {

releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml

@@ -0,0 +1,15 @@
+# Each section from every release note are combined when the
+# CHANGELOG-DCA.rst is rendered. So the text needs to be worded so that
+# it does not depend on any information only available in another
+# section. This may mean repeating some details, but each section
+# must be readable independently of the other.
+#
+# Each section note must be formatted as reStructuredText.
+---
+fixes:
+  - |
+    Single Step Instrumentation no longer instruments every eligible pod while
+    the Cluster Agent is still waiting for the first Remote Config workload
+    selection snapshot. Pods created in that window are left unchanged until
+    remote policies arrive, or until an empty snapshot confirms that the
+    organization has no remote policies.