pkg/ownership/cachesource/lifecycle_test.go
Rama main · réplica de lectura del repositorio original.
package cachesource
import (
"context"
"fmt"
"math/rand/v2"
"slices"
"testing"
"time"
apimeta "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/types"
"github.com/skyhook-io/radar/pkg/k8score"
"github.com/skyhook-io/radar/pkg/ownership"
"github.com/skyhook-io/radar/pkg/ownership/ownershiptest"
"github.com/skyhook-io/radar/pkg/resourceid"
)
// driver changes a rendered World and the World alike, so after each change
// the source can be checked against Scan of the World.
type driver struct {
t *testing.T
r *rendered
w ownershiptest.World
rev int
}
func (d *driver) resource(o ownershiptest.Object) (ownershiptest.Kind, schema.GroupVersionResource) {
k, _ := ownershiptest.KindOf(o.Ref.GroupKind())
return k, schema.GroupVersionResource{Group: k.Group, Version: k.Version, Resource: k.Resource}
}
func (d *driver) nextVersion() string {
d.rev++
return fmt.Sprint(1000 + d.rev)
}
// apply makes the API server change o: through the typed fake's tracker for
// typed kinds, so it keeps the World's resource version, and through the
// dynamic client otherwise.
func (d *driver) apply(verb string, o ownershiptest.Object) {
d.t.Helper()
ctx := context.Background()
obj, typed := apiObject(d.t, o)
_, gvr := d.resource(o)
var err error
if typed {
tracker := d.r.typed.Tracker()
switch verb {
case "create":
err = tracker.Create(gvr, obj, o.Ref.Namespace)
case "update":
err = tracker.Update(gvr, obj, o.Ref.Namespace)
case "delete":
err = tracker.Delete(gvr, o.Ref.Namespace, o.Ref.Name)
}
} else {
client := d.r.dynamic.Resource(gvr).Namespace(o.Ref.Namespace)
u := obj.(*unstructured.Unstructured)
switch verb {
case "create":
_, err = client.Create(ctx, u, metav1.CreateOptions{})
case "update":
_, err = client.Update(ctx, u, metav1.UpdateOptions{})
case "delete":
err = client.Delete(ctx, o.Ref.Name, metav1.DeleteOptions{})
}
}
if err != nil {
d.t.Fatalf("%s %v: %v", verb, o.Ref, err)
}
}
func (d *driver) index(ref resourceid.Ref) int {
return slices.IndexFunc(d.w.Objects, func(o ownershiptest.Object) bool { return o.Ref == ref })
}
func (d *driver) create(o ownershiptest.Object) {
o.ResourceVersion = d.nextVersion()
d.apply("create", o)
d.w.Objects = append(d.w.Objects, o)
}
func (d *driver) update(ref resourceid.Ref, change func(*ownershiptest.Object)) {
i := d.index(ref)
o := d.w.Objects[i]
o.OwnerReferences = slices.Clone(o.OwnerReferences)
change(&o)
o.ResourceVersion = d.nextVersion()
d.apply("update", o)
d.w.Objects[i] = o
}
func (d *driver) delete(ref resourceid.Ref) {
i := d.index(ref)
d.apply("delete", d.w.Objects[i])
d.w.Objects = slices.Delete(d.w.Objects, i, i+1)
}
// fence waits until every store Radar has loaded holds exactly the World's
// objects of its kind and namespaces, by UID and resource version.
func (d *driver) fence() {
d.t.Helper()
deadline := time.Now().Add(5 * time.Second)
for {
stale := d.staleKind()
if stale == "" {
return
}
if time.Now().After(deadline) {
d.t.Fatalf("fence: %s never caught up", stale)
}
time.Sleep(5 * time.Millisecond)
}
}
func (d *driver) staleKind() string {
for _, k := range ownershiptest.Kinds {
c := d.w.Collection[k.GroupKind]
if !k.HasGCVerbs || c.State != ownershiptest.Collected {
continue
}
want := map[string]string{}
for _, o := range d.w.Objects {
if o.Ref.GroupKind() == k.GroupKind && (c.Namespaces == nil || !k.Namespaced || slices.Contains(c.Namespaces, o.Ref.Namespace)) {
want[o.Ref.Namespace+"/"+o.Ref.Name] = string(o.UID) + "@" + o.ResourceVersion
}
}
got := map[string]string{}
var items []any
if k.Typed {
objs, ok := d.r.caches.Typed.ReadableObjects(k.Resource, nil)
if !ok {
return k.GroupKind.String() + " (not loaded)"
}
for _, o := range objs {
items = append(items, o)
}
} else {
// Every watch the kind is collected through must have loaded,
// so an empty store isn't mistaken for a missing one.
gvr := schema.GroupVersionResource{Group: k.Group, Version: k.Version, Resource: k.Resource}
scopes := []string{""}
if k.Namespaced && c.Namespaces != nil {
scopes = c.Namespaces
}
for _, ns := range scopes {
if _, _, synced := d.r.caches.Dynamic.ReadableObject(gvr, ns, "probe"); !synced {
return k.GroupKind.String() + " (not loaded in " + ns + ")"
}
}
objs, _ := d.r.caches.Dynamic.ListWatchedReadOnly(gvr)
for _, o := range objs {
items = append(items, o)
}
}
for _, item := range items {
m, err := apimeta.Accessor(item)
if err != nil {
continue
}
got[m.GetNamespace()+"/"+m.GetName()] = string(m.GetUID()) + "@" + m.GetResourceVersion()
}
if fmt.Sprint(want) != fmt.Sprint(got) {
return k.GroupKind.String()
}
}
return ""
}
// step makes one random change: reparent, orphan, delete, recreate under the
// same name with a new UID, start deleting, create a dependent, add or drop a
// manager's member, or change what a manager's status says.
func (d *driver) step(r *rand.Rand, n int) string {
pick := func() ownershiptest.Object { return d.w.Objects[r.IntN(len(d.w.Objects))] }
pickManager := func() (ownershiptest.Object, bool) {
var managers []ownershiptest.Object
for _, o := range d.w.Objects {
if ownership.IsRecordingManager(o.Ref.GroupKind()) {
managers = append(managers, o)
}
}
if len(managers) == 0 {
return ownershiptest.Object{}, false
}
return managers[r.IntN(len(managers))], true
}
newManager := func() string {
kind := []resourceid.GroupKind{ownership.ArgoApplication, ownership.FluxKustomization, ownership.FluxHelmRelease}[r.IntN(3)]
k, _ := ownershiptest.KindOf(kind)
member := pick().Ref
o := ownershiptest.Object{
Ref: resourceid.Ref{Group: kind.Group, Kind: kind.Kind, Namespace: "alpha", Name: fmt.Sprintf("manager-%d", n)},
APIVersion: k.APIVersion(), UID: types.UID(fmt.Sprintf("uid-manager-%d", n)),
Membership: ownership.MembersRecorded, Members: []resourceid.Ref{member}, DestinationNamespace: "alpha",
}
d.create(o)
return fmt.Sprintf("%v created listing %v", o.Ref, member)
}
switch r.IntN(8) {
case 6:
m, ok := pickManager()
if !ok {
return newManager()
}
if len(m.Members) > 0 && r.IntN(2) == 0 {
gone := m.Members[r.IntN(len(m.Members))]
d.update(m.Ref, func(o *ownershiptest.Object) {
o.Members = slices.DeleteFunc(slices.Clone(o.Members), func(ref resourceid.Ref) bool { return ref == gone })
})
return fmt.Sprintf("%v drops member %v", m.Ref, gone)
}
added := pick().Ref
d.update(m.Ref, func(o *ownershiptest.Object) { o.Members = append(slices.Clone(o.Members), added) })
return fmt.Sprintf("%v adds member %v", m.Ref, added)
case 7:
m, ok := pickManager()
if !ok {
return newManager()
}
state := ownership.MembershipState(r.IntN(3))
form := uint8(r.IntN(3))
d.update(m.Ref, func(o *ownershiptest.Object) { o.Membership, o.Form = state, form })
return fmt.Sprintf("%v says %v in form %d", m.Ref, state, form)
case 0:
o, owner := pick(), pick()
d.update(o.Ref, func(o *ownershiptest.Object) {
o.OwnerReferences = append(o.OwnerReferences, owner.OwnerRef(r.IntN(2) == 0))
})
return fmt.Sprintf("%v gains owner %v", o.Ref, owner.Ref)
case 1:
o := pick()
d.update(o.Ref, func(o *ownershiptest.Object) { o.OwnerReferences = nil })
return fmt.Sprintf("%v loses its owners", o.Ref)
case 2:
if len(d.w.Objects) < 4 {
return "skip"
}
o := pick()
d.delete(o.Ref)
return fmt.Sprintf("%v deleted", o.Ref)
case 3:
o := pick()
d.delete(o.Ref)
o.UID = types.UID(fmt.Sprintf("%s-again-%d", o.UID, n))
d.create(o)
return fmt.Sprintf("%v recreated as %s", o.Ref, o.UID)
case 4:
o := pick()
at := metav1.NewTime(time.Date(2026, 10, 17, 11, 0, n, 0, time.UTC))
d.update(o.Ref, func(o *ownershiptest.Object) {
o.DeletionTimestamp = &at
o.Finalizers = append(slices.Clone(o.Finalizers), "example.io/hold")
})
return fmt.Sprintf("%v starts deleting", o.Ref)
default:
owner := pick()
ns := owner.Ref.Namespace
if ns == "" {
ns = "alpha"
}
kind := []resourceid.GroupKind{{Kind: "Pod"}, {Group: "example.io", Kind: "Gadget"}}[r.IntN(2)]
k, _ := ownershiptest.KindOf(kind)
o := ownershiptest.Object{
Ref: resourceid.Ref{Group: kind.Group, Kind: kind.Kind, Namespace: ns, Name: fmt.Sprintf("new-%d", n)},
APIVersion: k.APIVersion(), UID: types.UID(fmt.Sprintf("uid-new-%d", n)),
OwnerReferences: []metav1.OwnerReference{owner.OwnerRef(true)},
}
d.create(o)
return fmt.Sprintf("%v created under %v", o.Ref, owner.Ref)
}
}
func TestSourceConformsThroughChanges(t *testing.T) {
histories := 12
if testing.Short() {
histories = 3
}
for seed := range uint64(histories) {
t.Run(fmt.Sprint(seed), func(t *testing.T) {
w := ownershiptest.Generate(seed)
d := &driver{t: t, r: render(t, w), w: w}
changed := 0
r := rand.New(rand.NewPCG(seed, 7))
for n := range 10 {
change := d.step(r, n)
if change == "skip" {
continue
}
d.fence()
want, _ := ownershiptest.Scan(d.w).Begin()
got := begin(t, d.r.source())
if diffs := ownershiptest.Conform(t, want, got, ownershiptest.QueriesFor(d.w)); diffs > 0 {
t.Fatalf("after step %d (%s): %d differences", n, change, diffs)
}
if err := got.ValidateCaches(); err != nil {
t.Fatalf("after step %d (%s): ValidateCaches: %v", n, change, err)
}
changed++
}
if changed < 5 {
t.Fatalf("only %d of 10 steps changed the World", changed)
}
})
}
}
// The fence must not mistake a missing store for an empty one: a kind the
// World collects, with no objects, is caught up only once a store serving it
// has loaded.
func TestFenceWaitsForAnEmptyStoreToLoad(t *testing.T) {
f := newFixture(t, k8score.CacheConfig{}, nil)
w := ownershiptest.World{Collection: map[resourceid.GroupKind]ownershiptest.Collection{}}
for _, k := range ownershiptest.Kinds {
w.Collection[k.GroupKind] = ownershiptest.Collection{State: ownershiptest.NotCollected}
}
w.Collection[widgetKind] = ownershiptest.Collection{OnDemand: true}
d := &driver{t: t, r: &rendered{caches: f.caches}, w: w}
if _, _, ok := f.caches.Dynamic.ReadableObject(widgets, "", "probe"); ok {
t.Fatal("scenario: no store may serve Widgets yet")
}
if stale := d.staleKind(); stale == "" {
t.Fatal("the fence passed with no store serving Widgets")
}
}