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 6bb5efd85f72..ab70b3a2613c 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" @@ -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,14 +68,35 @@ 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)) { @@ -78,6 +104,7 @@ func (m *TargetMutator) onRemoteConfigUpdate(updates map[string]state.RawConfig, 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() } diff --git a/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go b/pkg/clusteragent/admission/mutate/autoinstrumentation/rc_policies_test.go index 00408dccccc7..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" ) @@ -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"}))) +} 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/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 de4b1e84e212..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 @@ -128,6 +129,15 @@ type Client struct { 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 @@ -331,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 } @@ -370,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() @@ -382,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. @@ -401,6 +455,25 @@ func (c *Client) GetConfigs(product string) map[string]state.RawConfig { return c.state.GetConfigs(product) } +// 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 func (c *Client) SetCWSWorkloads(workloads []string) { c.cwsWorkloads.Store(workloads) @@ -525,22 +598,46 @@ func (c *Client) update() error { if err != nil { return err } + 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()) +} 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..f8061e300f45 --- /dev/null +++ b/releasenotes-dca/notes/ssi-inject-all-waits-rc-ef94f1e57ea71908.yaml @@ -0,0 +1,16 @@ +# 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, 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).