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

pkg/clusteragent/admission/mutate/autoinstrumentation/BUILD.bazel

@@ -100,7 +100,9 @@ dd_agent_go_test(
         "//pkg/clusteragent/admission/mutate/autoinstrumentation/libraryinjection",
         "//pkg/clusteragent/admission/mutate/common",
         "//pkg/config/mock",
+        "//pkg/config/remote/client",
         "//pkg/languagedetection/languagemodels",
+        "//pkg/proto/pbgo/core",
         "//pkg/remoteconfig/state",
         "//pkg/ssi/testutils",
         "//pkg/util/fxutil",

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"
@@ -23,6 +24,10 @@ var (
 	apmPolicyPrefixPattern = regexp.MustCompile(`^(\d+)\.`)
 )
 
+// rcInjectAllWaitTimeout bounds how long SSI inject-all is withheld while
+// waiting for the first remote-config answer. A var so tests can shorten it.
+var rcInjectAllWaitTimeout = time.Minute
+
 // sortRemotePolicyPaths preserves the numeric-prefix ordering used by the
 // APM_POLICIES product. It is intentionally local: Remote Config paths are
 // otherwise opaque and this must not be treated as a generic RC convention.
@@ -63,21 +68,43 @@ 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.
-	m.onRemoteConfigUpdate(client.GetConfigs(state.ProductApmPolicies), client.UpdateApplyStatus)
-	client.Subscribe(state.ProductApmPolicies, m.onRemoteConfigUpdate)
+
+	// WithInitialUpdate: a plain subscription only fires on changes, so "the
+	// backend has no policies for us" would never reach onRemoteConfigUpdate --
+	// and that callback is what releases inject-all.
+	client.SubscribeAll(
+		state.ProductApmPolicies,
+		rcclient.NewUpdateListener(m.onRemoteConfigUpdate),
+		rcclient.WithInitialUpdate(),
+	)
+
+	// Bound the wait: if remote config never answers, instrument rather than
+	// silently withhold SSI from a configuration that asked for it.
+	time.AfterFunc(rcInjectAllWaitTimeout, func() {
+		if m.allowInjectAll.CompareAndSwap(false, true) {
+			log.Warnf("auto-instrumentation: no remote config answer for %q after %s, applying SSI inject-all",
+				state.ProductApmPolicies, rcInjectAllWaitTimeout)
+		}
+	})
+}
+
+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 +143,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

@@ -8,12 +8,16 @@
 package autoinstrumentation
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	rcclient "github.com/DataDog/datadog-agent/pkg/config/remote/client"
+	pbgo "github.com/DataDog/datadog-agent/pkg/proto/pbgo/core"
 	"github.com/DataDog/datadog-agent/pkg/remoteconfig/state"
 	"github.com/DataDog/dd-policy-engine/go/policies"
 )
@@ -276,3 +280,132 @@ 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))
+}
+
+// emptyAnswerFetcher stands in for the remote-config service and answers every
+// poll with no configs at all.
+type emptyAnswerFetcher struct{}
+
+func (emptyAnswerFetcher) ClientGetConfigs(context.Context, *pbgo.ClientGetConfigsRequest) (*pbgo.ClientGetConfigsResponse, error) {
+	return &pbgo.ClientGetConfigsResponse{}, nil
+}
+
+// TestSubscribeRemoteConfig_TimeoutReleasesInjectAll checks the bound on the wait:
+// remote config that never answers must not silently withhold SSI from a
+// configuration that asked for it.
+func TestSubscribeRemoteConfig_TimeoutReleasesInjectAll(t *testing.T) {
+	previous := rcInjectAllWaitTimeout
+	rcInjectAllWaitTimeout = 10 * time.Millisecond
+	t.Cleanup(func() { rcInjectAllWaitTimeout = previous })
+
+	// A client that is never started never answers.
+	client, err := rcclient.NewClient(emptyAnswerFetcher{}, rcclient.WithoutTufVerification())
+	require.NoError(t, err)
+
+	m := newMatchMutator(t, rcSSIOnNoTargets, newMatchTestWmeta(t))
+	m.allowInjectAll.Store(false)
+
+	m.subscribeRemoteConfig(client)
+	require.False(t, m.allowInjectAll.Load(), "not before the deadline")
+
+	require.Eventually(t, func() bool {
+		return m.allowInjectAll.Load()
+	}, 5*time.Second, 10*time.Millisecond)
+
+	name, fromPolicy := matchedTarget(t, m, rcPod("ns", map[string]string{"app": "db"}))
+	require.Equal(t, "default", name)
+	require.False(t, fromPolicy)
+}
+
+// TestSubscribeRemoteConfig_UnsyncedProductKeepsInjectAllClosed is the other half:
+// with no completed poll for APM_POLICIES, an empty local cache is not an answer
+// and inject-all stays withheld.
+func TestSubscribeRemoteConfig_UnsyncedProductKeepsInjectAllClosed(t *testing.T) {
+	previous := rcInjectAllWaitTimeout
+	rcInjectAllWaitTimeout = time.Hour
+	t.Cleanup(func() { rcInjectAllWaitTimeout = previous })
+
+	client, err := rcclient.NewClient(emptyAnswerFetcher{}, rcclient.WithoutTufVerification())
+	require.NoError(t, err)
+
+	m := newMatchMutator(t, rcSSIOnNoTargets, newMatchTestWmeta(t))
+	m.allowInjectAll.Store(false)
+
+	m.subscribeRemoteConfig(client)
+
+	require.False(t, m.allowInjectAll.Load())
+	require.Nil(t, m.getMatchingTarget(rcPod("ns", map[string]string{"app": "db"})))
+}

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/BUILD.bazel

@@ -1,4 +1,5 @@
 load("@rules_go//go:def.bzl", "go_library")
+load("//bazel/rules/go:dd_agent_go_test.bzl", "dd_agent_go_test")
 
 go_library(
     name = "client",
@@ -19,3 +20,14 @@ go_library(
         "@org_uber_go_atomic//:atomic",
     ],
 )
+
+dd_agent_go_test(
+    name = "client_test",
+    srcs = ["client_test.go"],
+    embed = [":client"],
+    deps = [
+        "//pkg/proto/pbgo/core",
+        "//pkg/remoteconfig/state",
+        "@com_github_stretchr_testify//require",
+    ],
+)