pkg/k8score/dynamic_cache_retry_test.go
Rama main · réplica de lectura del repositorio original.
package k8score
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"slices"
"sync"
"sync/atomic"
"testing"
"time"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
fakediscovery "k8s.io/client-go/discovery/fake"
"k8s.io/client-go/dynamic"
dynamicfake "k8s.io/client-go/dynamic/fake"
"k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/rest"
ktesting "k8s.io/client-go/testing"
)
func TestProbeRetryNamespaceOutcomes(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, tc := range []struct {
name string
allowed bool
refused int
unresolved bool
}{
{"OK", true, 2, false}, {"Forbidden", false, 2, false}, {"exhausted", true, 100, true},
} {
t.Run(tc.name, func(t *testing.T) {
allowed := []string{"a"}
if tc.allowed {
allowed = append(allowed, "b")
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, allowed, map[string]int{"b": tc.refused}), NamespaceFallbacks: []string{"a", "b"}, ProbeRetryBackoff: []time.Duration{100 * time.Millisecond, time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
view := d.NewObservationView([]schema.GroupVersionResource{gvr})
// Exercise startup's owner, rather than relying on a later read.
d.WarmupParallel([]schema.GroupVersionResource{gvr}, time.Millisecond)
initial := d.Observation(gvr)
if got := view.NamespaceObservation(gvr, ""); got.ProbeRetrying != initial.ProbeRetrying || !slices.Equal(got.UnansweredNamespaces, initial.UnansweredNamespaces) {
t.Fatalf("initial view: %+v; cache: %+v", got, initial)
}
if !initial.ProbeRetrying || !slices.Equal(initial.UnansweredNamespaces, []string{"b"}) {
t.Fatalf("retry not published: %+v", initial)
}
if specific := d.NamespaceObservation(gvr, "a"); specific.ProbeRetrying || len(specific.UnansweredNamespaces) > 0 {
t.Fatalf("a inherited b's gap: %+v", specific)
}
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(3 * time.Second):
t.Fatal("retry did not finish")
}
got := d.Observation(gvr)
if got.ProbeRetrying || (len(got.UnansweredNamespaces) > 0) != tc.unresolved {
t.Fatalf("terminal observation: %+v", got)
}
for _, ns := range []string{"a", "b", "outside"} {
want := d.NamespaceObservation(gvr, ns)
got := view.NamespaceObservation(gvr, ns)
if got.Scope != want.Scope || !slices.Equal(got.Namespaces, want.Namespaces) || got.ProbeRetrying != want.ProbeRetrying || !slices.Equal(got.UnansweredNamespaces, want.UnansweredNamespaces) {
t.Fatalf("terminal %s view: %+v; cache: %+v", ns, got, want)
}
}
if d.HasCoveringInformer(gvr, "b") != (tc.allowed && !tc.unresolved) {
t.Fatalf("wrong namespace watch: %+v", got)
}
if !tc.unresolved && !d.WaitForSync(gvr, time.Second) {
t.Fatal("watch did not sync")
}
if tc.name == "Forbidden" && (got.State == DynamicObservationDenied || got.ReasonCode == "scope_probe_incomplete" || got.Truncated) {
t.Fatalf("denial did not become scope: %+v", got)
}
if tc.unresolved {
// Repeated reads must not renew the retry budget.
for range 10 {
if err := d.EnsureWatching(gvr); err != nil {
t.Fatal(err)
}
}
d.mu.RLock()
same := d.probeRetries[gvr] == r && !r.running
d.mu.RUnlock()
if !same {
t.Fatal("exhausted retry restarted")
}
}
})
}
}
func TestProbeRetryWholeKindOutcomes(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, tc := range []struct {
name string
allowed []string
refused int
fallbacks []string
want DynamicObservationState
whole bool
}{
{"OK", []string{""}, 1, nil, DynamicObservationSynced, false},
{"Forbidden", nil, 1, nil, DynamicObservationDenied, false},
// When the budget runs out the kind is reported unanswered; the slow
// recovery re-checks it later (TestProbeRecoveryAfterTheBudget).
{"exhausted", nil, 100, nil, DynamicObservationDeferred, true},
{"exhausted, with fallbacks", nil, 100, []string{"a"}, DynamicObservationDeferred, true},
} {
t.Run(tc.name, func(t *testing.T) {
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, tc.allowed, map[string]int{"": tc.refused}), NamespaceFallbacks: tc.fallbacks, ProbeRetryBackoff: []time.Duration{50 * time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); !IsClientRateLimited(err) {
t.Fatalf("initial walk: %v", err)
}
if !d.Observation(gvr).ProbeRetrying {
t.Fatal("whole-kind retry not published")
}
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not finish")
}
if tc.want == DynamicObservationSynced && !d.WaitForSync(gvr, time.Second) {
t.Fatal("watch did not sync")
}
got := d.Observation(gvr)
if got.ProbeRetrying || got.State != tc.want || (got.ReasonCode == "scope_probe_incomplete") != tc.whole {
t.Fatalf("terminal observation: %+v", got)
}
})
}
}
// A 401 rejects Radar's credentials, not one scope. A retry that meets one
// records it as such, stops, and settling the walk doesn't overwrite it.
func TestProbeRetryKeepsRejectedCredentials(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, tc := range []struct {
name string
fallbacks []string
// throttled is how often each namespace's probe is refused by the
// client-side rate limiter before it gets the 401: enough to outlast
// the first walk, which retries throttled candidates once itself.
throttled map[string]int
// forbiddenClusterWide answers the cluster-wide check with a 403, so
// the walk falls back to namespaces.
forbiddenClusterWide bool
}{
{"whole kind", nil, map[string]int{"": 1}, false},
{"namespace", []string{"a", "b"}, map[string]int{"a": 2, "b": 2}, true},
} {
t.Run(tc.name, func(t *testing.T) {
var mu sync.Mutex
probed := map[string]int{}
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{gvr: "WidgetList"})
dyn.PrependReactor("list", "widgets", func(action ktesting.Action) (bool, runtime.Object, error) {
ns := action.GetNamespace()
mu.Lock()
defer mu.Unlock()
probed[ns]++
if ns == "" && tc.forbiddenClusterWide {
return true, nil, apierrors.NewForbidden(gvr.GroupResource(), "", errors.New("collector cannot list"))
}
if tc.throttled[ns] > 0 {
tc.throttled[ns]--
return true, nil, errClientRateLimited
}
return true, nil, apierrors.NewUnauthorized("token rejected")
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, NamespaceFallbacks: tc.fallbacks, ProbeRetryBackoff: []time.Duration{20 * time.Millisecond, 20 * time.Millisecond}, ProbeRecoveryIntervals: []time.Duration{10 * time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
_ = d.EnsureWatching(gvr)
mu.Lock()
bBefore := probed["b"]
mu.Unlock()
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
if r == nil {
t.Fatal("no retry started for the unanswered check")
}
select {
case <-r.done:
case <-time.After(3 * time.Second):
t.Fatal("retry did not finish")
}
if got := d.Observation(gvr); got.State != DynamicObservationDenied || got.ReasonCode != "credentials_rejected" {
t.Fatalf("observation = %+v, want denied/credentials_rejected", got)
}
// Nor does the slow recovery after the budget probe again.
deadline := time.Now().Add(time.Second)
for ownerLive(d, r) && time.Now().Before(deadline) {
time.Sleep(5 * time.Millisecond)
}
mu.Lock()
total := 0
for _, n := range probed {
total += n
}
mu.Unlock()
time.Sleep(100 * time.Millisecond)
mu.Lock()
defer mu.Unlock()
after := 0
for _, n := range probed {
after += n
}
if ownerLive(d, r) || after != total {
t.Fatalf("re-checks went on after a 401: owner live %v, %d probes then %d", ownerLive(d, r), total, after)
}
if len(tc.fallbacks) > 1 && probed["b"] != bBefore {
t.Fatalf("the retry went on probing after a 401: b probed %d times, %d before the retry", probed["b"], bBefore)
}
})
}
}
func TestProbeRetryCandidateCapIsOnlySetup(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, []string{"a"}, nil), NamespaceFallbacks: []string{"a"}, NamespaceFallbacksTruncated: true})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); err != nil {
t.Fatal(err)
}
if !d.WaitForSync(gvr, time.Second) {
t.Fatal("watch did not sync")
}
got := d.Observation(gvr)
if !got.Truncated || got.ProbeRetrying || len(got.UnansweredNamespaces) > 0 {
t.Fatalf("cap became unanswered: %+v", got)
}
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
if r != nil {
t.Fatal("cap started a retry")
}
}
func TestProbeRetryOneLoopPerGVR(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, nil, map[string]int{"b": 100}), ProbeRetryBackoff: []time.Duration{time.Hour}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
o := probeOutcome{startedAt: time.Now(), unanswered: []string{"b"}}
d.noteWalk(gvr, o)
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
var wg sync.WaitGroup
for range 20 {
wg.Go(func() { d.noteWalk(gvr, o) })
}
wg.Wait()
d.mu.RLock()
same := d.probeRetries[gvr] == r && r.running
d.mu.RUnlock()
if !same {
t.Fatal("multiple retry owners")
}
d.Stop()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("Stop leaked scheduled retry")
}
d.mu.RLock()
running := r.running
d.mu.RUnlock()
if running {
t.Fatal("Stop left retry running")
}
}
func TestProbeRetryDeadlineIncludesFirstWalk(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
var probes atomic.Int32
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{gvr: "WidgetList"})
dyn.PrependReactor("list", "widgets", func(ktesting.Action) (bool, runtime.Object, error) {
probes.Add(1)
return true, nil, context.DeadlineExceeded
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, ProbeRetryBackoff: []time.Duration{time.Hour}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
d.noteWalk(gvr, probeOutcome{startedAt: time.Now().Add(-probeRetryBudget + 50*time.Millisecond), unanswered: []string{"b"}})
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry budget reset after first walk")
}
if probes.Load() != 0 || !slices.Equal(d.Observation(gvr).UnansweredNamespaces, []string{"b"}) {
t.Fatal("deadline lost unanswered check")
}
}
func TestProbeRetryLaterReadClearsUnanswered(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, []string{"a", "b"}, map[string]int{"b": 3}), NamespaceFallbacks: []string{"a", "b"}, ProbeRetryBackoff: []time.Duration{time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); err != nil {
t.Fatal(err)
}
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not exhaust")
}
if !slices.Equal(d.Observation(gvr).UnansweredNamespaces, []string{"b"}) {
t.Fatal("missing exhausted namespace")
}
if err := d.EnsureWatchingContext(t.Context(), gvr, "b"); err != nil {
t.Fatal(err)
}
got := d.Observation(gvr)
if got.ProbeRetrying || len(got.UnansweredNamespaces) > 0 || !d.HasCoveringInformer(gvr, "b") {
t.Fatalf("successful read did not settle gap: %+v", got)
}
}
func TestProbeRetryStopCancelsInFlightRequest(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
entered, canceled := make(chan struct{}), make(chan struct{})
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
close(entered)
<-r.Context().Done()
close(canceled)
}))
defer srv.Close()
client, err := dynamic.NewForConfig(&rest.Config{Host: srv.URL})
if err != nil {
t.Fatal(err)
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: client, ProbeRetryBackoff: []time.Duration{time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
// Inject an already unanswered walk to isolate the retry request's cancellation.
d.noteWalk(gvr, probeOutcome{startedAt: time.Now(), unanswered: []string{"b"}})
select {
case <-entered:
case <-time.After(time.Second):
t.Fatal("retry request not started")
}
d.mu.RLock()
retry := d.probeRetries[gvr]
d.mu.RUnlock()
d.Stop()
select {
case <-retry.done:
case <-time.After(time.Second):
t.Fatal("Stop leaked in-flight retry")
}
select {
case <-canceled:
case <-time.After(time.Second):
t.Fatal("Stop did not cancel HTTP request")
}
}
func TestProbeRetryNoInformerNamespaceIsolation(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, nil, map[string]int{"b": 100}), NamespaceFallbacks: []string{"a", "b"}, ProbeRetryBackoff: []time.Duration{time.Hour}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); !IsClientRateLimited(err) {
t.Fatalf("initial walk: %v", err)
}
a, b := d.NamespaceObservation(gvr, "a"), d.NamespaceObservation(gvr, "b")
if a.ProbeRetrying || len(a.UnansweredNamespaces) > 0 || a.ReasonCode == "scope_probe_incomplete" || a.Scope != DynamicObservationScopeExplicitNamespaces {
t.Fatalf("answered namespace inherited gap: %+v", a)
}
if !b.ProbeRetrying || !slices.Equal(b.UnansweredNamespaces, []string{"b"}) {
t.Fatalf("unanswered namespace lost gap: %+v", b)
}
}
func TestProbeRetryAllNamespacesForbiddenBecomeScope(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, nil, map[string]int{"b": 2}), NamespaceFallbacks: []string{"b"}, ProbeRetryBackoff: []time.Duration{time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); !IsClientRateLimited(err) {
t.Fatalf("initial walk: %v", err)
}
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not finish")
}
got := d.Observation(gvr)
if got.ProbeRetrying || len(got.UnansweredNamespaces) > 0 || got.Scope != DynamicObservationScopeExplicitNamespaces || len(got.Namespaces) > 0 {
t.Fatalf("Forbidden did not become empty collector scope: %+v", got)
}
}
func TestProbeRetryClusterScopedDenialStaysKindAccess(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
discovery, err := NewResourceDiscovery(fake.NewClientset().Discovery())
if err != nil {
t.Fatal(err)
}
discovery.AddAPIResource(APIResource{Group: gvr.Group, Version: gvr.Version, Name: gvr.Resource, Kind: "Widget", Namespaced: false, Verbs: []string{"list", "watch"}})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, nil, map[string]int{"": 1}), Discovery: discovery, NamespaceFallbacks: []string{"b"}, NamespaceFallbacksTruncated: true, ProbeRetryBackoff: []time.Duration{time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); !IsClientRateLimited(err) {
t.Fatalf("initial walk: %v", err)
}
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not finish")
}
got := d.Observation(gvr)
if got.State != DynamicObservationDenied || got.Scope == DynamicObservationScopeExplicitNamespaces || got.Truncated || got.ProbeRetrying || len(got.UnansweredNamespaces) > 0 {
t.Fatalf("cluster-scoped denial changed scope: %+v", got)
}
}
func TestProbeRetryExplicitNamespaceFirst(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, clusterWide := range []bool{false, true} {
t.Run(fmt.Sprintf("clusterWide=%v", clusterWide), func(t *testing.T) {
allowed, refused := []string{"a", "b"}, map[string]int{"a": 2, "b": 1}
if clusterWide {
allowed, refused = []string{"", "a", "b"}, map[string]int{"": 1}
}
dyn := rateLimitedLists(gvr, allowed, refused)
for _, ns := range []string{"a", "b"} {
if err := dyn.Tracker().Add(&unstructured.Unstructured{Object: map[string]any{"apiVersion": "example.io/v1", "kind": "Widget", "metadata": map[string]any{"namespace": ns, "name": "robusta-" + ns}}}); err != nil {
t.Fatal(err)
}
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, NamespaceFallbacks: []string{"a", "b"}, ProbeRetryBackoff: []time.Duration{50 * time.Millisecond, 500 * time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatchingContext(t.Context(), gvr, "b"); !IsClientRateLimited(err) {
t.Fatalf("initial explicit read: %v", err)
}
d.mu.RLock()
r := d.probeRetries[gvr]
fullWalk, whole := r.fullWalk, r.whole
d.mu.RUnlock()
if fullWalk || whole != clusterWide {
t.Fatalf("explicit walk lost its scope: fullWalk=%v whole=%v", fullWalk, whole)
}
if !clusterWide {
deadline := time.Now().Add(time.Second)
for !d.HasCoveringInformer(gvr, "b") && time.Now().Before(deadline) {
time.Sleep(time.Millisecond)
}
if !d.HasCoveringInformer(gvr, "b") || d.HasCoveringInformer(gvr, "a") {
t.Fatal("expected b recovered while a remains unanswered")
}
if !d.WaitForSync(gvr, time.Second) {
t.Fatal("b did not sync")
}
if _, err := d.Count(gvr, nil); err == nil {
t.Fatalf("incomplete fanout produced an all-namespace count: %v", err)
}
if o := d.Observation(gvr); !o.ProbeRetrying || !slices.Equal(o.UnansweredNamespaces, []string{"a"}) {
t.Fatalf("one namespace settled the scope: %+v", o)
}
}
select {
case <-r.done:
case <-time.After(2 * time.Second):
t.Fatal("retry did not finish")
}
if !d.WaitForSync(gvr, time.Second) || !d.HasCoveringInformer(gvr, "a") || !d.HasCoveringInformer(gvr, "b") {
t.Fatal("both candidates were not eventually watched")
}
if got, err := d.Count(gvr, nil); err != nil || got != 2 {
t.Fatalf("all-namespace count=%d err=%v", got, err)
}
if clusterWide && d.Observation(gvr).Scope != DynamicObservationScopeCluster {
t.Fatalf("cluster-wide access became a namespace-only informer: %+v", d.Observation(gvr))
}
})
}
}
func TestProbeRetryExhaustedNormalReadRecovery(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, read := range []string{"EnsureWatching", "List", "Get"} {
t.Run(read, func(t *testing.T) {
var allowed atomic.Bool
obj := &unstructured.Unstructured{Object: map[string]any{"apiVersion": "example.io/v1", "kind": "Widget", "metadata": map[string]any{"name": "robusta"}}}
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{gvr: "WidgetList"}, obj)
dyn.PrependReactor("list", "widgets", func(ktesting.Action) (bool, runtime.Object, error) {
if !allowed.Load() {
return true, nil, errClientRateLimited
}
return false, nil, nil
})
discovery, err := NewResourceDiscovery(fake.NewClientset().Discovery())
if err != nil {
t.Fatal(err)
}
discovery.AddAPIResource(APIResource{Group: gvr.Group, Version: gvr.Version, Name: gvr.Resource, Kind: "Widget", Namespaced: false, Verbs: []string{"get", "list", "watch"}})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, Discovery: discovery, ProbeRetryBackoff: []time.Duration{time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); !IsClientRateLimited(err) {
t.Fatalf("initial read: %v", err)
}
d.mu.RLock()
r := d.probeRetries[gvr]
deadline := r.deadline
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not exhaust")
}
allowed.Store(true)
switch read {
case "EnsureWatching":
err = d.EnsureWatching(gvr)
case "List":
_, err = d.List(gvr, "")
case "Get":
_, err = d.Get(gvr, "", "robusta")
}
if err != nil || !d.WaitForSync(gvr, time.Second) {
t.Fatalf("normal read did not recover: %v", err)
}
if o := d.Observation(gvr); o.ProbeRetrying || o.State != DynamicObservationSynced || o.ReasonCode == "scope_probe_incomplete" {
t.Fatalf("exhausted observation did not recover: %+v", o)
}
d.mu.RLock()
same := d.probeRetries[gvr] == r && !r.running && r.deadline.Equal(deadline)
d.mu.RUnlock()
if !same {
t.Fatal("on-demand read renewed the automatic retry")
}
})
}
}
func TestProbeRetryMergesNewNamespaces(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, exhausted := range []bool{false, true} {
t.Run(fmt.Sprintf("exhausted=%v", exhausted), func(t *testing.T) {
var cAllowed atomic.Bool
dyn := rateLimitedLists(gvr, []string{"b", "c"}, map[string]int{"b": 100, "c": 1})
dyn.PrependReactor("list", "widgets", func(a ktesting.Action) (bool, runtime.Object, error) {
if a.GetNamespace() == "c" && cAllowed.Load() {
return true, &unstructured.UnstructuredList{}, nil
}
return false, nil, nil
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, NamespaceFallbacks: []string{"b"}, ProbeRetryBackoff: []time.Duration{100 * time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
startedAt := time.Now()
if exhausted {
startedAt = startedAt.Add(-probeRetryBudget)
}
d.noteWalk(gvr, probeOutcome{startedAt: startedAt, fullWalk: true, unanswered: []string{"b"}})
d.mu.RLock()
r, deadline := d.probeRetries[gvr], d.probeRetries[gvr].deadline
d.mu.RUnlock()
if exhausted {
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not exhaust")
}
}
if err := d.EnsureWatchingContext(t.Context(), gvr, "c"); !IsClientRateLimited(err) {
t.Fatalf("new namespace read: %v", err)
}
o := d.Observation(gvr)
if !slices.Equal(o.UnansweredNamespaces, []string{"b", "c"}) || o.ProbeRetrying == exhausted {
t.Fatalf("new unanswered namespace was discarded: %+v", o)
}
if c := d.NamespaceObservation(gvr, "c"); !slices.Equal(c.UnansweredNamespaces, []string{"b", "c"}) || c.ProbeRetrying == exhausted {
t.Fatalf("c was misclassified as collector scope: %+v", c)
}
d.mu.RLock()
same := d.probeRetries[gvr] == r && r.deadline.Equal(deadline)
d.mu.RUnlock()
if !same {
t.Fatal("new namespace replaced the retry owner or deadline")
}
if exhausted {
cAllowed.Store(true)
if err := d.EnsureWatching(gvr); err != nil {
t.Fatalf("new exhausted namespace did not recover on demand: %v", err)
}
} else {
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not finish")
}
}
if !d.HasCoveringInformer(gvr, "c") || !slices.Equal(d.Observation(gvr).UnansweredNamespaces, []string{"b"}) {
t.Fatalf("merged namespace did not recover independently: %+v", d.Observation(gvr))
}
d.mu.RLock()
same = d.probeRetries[gvr] == r && !r.running && r.deadline.Equal(deadline)
d.mu.RUnlock()
if !same {
t.Fatal("merged namespace recovery renewed the automatic retry")
}
})
}
}
func TestProbeRetrySharedStartupLimit(t *testing.T) {
const limit, kinds = 2, 12
var active, peak atomic.Int32
entered, release := make(chan struct{}, kinds), make(chan struct{})
var requests atomic.Int32
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
n := active.Add(1)
defer active.Add(-1)
for old := peak.Load(); n > old && !peak.CompareAndSwap(old, n); old = peak.Load() {
}
requests.Add(1)
entered <- struct{}{}
select {
case <-release:
case <-r.Context().Done():
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusForbidden)
fmt.Fprint(w, `{"kind":"Status","apiVersion":"v1","status":"Failure","reason":"Forbidden","code":403}`)
}))
t.Cleanup(srv.Close)
t.Cleanup(func() { close(release) })
client, err := dynamic.NewForConfig(&rest.Config{Host: srv.URL, QPS: 10000, Burst: 10000})
if err != nil {
t.Fatal(err)
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: client, ProbeRetryBackoff: []time.Duration{time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
d.probeSlots = make(chan struct{}, limit)
gvrs := make([]schema.GroupVersionResource, kinds)
for i := range gvrs {
gvrs[i] = schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: fmt.Sprintf("widgets%d", i)}
}
startupDone := make(chan struct{})
go func() {
d.WarmupParallel(gvrs[:limit], time.Millisecond)
close(startupDone)
}()
for range limit {
select {
case <-entered:
case <-time.After(time.Second):
t.Fatal("startup probes did not enter")
}
}
for i, gvr := range gvrs[limit:] {
o := probeOutcome{startedAt: time.Now(), whole: i%2 == 0, fullWalk: true}
if !o.whole {
o.unanswered = []string{"b"}
}
d.noteWalk(gvr, o)
}
select {
case <-entered:
t.Fatal("retry bypassed occupied startup probe slots")
case <-time.After(50 * time.Millisecond):
}
d.Stop()
select {
case <-startupDone:
case <-time.After(time.Second):
t.Fatal("startup did not cancel")
}
for _, gvr := range gvrs[limit:] {
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("queued retry did not cancel")
}
}
if peak.Load() != limit || requests.Load() != limit {
t.Fatalf("peak=%d requests=%d want=%d", peak.Load(), requests.Load(), limit)
}
}
func TestProbeRetryExplicitNamespaceOutsideFallback(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, exhausted := range []bool{false, true} {
t.Run(fmt.Sprintf("exhausted=%v", exhausted), func(t *testing.T) {
dyn := rateLimitedLists(gvr, []string{"a", "c"}, map[string]int{"": 1})
var recovered atomic.Bool
dyn.PrependReactor("list", "widgets", func(ktesting.Action) (bool, runtime.Object, error) {
if exhausted && !recovered.Load() {
return true, nil, errClientRateLimited
}
return false, nil, nil
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, NamespaceFallbacks: []string{"a"}, ProbeRetryBackoff: []time.Duration{50 * time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatchingContext(t.Context(), gvr, "c"); !IsClientRateLimited(err) {
t.Fatalf("initial explicit read: %v", err)
}
if o := d.Observation(gvr); !slices.Equal(o.UnansweredNamespaces, []string{"c"}) {
t.Fatalf("cluster-wide gap lost the requested namespace: %+v", o)
}
d.mu.RLock()
r := d.probeRetries[gvr]
deadline := r.deadline
d.mu.RUnlock()
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not finish")
}
if exhausted {
recovered.Store(true)
if err := d.EnsureWatchingContext(t.Context(), gvr, "c"); !IsClientRateLimited(err) {
t.Fatalf("first recovery probe: %v", err)
}
if err := d.EnsureWatchingContext(t.Context(), gvr, "c"); err != nil {
t.Fatal(err)
}
d.mu.RLock()
whole := r.whole
d.mu.RUnlock()
if whole || d.NamespaceObservation(gvr, "other").ReasonCode == "scope_probe_incomplete" {
t.Fatal("answered cluster-wide denial left a whole-kind gap")
}
if err := d.EnsureWatching(gvr); err != nil {
t.Fatal(err)
}
}
if !d.HasCoveringInformer(gvr, "a") || !d.HasCoveringInformer(gvr, "c") || !d.WaitForSync(gvr, time.Second) {
t.Fatal("fallback walk dropped the explicit namespace")
}
if o := d.Observation(gvr); o.ProbeRetrying || len(o.UnansweredNamespaces) > 0 || o.ReasonCode == "scope_probe_incomplete" {
t.Fatalf("scope did not recover: %+v", o)
}
d.mu.RLock()
same := d.probeRetries[gvr] == r && !r.running && r.deadline.Equal(deadline)
d.mu.RUnlock()
if !same {
t.Fatal("outside-fallback recovery renewed the retry")
}
})
}
}
func TestProbeRetryPreservesPriorFullWalk(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
dyn := rateLimitedLists(gvr, []string{"a", "c"}, map[string]int{"c": 1})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, NamespaceFallbacks: []string{"a"}, ProbeRetryBackoff: []time.Duration{50 * time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); err != nil || !d.WaitForSync(gvr, time.Second) {
t.Fatalf("initial full walk: %v", err)
}
before := len(dyn.Actions())
if err := d.EnsureWatchingContext(t.Context(), gvr, "c"); !IsClientRateLimited(err) {
t.Fatalf("explicit namespace probe: %v", err)
}
d.mu.RLock()
r := d.probeRetries[gvr]
fullWalk := r.fullWalk
d.mu.RUnlock()
if !fullWalk {
t.Fatal("explicit gap discarded the already-completed fallback walk")
}
select {
case <-r.done:
case <-time.After(time.Second):
t.Fatal("retry did not finish")
}
if !d.HasCoveringInformer(gvr, "c") || !d.WaitForSync(gvr, time.Second) {
t.Fatal("explicit gap did not recover")
}
for _, action := range dyn.Actions()[before:] {
if action.GetVerb() == "list" && action.GetNamespace() == "a" {
t.Fatal("explicit retry needlessly repeated the settled fallback walk")
}
}
}
func TestProbeRetrySingleFallbackCountsRemainUnavailable(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, exhausted := range []bool{false, true} {
t.Run(fmt.Sprintf("exhausted=%v", exhausted), func(t *testing.T) {
backoff, refused := time.Hour, 1
if exhausted {
backoff, refused = time.Millisecond, 2
}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, []string{"a", "c"}, map[string]int{"c": refused}), NamespaceFallbacks: []string{"a"}, ProbeRetryBackoff: []time.Duration{backoff}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); err != nil || !d.WaitForSync(gvr, time.Second) {
t.Fatalf("initial fallback sync: %v", err)
}
if n, err := d.Count(gvr, nil); err != nil || n != 0 {
t.Fatalf("initial count: %d, %v", n, err)
}
if err := d.EnsureWatchingContext(t.Context(), gvr, "c"); !IsClientRateLimited(err) {
t.Fatalf("explicit namespace probe: %v", err)
}
if exhausted {
d.mu.RLock()
done := d.probeRetries[gvr].done
d.mu.RUnlock()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("retry did not exhaust")
}
}
if o := d.Observation(gvr); o.ProbeRetrying == exhausted || !slices.Equal(o.UnansweredNamespaces, []string{"c"}) {
t.Fatalf("missing unresolved scope: %+v", o)
}
_, countErr := d.Count(gvr, nil)
_, counted := d.CountWatched(nil)[gvr]
if countErr == nil || counted {
t.Fatalf("unanswered c accepted as authoritative: Count error=%v, CountWatched present=%v", countErr, counted)
}
if n, err := d.Count(gvr, []string{"a"}); err != nil || n != 0 {
t.Fatalf("answered namespace count unavailable: %d, %v", n, err)
}
if n, ok := d.CountWatched([]string{"a"})[gvr]; !ok || n != 0 {
t.Fatalf("answered namespace missing from watched counts: %d, %v", n, ok)
}
if err := d.EnsureWatchingContext(t.Context(), gvr, "c"); err != nil || !d.WaitForSync(gvr, time.Second) {
t.Fatalf("explicit scope did not recover: %v", err)
}
if n, err := d.Count(gvr, nil); err != nil || n != 0 {
t.Fatalf("recovered count unavailable: %d, %v", n, err)
}
if n, ok := d.CountWatched(nil)[gvr]; !ok || n != 0 {
t.Fatalf("recovered watched count unavailable: %d, %v", n, ok)
}
})
}
}
func TestProbeStartupQueueDoesNotConsumeWalkBudget(t *testing.T) {
const limit, kinds = 2, 12
gvrs := make([]schema.GroupVersionResource, kinds)
listKinds := map[schema.GroupVersionResource]string{}
for i := range gvrs {
gvrs[i] = schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: fmt.Sprintf("widgets%d", i)}
listKinds[gvrs[i]] = "WidgetList"
}
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), listKinds)
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
d.probeSlots = make(chan struct{}, limit)
for range limit {
d.probeSlots <- struct{}{}
}
results := make(chan probeOutcome, kinds)
for _, gvr := range gvrs {
go func() { results <- d.walkScopes(gvr, "") }()
}
select {
case <-results:
t.Fatal("queued walk bypassed startup admission")
case <-time.After(50 * time.Millisecond):
}
releasedAt := time.Now()
for range limit {
<-d.probeSlots
}
for range kinds {
select {
case o := <-results:
if o.startedAt.Before(releasedAt) || !o.complete || o.whole || o.err != nil {
t.Fatalf("queue time consumed the walk budget: %+v release=%v", o, releasedAt)
}
case <-time.After(time.Second):
t.Fatal("queued startup walk did not answer")
}
}
}
func TestProbeRetrySingleNamespaceAnswerDoesNotSettle(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, []string{"a", "b"}, map[string]int{"b": 1}), NamespaceFallbacks: []string{"a", "b"}, ProbeRetryBackoff: []time.Duration{500 * time.Millisecond}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatchingContext(t.Context(), gvr, "b"); !IsClientRateLimited(err) {
t.Fatalf("initial explicit read: %v", err)
}
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
if err := d.EnsureWatchingContext(t.Context(), gvr, "b"); err != nil {
t.Fatal(err)
}
if d.HasCoveringInformer(gvr, "") || !d.Observation(gvr).ProbeRetrying {
t.Fatalf("one answered namespace settled the fallback scope: %+v", d.Observation(gvr))
}
select {
case <-r.done:
case <-time.After(2 * time.Second):
t.Fatal("full fallback walk did not finish")
}
if !d.HasCoveringInformer(gvr, "a") || !d.HasCoveringInformer(gvr, "b") || !d.WaitForSync(gvr, time.Second) {
t.Fatal("explicit answer prevented the remaining fallback walk")
}
}
// Checks still unanswered when the retry budget runs out keep being re-checked,
// slowly, so a reader of the cache alone gets the kind once the API answers.
// Each re-check walks as a first probe would: an identity limited to some
// namespaces falls back to them instead of watching cluster-wide on a guess.
func TestProbeRecoveryAfterTheBudget(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, tc := range []struct {
name string
listable []string
throttled map[string]int
fallbacks []string
pinned string
watched string
}{
{"cluster-wide identity", []string{""}, map[string]int{"": 3}, nil, "", ""},
{"identity limited to a namespace", []string{"a"}, map[string]int{"": 3}, []string{"a"}, "", "a"},
{"pinned namespace", []string{"shop"}, map[string]int{"shop": 3}, nil, "shop", "shop"},
} {
t.Run(tc.name, func(t *testing.T) {
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: rateLimitedLists(gvr, tc.listable, tc.throttled),
NamespaceFallbacks: tc.fallbacks,
NamespaceScoped: tc.pinned != "",
Namespace: tc.pinned,
ProbeRetryBackoff: []time.Duration{10 * time.Millisecond},
ProbeRecoveryIntervals: []time.Duration{20 * time.Millisecond},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
_ = d.EnsureWatching(gvr)
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
if r == nil {
t.Fatal("no retry started")
}
select {
case <-r.done:
case <-time.After(3 * time.Second):
t.Fatal("retry did not finish")
}
if o := d.Observation(gvr); o.ProbeRetrying {
t.Fatalf("an exhausted budget must not read as still retrying: %+v", o)
}
deadline := time.Now().Add(3 * time.Second)
for !d.HasCoveringInformer(gvr, tc.watched) && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
if !d.HasCoveringInformer(gvr, tc.watched) {
t.Fatalf("recovery never watched the kind: %+v", d.Observation(gvr))
}
if tc.watched != "" && d.HasCoveringInformer(gvr, "elsewhere") {
t.Fatal("recovery watched cluster-wide instead of the namespace")
}
})
}
}
// A retry record that settled has no owner left, so a gap a later read finds
// starts a fresh retry instead of waiting forever.
func TestProbeRetryRearmsAfterSettling(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: rateLimitedLists(gvr, []string{"a", "c"}, map[string]int{"a": 2, "c": 2}),
NamespaceFallbacks: []string{"a"},
ProbeRetryBackoff: []time.Duration{10 * time.Millisecond, 10 * time.Millisecond},
ProbeRecoveryIntervals: []time.Duration{time.Hour},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
_ = d.EnsureWatching(gvr)
d.mu.RLock()
first := d.probeRetries[gvr]
d.mu.RUnlock()
if first == nil {
t.Fatal("no retry for the throttled fallback namespace")
}
<-first.done
d.mu.RLock()
settled := first.settled()
d.mu.RUnlock()
if !settled || !d.HasCoveringInformer(gvr, "a") {
t.Fatalf("first retry didn't settle on namespace a: %+v", d.Observation(gvr))
}
_ = d.EnsureWatchingContext(t.Context(), gvr, "c")
deadline := time.Now().Add(3 * time.Second)
for !d.HasCoveringInformer(gvr, "c") && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
d.mu.RLock()
rearmed := d.probeRetries[gvr] != first
d.mu.RUnlock()
if !rearmed || !d.HasCoveringInformer(gvr, "c") {
t.Fatalf("a gap found after settling was never retried: rearmed=%v %+v", rearmed, d.Observation(gvr))
}
if ownerLive(d, first) {
t.Fatal("the settled record's owner outlived it")
}
}
// Stopping the cache ends a recovery waiting for its next re-check, which
// would otherwise hold the stopped cache until then.
func TestProbeRecoveryStopsWithTheCache(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: rateLimitedLists(gvr, []string{""}, map[string]int{"": 100}),
ProbeRetryBackoff: []time.Duration{10 * time.Millisecond},
ProbeRecoveryIntervals: []time.Duration{time.Hour},
})
if err != nil {
t.Fatal(err)
}
_ = d.EnsureWatching(gvr)
d.mu.RLock()
r := d.probeRetries[gvr]
d.mu.RUnlock()
<-r.done
if !ownerLive(d, r) {
t.Fatal("no recovery after the budget ran out")
}
d.Stop()
deadline := time.Now().Add(time.Second)
for ownerLive(d, r) && time.Now().Before(deadline) {
time.Sleep(5 * time.Millisecond)
}
if ownerLive(d, r) {
t.Fatal("the recovery outlived Stop")
}
}
func ownerLive(d *DynamicResourceCache, r *probeRetry) bool {
d.mu.RLock()
defer d.mu.RUnlock()
return r.owned
}
// A cluster-wide watch started on a failed check is refused, under the same
// lock that would replace them, while namespace watches exist.
func TestFailedOpenWatchKeepsNamespaceWatches(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, []string{"", "a"}, nil)})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.startWatching(gvr, "a"); err != nil {
t.Fatal(err)
}
if err := d.startWatchingScope(gvr, "", true); !errors.Is(err, errNamespaceWatchesKept) {
t.Fatalf("failed-open cluster-wide watch = %v", err)
}
if !d.HasCoveringInformer(gvr, "a") || d.HasCoveringInformer(gvr, "elsewhere") {
t.Fatal("the namespace watch was replaced")
}
if err := d.startWatchingScope(gvr, "", false); err != nil || !d.HasCoveringInformer(gvr, "elsewhere") {
t.Fatalf("a passed cluster-wide check must still replace namespace watches: %v", err)
}
}
// A recovery walk whose cluster-wide check fails rather than passes (here a
// 503 from a proxy) must not replace namespace watches that work with a
// cluster-wide one.
func TestProbeRecoveryKeepsNamespaceWatchesOnAFailedCheck(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
var mu sync.Mutex
clusterCalls := 0
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{gvr: "WidgetList"})
dyn.PrependReactor("list", "widgets", func(action ktesting.Action) (bool, runtime.Object, error) {
if action.GetNamespace() != "" {
return false, nil, nil
}
mu.Lock()
defer mu.Unlock()
clusterCalls++
if clusterCalls <= 2 {
return true, nil, errClientRateLimited
}
return true, nil, apierrors.NewServiceUnavailable("proxy unavailable")
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: dyn,
ProbeRetryBackoff: []time.Duration{50 * time.Millisecond},
ProbeRecoveryIntervals: []time.Duration{10 * time.Millisecond},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
_ = d.EnsureWatching(gvr)
// A read started a namespace watch before the background retries ran.
if err := d.startWatching(gvr, "a"); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(500 * time.Millisecond)
for time.Now().Before(deadline) {
mu.Lock()
calls := clusterCalls
mu.Unlock()
if calls >= 5 {
break
}
time.Sleep(10 * time.Millisecond)
}
mu.Lock()
calls := clusterCalls
mu.Unlock()
if calls < 5 {
t.Fatalf("recovery made %d cluster-wide checks, so the failed ones were never tried", calls)
}
d.mu.RLock()
_, clusterWide := d.informers[informerKey{gvr: gvr}]
d.mu.RUnlock()
if clusterWide || !d.HasCoveringInformer(gvr, "a") {
t.Fatalf("a failed cluster-wide check replaced the namespace watch: clusterWide=%v", clusterWide)
}
}
// A retained credential rejection is never reported as an unfinished check,
// whatever the retry record still holds.
func TestObservationKeepsCredentialRejection(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, nil, nil)})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
d.retainObservation(gvr, DynamicObservationDenied, "credentials_rejected", false)
d.mu.Lock()
d.probeRetries[gvr] = &probeRetry{whole: true}
d.mu.Unlock()
if o := d.Observation(gvr); o.ReasonCode != "credentials_rejected" {
t.Fatalf("observation = %+v", o)
}
}
// A new gap on a record whose owner can no longer take it, because the owner
// ended or met rejected credentials, gets a fresh owner that also takes over
// every gap still outstanding.
func TestProbeRetryRearmTakesOverOutstandingGaps(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
for _, tc := range []struct {
name string
owned, rejected bool
}{
{"owner ended", false, false},
{"owner rejected but not yet released", true, true},
} {
t.Run(tc.name, func(t *testing.T) {
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: rateLimitedLists(gvr, nil, map[string]int{"b": 100, "c": 100, "d": 100}),
NamespaceFallbacks: []string{"a", "b", "c"},
ProbeRetryBackoff: []time.Duration{time.Hour},
ProbeRecoveryIntervals: []time.Duration{time.Hour},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
ctx, cancel := context.WithCancel(t.Context())
t.Cleanup(cancel)
prev := &probeRetry{unanswered: []string{"b", "c"}, fullWalk: true, owned: tc.owned, rejected: tc.rejected, answered: map[string]time.Time{}, ctx: ctx, cancel: cancel, done: make(chan struct{})}
d.mu.Lock()
d.probeRetries[gvr] = prev
d.mu.Unlock()
d.noteWalk(gvr, probeOutcome{unanswered: []string{"d"}, startedAt: time.Now()})
d.mu.RLock()
r := d.probeRetries[gvr]
gaps := slices.Sorted(slices.Values(r.unanswered))
live := r.owned && !r.rejected
d.mu.RUnlock()
if r == prev || !live || !slices.Equal(gaps, []string{"b", "c", "d"}) {
t.Fatalf("re-armed=%v live=%v gaps=%v, want a fresh owner of b, c and d", r != prev, live, gaps)
}
if ctx.Err() == nil {
t.Fatal("the replaced record's owner was not ended")
}
})
}
}
// A 401 met while namespaces already have watches: those keep their own
// state, but the kind and the namespaces still outstanding read as rejected
// credentials, not as checks that never answered.
func TestObservationShowsRejectionBesideWatchedNamespaces(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(gvr, []string{"a"}, nil)})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.startWatching(gvr, "a"); err != nil {
t.Fatal(err)
}
d.mu.Lock()
d.probeRetries[gvr] = &probeRetry{unanswered: []string{"b"}, fullWalk: true, rejected: true}
d.mu.Unlock()
if o := d.Observation(gvr); o.ReasonCode != "credentials_rejected" {
t.Fatalf("kind observation = %+v, want credentials_rejected", o)
}
if o := d.NamespaceObservation(gvr, "b"); o.ReasonCode != "credentials_rejected" {
t.Fatalf("observation of b = %+v, want credentials_rejected", o)
}
if o := d.NamespaceObservation(gvr, "a"); o.ReasonCode == "credentials_rejected" {
t.Fatalf("observation of watched a = %+v, want its own watch state", o)
}
}
// Under --namespace-scope, a pinned namespace whose check never answered is
// an unfinished check, as it is for a fallback namespace, not a failed one.
func TestPinnedNamespaceUnansweredIsIncomplete(t *testing.T) {
gvr := schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: rateLimitedLists(gvr, []string{"team"}, map[string]int{"team": 100}),
NamespaceScoped: true,
Namespace: "team",
ProbeRetryBackoff: []time.Duration{time.Hour},
ProbeRecoveryIntervals: []time.Duration{time.Hour},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
d.WarmupParallel([]schema.GroupVersionResource{gvr}, 5*time.Second)
o := d.Observation(gvr)
if o.ReasonCode != "scope_probe_incomplete" || !slices.Equal(o.UnansweredNamespaces, []string{"team"}) {
t.Fatalf("observation = %+v, want scope_probe_incomplete with team unanswered", o)
}
}
// Server discovery can list a cluster-scoped kind's deprecated version first.
// Neither the version a caller resolves nor the scope the access walk decides
// may depend on that order: a namespace probe of a cluster-scoped kind gets a
// 404, which the walk would read as access.
func TestClusterScopedKindDiscoveredDeprecatedVersionFirstStaysClusterScoped(t *testing.T) {
const group = "kueue.x-k8s.io"
fakeDiscovery := fake.NewClientset().Discovery().(*fakediscovery.FakeDiscovery)
fakeDiscovery.Resources = []*metav1.APIResourceList{
{GroupVersion: group + "/v1beta1", APIResources: []metav1.APIResource{{Name: "clusterqueues", Kind: "ClusterQueue", Namespaced: false, Verbs: []string{"get", "list", "watch"}}}},
{GroupVersion: group + "/v1beta2", APIResources: []metav1.APIResource{{Name: "clusterqueues", Kind: "ClusterQueue", Namespaced: false, Verbs: []string{"get", "list", "watch"}}}},
}
discovery, err := NewResourceDiscovery(fakeDiscovery)
if err != nil {
t.Fatal(err)
}
if res, ok := discovery.GetResourceWithGroup("ClusterQueue", group); !ok || res.Version != "v1beta2" || res.Namespaced {
t.Fatalf("resolved %+v ok=%v, want cluster-scoped v1beta2", res, ok)
}
for _, version := range []string{"v1beta1", "v1beta2"} {
t.Run(version, func(t *testing.T) {
gvr := schema.GroupVersionResource{Group: group, Version: version, Resource: "clusterqueues"}
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{gvr: "ClusterQueueList"})
dyn.PrependReactor("list", "clusterqueues", func(action ktesting.Action) (bool, runtime.Object, error) {
if action.GetNamespace() == "" {
return true, nil, apierrors.NewForbidden(gvr.GroupResource(), "", errors.New("cluster-wide list refused"))
}
return true, nil, apierrors.NewNotFound(gvr.GroupResource(), "")
})
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: dyn, Discovery: discovery, NamespaceFallbacks: []string{"team-a"}})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(gvr); err == nil {
t.Fatal("EnsureWatching succeeded for a cluster-scoped kind refused cluster-wide")
}
if d.HasCoveringInformer(gvr, "team-a") {
t.Fatal("started a namespaced informer for a cluster-scoped kind")
}
if got := d.Observation(gvr); got.State != DynamicObservationDenied || got.Scope == DynamicObservationScopeExplicitNamespaces {
t.Fatalf("observation %+v, want a kind-level denial", got)
}
})
}
}
// Each served version reads its own scope, whichever is the most stable, and a
// version discovery doesn't serve reads as namespaced rather than borrowing
// another version's scope.
func TestGvrIsNamespacedReadsTheRequestedVersion(t *testing.T) {
verbs := []string{"get", "list", "watch"}
fakeDiscovery := fake.NewClientset().Discovery().(*fakediscovery.FakeDiscovery)
fakeDiscovery.Resources = []*metav1.APIResourceList{
{GroupVersion: "widgets.example.io/v1beta1", APIResources: []metav1.APIResource{{Name: "widgets", Kind: "Widget", Namespaced: true, Verbs: verbs}}},
{GroupVersion: "widgets.example.io/v1", APIResources: []metav1.APIResource{{Name: "widgets", Kind: "Widget", Namespaced: false, Verbs: verbs}}},
{GroupVersion: "gadgets.example.io/v1beta1", APIResources: []metav1.APIResource{{Name: "gadgets", Kind: "Gadget", Namespaced: false, Verbs: verbs}}},
{GroupVersion: "gadgets.example.io/v1", APIResources: []metav1.APIResource{{Name: "gadgets", Kind: "Gadget", Namespaced: true, Verbs: verbs}}},
}
discovery, err := NewResourceDiscovery(fakeDiscovery)
if err != nil {
t.Fatal(err)
}
d := &DynamicResourceCache{config: DynamicCacheConfig{Discovery: discovery}}
for gvr, want := range map[schema.GroupVersionResource]bool{
{Group: "widgets.example.io", Version: "v1beta1", Resource: "widgets"}: true,
{Group: "widgets.example.io", Version: "v1", Resource: "widgets"}: false,
{Group: "widgets.example.io", Version: "v2", Resource: "widgets"}: true,
{Group: "gadgets.example.io", Version: "v1beta1", Resource: "gadgets"}: false,
{Group: "gadgets.example.io", Version: "v1", Resource: "gadgets"}: true,
} {
if got := d.gvrIsNamespaced(gvr); got != want {
t.Errorf("%s: gvrIsNamespaced=%v, want %v", gvr, got, want)
}
}
}
// A resource replaced under the same name with a different scope reads at its
// new scope once discovery is stale, so a cluster-scoped answer can't outlive
// the API and drop a namespace filter.
func TestGvrIsNamespacedRefreshesStaleDiscovery(t *testing.T) {
verbs := []string{"get", "list", "watch"}
gvr := schema.GroupVersionResource{Group: "widgets.example.io", Version: "v1", Resource: "widgets"}
fakeDiscovery := fake.NewClientset().Discovery().(*fakediscovery.FakeDiscovery)
fakeDiscovery.Resources = []*metav1.APIResourceList{
{GroupVersion: "widgets.example.io/v1", APIResources: []metav1.APIResource{{Name: "widgets", Kind: "Widget", Namespaced: false, Verbs: verbs}}},
}
discovery, err := NewResourceDiscovery(fakeDiscovery, WithDiscoveryCacheTTL(time.Nanosecond))
if err != nil {
t.Fatal(err)
}
d := &DynamicResourceCache{config: DynamicCacheConfig{Discovery: discovery}}
if d.gvrIsNamespaced(gvr) {
t.Fatal("cluster-scoped widgets read as namespaced")
}
fakeDiscovery.Resources = []*metav1.APIResourceList{
{GroupVersion: "widgets.example.io/v1", APIResources: []metav1.APIResource{{Name: "widgets", Kind: "Widget", Namespaced: true, Verbs: verbs}}},
}
time.Sleep(time.Millisecond)
if !d.gvrIsNamespaced(gvr) {
t.Fatal("widgets replaced as namespaced still read as cluster-scoped after discovery went stale")
}
}