From d9bf966e0c2ad63591bb3161016947c87dfdfedc Mon Sep 17 00:00:00 2001 From: Luc Vieillescazes Date: Tue, 25 Aug 2026 14:55:54 +0200 Subject: [PATCH 1/4] fix(ssi): wait for RC before inject-all Empty GetConfigs at Cluster Agent start is not a round-trip. Applying inject-all in that window instruments pods a remote deny would skip. --- .../mutate/autoinstrumentation/rc_policies.go | 39 +++++++++- .../autoinstrumentation/rc_policies_test.go | 74 +++++++++++++++++++ .../target_matching_test.go | 15 ++++ .../autoinstrumentation/target_mutator.go | 27 +++++-- pkg/config/remote/client/client.go | 13 ++++ ...-inject-all-waits-rc-ef94f1e57ea71908.yaml | 15 ++++ 6 files changed, 174 insertions(+), 9 deletions(-) create mode 100644 releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml diff --git a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies.go b/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies.go index 6bb5efd85f72..bb5cf8b50477 100644 --- a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies.go +++ b/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() } diff --git a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go b/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go index 00408dccccc7..fa64661b8cf2 100644 --- a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go +++ b/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)) +} diff --git a/pkg/clusteragent/admission/mutate/autoinstrumentation/target_matching_test.go b/pkg/clusteragent/admission/mutate/autoinstrumentation/target_matching_test.go index ffc8609210fb..4fe1ee840131 100644 --- a/pkg/clusteragent/admission/mutate/autoinstrumentation/target_matching_test.go +++ b/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)) diff --git a/pkg/clusteragent/admission/mutate/autoinstrumentation/target_mutator.go b/pkg/clusteragent/admission/mutate/autoinstrumentation/target_mutator.go index c8cf7c43f7a6..15e6c1b9b1d5 100644 --- a/pkg/clusteragent/admission/mutate/autoinstrumentation/target_mutator.go +++ b/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 diff --git a/pkg/config/remote/client/client.go b/pkg/config/remote/client/client.go index de4b1e84e212..c2626e723453 100644 --- a/pkg/config/remote/client/client.go +++ b/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 { diff --git a/releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml b/releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml new file mode 100644 index 000000000000..18a5341bc94e --- /dev/null +++ b/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. From 7a06ee93e00e69cba4d8769a022e49935c444abd Mon Sep 17 00:00:00 2001 From: Luc Vieillescazes Date: Tue, 1 Sep 2026 12:00:08 +0200 Subject: [PATCH 2/4] feat(remote-config): add WithInitialUpdate option A plain subscription only fires on config changes, so a subscriber cannot tell "the backend has no configs for me" from "the client has not asked yet". WithInitialUpdate delivers the product's current state once: during Subscribe if a poll already carried the product, otherwise on the first poll that does. Opt-in per subscription, so subscribers that do not pass it see no change in behaviour. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/config/remote/client/BUILD.bazel | 12 ++ pkg/config/remote/client/client.go | 132 +++++++++++++++---- pkg/config/remote/client/client_test.go | 166 ++++++++++++++++++++++++ 3 files changed, 286 insertions(+), 24 deletions(-) create mode 100644 pkg/config/remote/client/client_test.go diff --git a/pkg/config/remote/client/BUILD.bazel b/pkg/config/remote/client/BUILD.bazel index de757ab0ce0c..c9a6311c89ea 100644 --- a/pkg/config/remote/client/BUILD.bazel +++ b/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", + ], +) diff --git a/pkg/config/remote/client/client.go b/pkg/config/remote/client/client.go index c2626e723453..a7ac8fefe894 100644 --- a/pkg/config/remote/client/client.go +++ b/pkg/config/remote/client/client.go @@ -64,7 +64,8 @@ type Listener interface { // * status: The apply status indicating success, failure, or error details // // Behavior: - // - Called only when there are actual configuration changes to process + // - Called when there are configuration changes to process, and once with the + // current configs if the subscription used WithInitialUpdate // - May be skipped if signature verification fails and ShouldIgnoreSignatureExpiration() returns false // - Listeners should process all provided configurations and report their apply status // - The applyStateCallback must be called for proper state tracking and error reporting @@ -122,16 +123,21 @@ 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 listeners map[string][]Listener + // initialUpdates holds, per product, the listeners subscribed + // WithInitialUpdate that have not been delivered to yet. A product is removed + // once all of its listeners have been. Guarded by m. + initialUpdates map[string]map[Listener]struct{} + + // syncedProducts holds the products carried by a ClientGetConfigs round-trip + // that completed successfully. Guarded by m. + syncedProducts map[string]struct{} + // Elements that can be changed during the execution of listeners // They are atomics so that they don't have to share the top-level mutex // when in use @@ -335,6 +341,8 @@ func newClient(cf ConfigFetcher, opts ...func(opts *Options)) (*Client, error) { state: repository, backoffPolicy: backoffPolicy, listeners: make(map[string][]Listener), + initialUpdates: make(map[string]map[Listener]struct{}), + syncedProducts: make(map[string]struct{}), configFetcher: cf, }, nil } @@ -374,8 +382,37 @@ func (c *Client) SetAgentName(agentName string) { } } +// SubscribeOption customizes how a subscription is delivered. +type SubscribeOption func(*subscribeOptions) + +type subscribeOptions struct { + initialUpdate bool +} + +// WithInitialUpdate guarantees the subscriber is called once with the product's +// current state, including when that state is empty. If a poll carrying the +// product has already completed, the delivery happens during Subscribe; +// otherwise it happens on the first poll that carries it. +// +// By default a subscriber only sees changes, so it cannot tell "the backend has +// no configs for me" from "the client has not asked yet". Use this option when +// an empty config set is itself an answer the subscriber is gated on. It does +// not backdate the guarantee: a poll that did not carry the product is not an +// answer for it. +// +// The delivery may run on the calling goroutine, so the listener must not call +// back into the client. +func WithInitialUpdate() SubscribeOption { + return func(opts *subscribeOptions) { opts.initialUpdate = true } +} + // SubscribeAll subscribes to all events (config updates, state changed, ...) -func (c *Client) SubscribeAll(product string, listener Listener) { +func (c *Client) SubscribeAll(product string, listener Listener, opts ...SubscribeOption) { + var options subscribeOptions + for _, opt := range opts { + opt(&options) + } + c.m.Lock() defer c.m.Unlock() @@ -386,6 +423,19 @@ func (c *Client) SubscribeAll(product string, listener Listener) { } c.listeners[product] = append(c.listeners[product], listener) + if options.initialUpdate { + if c.hasSyncedProductLocked(product) { + // The product already has a backend answer, so deliver it here rather + // than waiting for the next poll. Every delivery happens under m, so a + // concurrent poll cannot notify this listener before we do. + listener.OnUpdate(c.state.GetConfigs(product), c.state.UpdateApplyStatus) + } else { + if c.initialUpdates[product] == nil { + c.initialUpdates[product] = make(map[Listener]struct{}) + } + c.initialUpdates[product][listener] = struct{}{} + } + } } // Subscribe subscribes to config updates of a product. @@ -405,12 +455,23 @@ 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() +// hasSyncedProduct reports whether a poll that requested the given product has +// completed successfully. An empty result counts: the backend answered, it simply +// had nothing for that product. +// +// The signal is product-scoped because a poll says nothing about a product it did +// not request: subscribing to a product does not backdate its sync. Callers must +// hold m. +func (c *Client) hasSyncedProductLocked(product string) bool { + _, ok := c.syncedProducts[product] + return ok +} + +// hasSyncedProduct is hasSyncedProductLocked, taking m. +func (c *Client) hasSyncedProduct(product string) bool { + c.m.Lock() + defer c.m.Unlock() + return c.hasSyncedProductLocked(product) } // SetCWSWorkloads updates the list of workloads that needs cws profiles @@ -537,23 +598,46 @@ func (c *Client) update() error { if err != nil { return err } - c.synced.Store(true) + c.m.Lock() + defer c.m.Unlock() + // The request holds the products that were actually on the wire, and this poll + // is an answer for every one of them -- including those it returned nothing + // for. + polledProducts := req.Client.GetProducts() + for _, product := range polledProducts { + c.syncedProducts[product] = struct{}{} + } // We don't want to force the products to reload config if nothing changed - // in the latest update. - if len(changedProducts) == 0 { + // in the latest update, unless a listener is still owed its initial update: + // an empty answer is not a config change, but it is the answer that listener + // is gated on. + if len(changedProducts) == 0 && len(c.initialUpdates) == 0 { return nil } - - c.m.Lock() - defer c.m.Unlock() for product, productListeners := range c.listeners { - if containsProduct(changedProducts, product) { - for _, listener := range productListeners { - if response.ConfigStatus == pbgo.ConfigStatus_CONFIG_STATUS_OK || - !listener.ShouldIgnoreSignatureExpiration() { - listener.OnUpdate(c.state.GetConfigs(product), c.state.UpdateApplyStatus) - } + changed := containsProduct(changedProducts, product) + // A poll says nothing about a product it did not request, so it cannot + // serve as that product's initial update. + var owed map[Listener]struct{} + if containsProduct(polledProducts, product) { + owed = c.initialUpdates[product] + } + if !changed && len(owed) == 0 { + continue + } + for _, listener := range productListeners { + _, owedInitialUpdate := owed[listener] + if !changed && !owedInitialUpdate { + continue } + if response.ConfigStatus == pbgo.ConfigStatus_CONFIG_STATUS_OK || + !listener.ShouldIgnoreSignatureExpiration() { + delete(owed, listener) + listener.OnUpdate(c.state.GetConfigs(product), c.state.UpdateApplyStatus) + } + } + if owed != nil && len(owed) == 0 { + delete(c.initialUpdates, product) } } return nil diff --git a/pkg/config/remote/client/client_test.go b/pkg/config/remote/client/client_test.go new file mode 100644 index 000000000000..d94b6955dd86 --- /dev/null +++ b/pkg/config/remote/client/client_test.go @@ -0,0 +1,166 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// This product includes software developed at Datadog (https://www.datadoghq.com/). +// Copyright 2016-present Datadog, Inc. + +package client + +import ( + "context" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + pbgo "github.com/DataDog/datadog-agent/pkg/proto/pbgo/core" + "github.com/DataDog/datadog-agent/pkg/remoteconfig/state" +) + +// fakeConfigFetcher answers every poll with an empty config set and records the +// products each poll asked for. +type fakeConfigFetcher struct { + mu sync.Mutex + requests [][]string +} + +func (f *fakeConfigFetcher) ClientGetConfigs(_ context.Context, req *pbgo.ClientGetConfigsRequest) (*pbgo.ClientGetConfigsResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.requests = append(f.requests, req.Client.GetProducts()) + return &pbgo.ClientGetConfigsResponse{}, nil +} + +func (f *fakeConfigFetcher) polledProducts() [][]string { + f.mu.Lock() + defer f.mu.Unlock() + return f.requests +} + +// recordingSubscriber captures the updates delivered to a subscription. +type recordingSubscriber struct { + mu sync.Mutex + updates []map[string]state.RawConfig +} + +func (r *recordingSubscriber) onUpdate(updates map[string]state.RawConfig, _ func(string, state.ApplyStatus)) { + r.mu.Lock() + defer r.mu.Unlock() + r.updates = append(r.updates, updates) +} + +func (r *recordingSubscriber) count() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.updates) +} + +// TestWithInitialUpdate_DeliversEmptyConfigSet covers the guarantee a plain +// subscription does not give: a poll that returned nothing for the product still +// reaches the subscriber, so it can tell an empty backend answer from a cache +// that was never filled. +func TestWithInitialUpdate_DeliversEmptyConfigSet(t *testing.T) { + c, err := NewClient(&fakeConfigFetcher{}, WithoutTufVerification()) + require.NoError(t, err) + + sub := &recordingSubscriber{} + c.SubscribeAll(state.ProductApmPolicies, NewUpdateListener(sub.onUpdate), WithInitialUpdate()) + + require.NoError(t, c.update()) + require.Equal(t, 1, sub.count()) + require.Empty(t, sub.updates[0]) + + // The guarantee is one-shot: later polls that change nothing stay silent. + require.NoError(t, c.update()) + require.Equal(t, 1, sub.count()) +} + +// TestSubscribe_DoesNotDeliverEmptyConfigSet documents the default behaviour +// that makes WithInitialUpdate necessary. +func TestSubscribe_DoesNotDeliverEmptyConfigSet(t *testing.T) { + c, err := NewClient(&fakeConfigFetcher{}, WithoutTufVerification()) + require.NoError(t, err) + + sub := &recordingSubscriber{} + c.Subscribe(state.ProductApmPolicies, sub.onUpdate) + + require.NoError(t, c.update()) + require.Zero(t, sub.count()) +} + +// TestWithInitialUpdate_WaitsForAPollCarryingTheProduct is the property callers +// gate behaviour on: polls that completed before the subscription never asked for +// the product, so they cannot serve as its initial update. +func TestWithInitialUpdate_WaitsForAPollCarryingTheProduct(t *testing.T) { + fetcher := &fakeConfigFetcher{} + c, err := NewClient(fetcher, WithProducts(state.ProductAgentConfig), WithoutTufVerification()) + require.NoError(t, err) + + // A poll completes before anyone subscribes to APM_POLICIES. + require.NoError(t, c.update()) + + sub := &recordingSubscriber{} + c.SubscribeAll(state.ProductApmPolicies, NewUpdateListener(sub.onUpdate), WithInitialUpdate()) + require.Zero(t, sub.count()) + + require.NoError(t, c.update()) + require.Equal(t, 1, sub.count()) + + polled := fetcher.polledProducts() + require.Equal(t, []string{state.ProductAgentConfig}, polled[0]) + require.Contains(t, polled[1], state.ProductApmPolicies) +} + +// TestWithInitialUpdate_LeavesOtherSubscribersAlone checks the opt-in is per +// subscription: a subscriber to the same product that did not ask for an initial +// update keeps seeing changes only. +func TestWithInitialUpdate_LeavesOtherSubscribersAlone(t *testing.T) { + c, err := NewClient(&fakeConfigFetcher{}, WithoutTufVerification()) + require.NoError(t, err) + + optedIn, plain := &recordingSubscriber{}, &recordingSubscriber{} + c.SubscribeAll(state.ProductApmPolicies, NewUpdateListener(optedIn.onUpdate), WithInitialUpdate()) + c.Subscribe(state.ProductApmPolicies, plain.onUpdate) + + require.NoError(t, c.update()) + require.Equal(t, 1, optedIn.count()) + require.Zero(t, plain.count()) +} + +// TestHasSyncedProduct_OnlyForProductsOnTheWire covers the signal callers use to +// read GetConfigs as an answer: a poll answers for the products it requested and +// for nothing else. +func TestHasSyncedProduct_OnlyForProductsOnTheWire(t *testing.T) { + c, err := NewClient(&fakeConfigFetcher{}, WithProducts(state.ProductAgentConfig), WithoutTufVerification()) + require.NoError(t, err) + + require.False(t, c.hasSyncedProduct(state.ProductAgentConfig), "no poll has completed yet") + + require.NoError(t, c.update()) + require.True(t, c.hasSyncedProduct(state.ProductAgentConfig)) + require.False(t, c.hasSyncedProduct(state.ProductApmPolicies), "that poll never asked for it") + + // Subscribing does not backdate the sync: it takes another poll. + c.Subscribe(state.ProductApmPolicies, func(map[string]state.RawConfig, func(string, state.ApplyStatus)) {}) + require.False(t, c.hasSyncedProduct(state.ProductApmPolicies)) + + require.NoError(t, c.update()) + require.True(t, c.hasSyncedProduct(state.ProductApmPolicies)) +} + +// TestWithInitialUpdate_DeliversDuringSubscribeWhenAlreadySynced covers the case +// the cluster-agent actually hits: the client has already polled by the time the +// subscriber shows up, so the answer it holds must not wait for another poll. +func TestWithInitialUpdate_DeliversDuringSubscribeWhenAlreadySynced(t *testing.T) { + c, err := NewClient(&fakeConfigFetcher{}, WithProducts(state.ProductApmPolicies), WithoutTufVerification()) + require.NoError(t, err) + require.NoError(t, c.update()) + + sub := &recordingSubscriber{} + c.SubscribeAll(state.ProductApmPolicies, NewUpdateListener(sub.onUpdate), WithInitialUpdate()) + + require.Equal(t, 1, sub.count(), "delivery should happen during Subscribe, not on the next poll") + + // And it stays a one-shot. + require.NoError(t, c.update()) + require.Equal(t, 1, sub.count()) +} From 381f7d09e867280ac8aabec2543043db1c9489fc Mon Sep 17 00:00:00 2001 From: Luc Vieillescazes Date: Tue, 1 Sep 2026 12:00:09 +0200 Subject: [PATCH 3/4] fix(ssi): bound the wait for RC before inject-all Subscribe with WithInitialUpdate so a confirmed-empty APM_POLICIES snapshot reaches onRemoteConfigUpdate, which is what releases inject-all, and so a snapshot the client already holds is applied without waiting for another poll. A successful poll of the local RC service is not proof that it reached the backend, so "no answer" cannot be told apart from "no policies". Bound the wait at one minute and instrument, rather than silently withhold SSI from a configuration that asked for it. Co-Authored-By: Claude Opus 5 (1M context) --- .../mutate/autoinstrumentation/BUILD.bazel | 2 + .../mutate/autoinstrumentation/rc_policies.go | 47 +++++++-------- .../autoinstrumentation/rc_policies_test.go | 59 +++++++++++++++++++ 3 files changed, 81 insertions(+), 27 deletions(-) diff --git a/pkg/clusteragent/admission/mutate/autoinstrumentation/BUILD.bazel b/pkg/clusteragent/admission/mutate/autoinstrumentation/BUILD.bazel index 518a9e217c15..cedc0c41a854 100644 --- a/pkg/clusteragent/admission/mutate/autoinstrumentation/BUILD.bazel +++ b/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", diff --git a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies.go b/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies.go index bb5cf8b50477..ab70b3a2613c 100644 --- a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies.go +++ b/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies.go @@ -24,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. @@ -70,34 +74,23 @@ func (m *TargetMutator) subscribeRemoteConfig(client *rcclient.Client) { log.Infof("auto-instrumentation: subscribing to remote config product %q for SSI policies", state.ProductApmPolicies) - 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 + // 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() { diff --git a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go b/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go index fa64661b8cf2..1e29621b8c0f 100644 --- a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go +++ b/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" ) @@ -350,3 +354,58 @@ func TestOnRemoteConfigUpdate_InvalidFirstSnapshotStaysPending(t *testing.T) { 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"}))) +} From caa859b8771ed0dacf5927ccdb5cd0b0d1fafe82 Mon Sep 17 00:00:00 2001 From: Luc Vieillescazes Date: Tue, 1 Sep 2026 12:28:04 +0200 Subject: [PATCH 4/4] docs(ssi): mention RC timeout in inject-all reno --- .../notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml b/releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml index 18a5341bc94e..f8061e300f45 100644 --- a/releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml +++ b/releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml @@ -11,5 +11,6 @@ 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. + remote policies arrive, an empty snapshot confirms that the organization + has no remote policies, or one minute elapses without an answer (in which + case instrumentation proceeds and a warning is logged).