pkg/k8score/dynamic_cache_denial_test.go
Rama main · réplica de lectura del repositorio original.
package k8score
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"slices"
"strings"
"sync/atomic"
"testing"
"time"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/dynamic"
dynamicfake "k8s.io/client-go/dynamic/fake"
"k8s.io/client-go/rest"
k8stesting "k8s.io/client-go/testing"
)
var deniedWidgets = schema.GroupVersionResource{Group: "example.io", Version: "v1", Resource: "widgets"}
// countedWidgetLists answers every list with answer(namespace) and counts the
// calls. A nil answer falls through to an empty list.
func countedWidgetLists(calls *atomic.Int64, answer func(ns string) error) *dynamicfake.FakeDynamicClient {
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{deniedWidgets: "WidgetList"})
dyn.PrependReactor("list", "widgets", func(action k8stesting.Action) (bool, runtime.Object, error) {
calls.Add(1)
if err := answer(action.GetNamespace()); err != nil {
return true, nil, err
}
return false, nil, nil
})
return dyn
}
func forbidden(string) error {
return apierrors.NewForbidden(deniedWidgets.GroupResource(), "", errors.New("collector cannot list"))
}
func deniedCache(t *testing.T, calls *atomic.Int64, answer func(string) error) *DynamicResourceCache {
t.Helper()
d, err := NewDynamicResourceCache(DynamicCacheConfig{
DynamicClient: countedWidgetLists(calls, answer),
NamespaceFallbacks: []string{"a", "b"},
ProbeRetryBackoff: []time.Duration{time.Hour},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
return d
}
func expireDenials(d *DynamicResourceCache) {
d.mu.Lock()
defer d.mu.Unlock()
for key, denial := range d.deniedScopes {
denial.at = denial.at.Add(-deniedScopeTTL)
d.deniedScopes[key] = denial
}
}
func TestCompleteDenialAnswersRepeatedReads(t *testing.T) {
var calls atomic.Int64
d := deniedCache(t, &calls, forbidden)
first := d.EnsureWatching(deniedWidgets)
if !apierrors.IsForbidden(first) {
t.Fatalf("first read: %v", first)
}
if got := calls.Load(); got != 3 {
t.Fatalf("first read made %d checks, want the cluster-wide one and two namespaces", got)
}
observed := d.Observation(deniedWidgets)
for range 5 {
if _, err := d.List(deniedWidgets, ""); !apierrors.IsForbidden(err) || err.Error() != first.Error() {
t.Fatalf("repeated read returned %v, want %v", err, first)
}
}
if got := calls.Load(); got != 3 {
t.Fatalf("repeated reads walked again: %d checks", got)
}
if got := d.Observation(deniedWidgets); got.State != observed.State || !got.ObservedAt.Equal(*observed.ObservedAt) {
t.Fatalf("a remembered answer changed the observation: %+v, was %+v", got, observed)
}
}
func TestDenialExpiryFindsNewGrant(t *testing.T) {
for _, tc := range []struct {
name string
granted string
want string
}{
{"cluster-wide", "", ""},
{"one candidate namespace", "b", "b"},
} {
t.Run(tc.name, func(t *testing.T) {
var calls, access atomic.Int64
d := deniedCache(t, &calls, func(ns string) error {
if access.Load() == 1 && ns == tc.granted {
return nil
}
return forbidden(ns)
})
if err := d.EnsureWatching(deniedWidgets); !apierrors.IsForbidden(err) {
t.Fatalf("before the grant: %v", err)
}
access.Store(1)
if err := d.EnsureWatching(deniedWidgets); !apierrors.IsForbidden(err) {
t.Fatalf("within the TTL: %v", err)
}
expireDenials(d)
if err := d.EnsureWatching(deniedWidgets); err != nil {
t.Fatalf("after the TTL: %v", err)
}
if !d.HasCoveringInformer(deniedWidgets, tc.want) {
t.Fatalf("no watch of %q after the grant", tc.want)
}
})
}
}
// Only a walk whose every check was answered Forbidden is a denial. Each of
// these leaves the next read free to ask again.
func TestOnlyCompleteRefusalsAreRemembered(t *testing.T) {
unauthorized := apierrors.NewUnauthorized("token expired")
for _, tc := range []struct {
name string
answer func(ns string) error
}{
{"namespace rate limited", func(ns string) error {
if ns == "b" {
return errClientRateLimited
}
return forbidden(ns)
}},
{"namespace timed out", func(ns string) error {
if ns == "b" {
return context.DeadlineExceeded
}
return forbidden(ns)
}},
{"cluster-wide check timed out", func(string) error { return context.DeadlineExceeded }},
{"credentials rejected", func(string) error { return unauthorized }},
{"credentials rejected in a namespace", func(ns string) error {
if ns == "a" {
return unauthorized
}
return forbidden(ns)
}},
} {
t.Run(tc.name, func(t *testing.T) {
var calls atomic.Int64
d := deniedCache(t, &calls, tc.answer)
if err := d.EnsureWatching(deniedWidgets); err == nil {
t.Fatal("expected an error")
}
if err := d.recentDenial(deniedWidgets, ""); err != nil {
t.Fatalf("remembered as a denial: %v", err)
}
})
}
}
// A walk for all namespaces only reached its candidates, and a refused
// namespace says nothing about the others.
func TestDenialCoversOnlyTheScopeWalked(t *testing.T) {
var calls atomic.Int64
d := deniedCache(t, &calls, forbidden)
if err := d.EnsureWatching(deniedWidgets); !apierrors.IsForbidden(err) {
t.Fatal(err)
}
if _, err := d.List(deniedWidgets, "outside"); !apierrors.IsForbidden(err) {
t.Fatalf("namespace read: %v", err)
}
if got := calls.Load(); got != 5 {
t.Fatalf("%d checks, want the namespace read's own two after the walk's three", got)
}
if _, err := d.List(deniedWidgets, "outside"); !apierrors.IsForbidden(err) {
t.Fatalf("repeated namespace read: %v", err)
}
if got := calls.Load(); got != 5 {
t.Fatalf("repeated namespace read walked again: %d checks", got)
}
}
// The startup passes record a kind's refusal before anything reads it, so the
// first read needs no walk of its own.
func TestStartupDenialsAnswerTheFirstRead(t *testing.T) {
for name, startup := range map[string]func(d *DynamicResourceCache){
"warmup": func(d *DynamicResourceCache) {
d.WarmupParallel([]schema.GroupVersionResource{deniedWidgets}, time.Millisecond)
},
"small CRD scan": func(d *DynamicResourceCache) {
d.watchSmallCRDs([]schema.GroupVersionResource{deniedWidgets})
},
} {
t.Run(name, func(t *testing.T) {
var calls atomic.Int64
d := deniedCache(t, &calls, forbidden)
startup(d)
before := calls.Load()
if err := d.EnsureWatching(deniedWidgets); !apierrors.IsForbidden(err) || !strings.Contains(err.Error(), "no access to widgets.example.io/v1") {
t.Fatalf("first read: %v", err)
}
if got := calls.Load(); got != before {
t.Fatalf("first read walked again: %d checks after %d", got, before)
}
})
}
}
// A later check that found something other than a refusal replaces the memory
// of one, the same way it replaces the retained observation.
func TestLaterIncompleteCheckClearsDenial(t *testing.T) {
var calls, slow atomic.Int64
d := deniedCache(t, &calls, func(ns string) error {
if slow.Load() == 1 {
return context.DeadlineExceeded
}
return forbidden(ns)
})
if err := d.EnsureWatching(deniedWidgets); !apierrors.IsForbidden(err) {
t.Fatal(err)
}
slow.Store(1)
d.WarmupParallel([]schema.GroupVersionResource{deniedWidgets}, time.Millisecond)
if err := d.recentDenial(deniedWidgets, ""); err != nil {
t.Fatalf("denial survived an unanswered check: %v", err)
}
}
func TestStoppedCacheServesNoDenial(t *testing.T) {
var calls atomic.Int64
d := deniedCache(t, &calls, forbidden)
if err := d.EnsureWatching(deniedWidgets); !apierrors.IsForbidden(err) {
t.Fatal(err)
}
d.Stop()
if err := d.recentDenial(deniedWidgets, ""); err != nil {
t.Fatalf("stopped cache answered from memory: %v", err)
}
}
// candidateBarrierClient refuses every list. Namespace checks wait until want
// of them are in flight, or for patience, and peak is the most it saw at once.
// It is an HTTP server because the fake dynamic client runs reactors one at a
// time.
func candidateBarrierClient(t *testing.T, want int32, patience time.Duration) (client dynamic.Interface, peak *atomic.Int32) {
t.Helper()
var inFlight atomic.Int32
peak = &atomic.Int32{}
full := make(chan struct{})
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if strings.Contains(r.URL.Path, "/namespaces/") {
n := inFlight.Add(1)
defer inFlight.Add(-1)
for old := peak.Load(); n > old && !peak.CompareAndSwap(old, n); old = peak.Load() {
}
if n == want {
close(full)
}
select {
case <-full:
case <-time.After(patience):
}
}
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)
client, err := dynamic.NewForConfig(&rest.Config{Host: srv.URL, QPS: 10000, Burst: 10000})
if err != nil {
t.Fatal(err)
}
return client, peak
}
func TestSpareProbeSlotsCheckCandidatesAtOnce(t *testing.T) {
namespaces := []string{"a", "b", "c", "d", "e", "f", "g", "h"}
client, peak := candidateBarrierClient(t, int32(len(namespaces)), time.Second)
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: client, NamespaceFallbacks: namespaces})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
if err := d.EnsureWatching(deniedWidgets); !apierrors.IsForbidden(err) {
t.Fatal(err)
}
if got := peak.Load(); got != int32(len(namespaces)) {
t.Fatalf("%d candidates in flight at once, want %d", got, len(namespaces))
}
if err := d.recentDenial(deniedWidgets, ""); err == nil {
t.Fatal("a walk checked at once did not end in a complete denial")
}
}
// Other walks holding the slots leave this one to its own: it never adds to
// what is running, so a busy startup stays within the shared limit.
func TestBusyProbeSlotsCheckCandidatesInTurn(t *testing.T) {
namespaces := []string{"a", "b", "c"}
client, peak := candidateBarrierClient(t, int32(len(namespaces)), 5*time.Millisecond)
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: client, NamespaceFallbacks: namespaces})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
d.probeSlots = make(chan struct{}, 1)
if err := d.EnsureWatching(deniedWidgets); !apierrors.IsForbidden(err) {
t.Fatal(err)
}
if got := peak.Load(); got != 1 {
t.Fatalf("%d candidates in flight with no spare slot, want 1", got)
}
}
// Candidates finish in any order; the scopes a walk returns must not.
func TestConcurrentWalkKeepsCandidateOrder(t *testing.T) {
namespaces := []string{"e", "d", "c", "b", "a"}
d, err := NewDynamicResourceCache(DynamicCacheConfig{DynamicClient: rateLimitedLists(deniedWidgets, namespaces, nil), NamespaceFallbacks: namespaces})
if err != nil {
t.Fatal(err)
}
t.Cleanup(d.Stop)
scopes, complete, err := d.probeScopes(deniedWidgets, "")
if err != nil || !complete || !slices.Equal(scopes, namespaces) {
t.Fatalf("scopes %v complete %v err %v, want %v", scopes, complete, err, namespaces)
}
}