homelab/ci/images/kuma-sync/internal/controller/gc_test.go

package controller

import (
	"context"
	"errors"
	"fmt"
	"sync/atomic"
	"testing"
	"time"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"

	v1 "gitea.renner.uno/konairius/homelab/kuma-sync/api/v1alpha1"
	"gitea.renner.uno/konairius/homelab/kuma-sync/internal/kumaclient"
)

type gcEnv struct {
	gc   *GC
	kuma *kumaclient.Fake
	cnt  *counting
}

func newGCEnv(t *testing.T, funcs *interceptor.Funcs, objs ...client.Object) *gcEnv {
	t.Helper()
	s := gcScheme(t)
	b := fake.NewClientBuilder().WithScheme(s).WithObjects(append(objs, namespace("default"))...)
	if funcs != nil {
		b = b.WithInterceptorFuncs(*funcs)
	}
	kf := kumaclient.NewFake()
	cnt := &counting{Client: kf}
	c := b.Build()
	return &gcEnv{gc: &GC{Client: c, APIReader: c, Kuma: cnt, Interval: time.Hour}, kuma: kf, cnt: cnt}
}

func gcScheme(t *testing.T) *runtime.Scheme {
	t.Helper()
	s := runtime.NewScheme()
	if err := v1.AddToScheme(s); err != nil {
		t.Fatal(err)
	}
	if err := corev1.AddToScheme(s); err != nil {
		t.Fatal(err)
	}
	return s
}

func namespace(name string) *corev1.Namespace {
	return &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: name}}
}

// apiServerWith is a fake API server that always has the default namespace.
func apiServerWith(t *testing.T, objs ...client.Object) client.Reader {
	t.Helper()
	return fake.NewClientBuilder().WithScheme(gcScheme(t)).WithObjects(append(objs, namespace("default"))...).Build()
}

// runs calls RunOnce n times and returns the last error.
func (e *gcEnv) runs(t *testing.T, n int) error {
	t.Helper()
	var err error
	for range n {
		err = e.gc.RunOnce(t.Context())
	}
	return err
}

func TestGCConfirmsWithAPIReaderBeforeOrphanDelete(t *testing.T) {
	e := newGCEnv(t, nil) // stale cache: knows no resources
	id := owned(t, e, "default/a")
	e.gc.APIReader = apiServerWith(t, httpMonitor("a"))
	if err := e.gc.RunOnce(t.Context()); err != nil {
		t.Fatal(err)
	}
	if e.kuma.Get(id) == nil || e.cnt.deletes != 0 {
		t.Error("monitor of a resource that exists on the API server was deleted")
	}

	// Truly absent on the API server: deleted.
	e.gc.APIReader = apiServerWith(t)
	if err := e.runs(t, 2); err != nil {
		t.Fatal(err)
	}
	if e.kuma.Get(id) != nil {
		t.Error("confirmed orphan survived")
	}
}

func TestGCSkipsOnAPIReaderError(t *testing.T) {
	e := newGCEnv(t, nil)
	id := owned(t, e, "default/gone")
	s := runtime.NewScheme()
	if err := v1.AddToScheme(s); err != nil {
		t.Fatal(err)
	}
	e.gc.APIReader = fake.NewClientBuilder().WithScheme(s).WithInterceptorFuncs(interceptor.Funcs{
		Get: func(context.Context, client.WithWatch, client.ObjectKey, client.Object, ...client.GetOption) error {
			return errors.New("apiserver unavailable")
		},
	}).Build()
	if err := e.gc.RunOnce(t.Context()); err == nil {
		t.Fatal("expected the API reader error to be recorded")
	}
	if e.kuma.Get(id) == nil || e.cnt.deletes != 0 {
		t.Error("deleted without confirmation")
	}
}

func TestGCMalformedOwnerKeyNeverDeleted(t *testing.T) {
	e := newGCEnv(t, nil)
	var ids []int64
	for _, key := range []string{"noslash", "/name", "ns/", "a/b/c"} {
		ids = append(ids, owned(t, e, key))
	}
	if err := e.gc.RunOnce(t.Context()); err == nil {
		t.Fatal("expected errors for malformed keys")
	}
	for _, id := range ids {
		if e.kuma.Get(id) == nil {
			t.Errorf("monitor %d with a malformed key was deleted", id)
		}
	}
}

func owned(t *testing.T, e *gcEnv, key string) int64 {
	t.Helper()
	id, err := e.kuma.Create(t.Context(), kumaclient.Monitor{Type: "group", Name: key, OwnerKey: key})
	if err != nil {
		t.Fatal(err)
	}
	return id
}

func TestGCDeletesOrphans(t *testing.T) {
	e := newGCEnv(t, nil, httpMonitor("live"))
	live := owned(t, e, "default/live")
	gone := owned(t, e, "default/gone")
	if err := e.runs(t, 2); err != nil {
		t.Fatal(err)
	}
	if e.kuma.Get(gone) != nil {
		t.Error("orphan survived")
	}
	if e.kuma.Get(live) == nil {
		t.Error("live monitor deleted")
	}
}

func TestGCIgnoresUntagged(t *testing.T) {
	e := newGCEnv(t, nil)
	foreign := e.kuma.AddForeign(kumaclient.Monitor{Name: "foreign"})
	if err := e.gc.RunOnce(t.Context()); err != nil {
		t.Fatal(err)
	}
	if e.kuma.Get(foreign) == nil || e.cnt.deletes != 0 {
		t.Errorf("foreign monitor touched: deletes=%d", e.cnt.deletes)
	}
}

func TestGCKeepsLiveResources(t *testing.T) {
	e := newGCEnv(t, nil, httpMonitor("a"), httpMonitor("b"))
	owned(t, e, "default/a")
	owned(t, e, "default/b")
	if err := e.gc.RunOnce(t.Context()); err != nil {
		t.Fatal(err)
	}
	if e.cnt.deletes != 0 {
		t.Errorf("deletes = %d", e.cnt.deletes)
	}
}

func TestGCKeepsResourcesBeingDeleted(t *testing.T) {
	e := newGCEnv(t, nil, deleting(httpMonitor("a")))
	owned(t, e, "default/a")
	if err := e.gc.RunOnce(t.Context()); err != nil {
		t.Fatal(err)
	}
	if e.cnt.deletes != 0 {
		t.Errorf("the finalizer path owns this deletion, deletes = %d", e.cnt.deletes)
	}
}

func TestGCAbortsOnListError(t *testing.T) {
	e := newGCEnv(t, &interceptor.Funcs{
		List: func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error {
			return errors.New("cache not ready")
		},
	})
	owned(t, e, "default/gone")
	if err := e.gc.RunOnce(t.Context()); err == nil {
		t.Fatal("expected error")
	}
	if e.cnt.deletes != 0 {
		t.Errorf("deletes = %d, want 0", e.cnt.deletes)
	}
}

func TestGCAbortsOnKumaListError(t *testing.T) {
	e := newGCEnv(t, nil)
	owned(t, e, "default/gone")
	e.kuma.SetErrs(errors.New("boom"), nil, nil, nil)
	if err := e.gc.RunOnce(t.Context()); err == nil {
		t.Fatal("expected error")
	}
	if e.cnt.deletes != 0 {
		t.Errorf("deletes = %d, want 0", e.cnt.deletes)
	}
}

func TestGCNoDeletesOnPartialList(t *testing.T) {
	e := newGCEnv(t, nil)
	owned(t, e, "default/gone")
	e.kuma.SetListPartial(errors.Join(&kumaclient.ConflictError{MonitorID: 9, Keys: []string{"default/x", "default/y"}}))
	if err := e.gc.RunOnce(t.Context()); err == nil {
		t.Fatal("expected error")
	}
	if e.cnt.deletes != 0 {
		t.Errorf("deletes = %d, want 0", e.cnt.deletes)
	}
}

func TestGCDeletesDuplicateExtras(t *testing.T) {
	e := newGCEnv(t, nil, httpMonitor("a"))
	first := owned(t, e, "default/a")
	second := owned(t, e, "default/a")
	third := owned(t, e, "default/a")
	if err := e.gc.RunOnce(t.Context()); err != nil {
		t.Fatal(err)
	}
	if e.kuma.Get(first) == nil {
		t.Error("lowest ID must stay")
	}
	if e.kuma.Get(second) != nil || e.kuma.Get(third) != nil {
		t.Error("extras must be deleted")
	}
}

func TestGCDeleteNotFoundIsSuccess(t *testing.T) {
	e := newGCEnv(t, nil)
	owned(t, e, "default/gone")
	e.kuma.SetErrs(nil, nil, nil, fmt.Errorf("delete: %w", kumaclient.ErrNotFound))
	if err := e.runs(t, 2); err != nil {
		t.Fatalf("ErrNotFound must count as success: %v", err)
	}
}

// waiter is a CacheSyncer whose result the test controls.
type waiter struct {
	release chan struct{}
	result  bool
	started atomic.Bool
}

func (w *waiter) WaitForCacheSync(ctx context.Context) bool {
	w.started.Store(true)
	select {
	case <-w.release:
		return w.result
	case <-ctx.Done():
		return false
	}
}

func TestGCWaitsForCacheSync(t *testing.T) {
	e := newGCEnv(t, nil)
	owned(t, e, "default/gone") // would look like an orphan against an empty cache
	w := &waiter{release: make(chan struct{}), result: true}
	e.gc.Cache = w
	e.gc.Interval = 20 * time.Millisecond

	ctx, cancel := context.WithCancel(t.Context())
	done := make(chan error, 1)
	go func() { done <- e.gc.Start(ctx) }()

	deadline := time.Now().Add(5 * time.Second)
	for !w.started.Load() && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	time.Sleep(100 * time.Millisecond)
	if e.kuma.Get(1) == nil {
		t.Fatal("GC ran before the cache synced")
	}

	close(w.release)
	deadline = time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) && e.kuma.Get(1) != nil {
		time.Sleep(5 * time.Millisecond)
	}
	if e.kuma.Get(1) != nil {
		t.Error("GC did not run after the cache synced")
	}
	cancel()
	if err := <-done; err != nil {
		t.Errorf("Start returned %v on shutdown", err)
	}
}

func TestGCOrphanGraceDelaysDelete(t *testing.T) {
	e := newGCEnv(t, nil)
	id := owned(t, e, "default/gone")
	e.gc.OrphanGrace = time.Hour
	clock := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
	e.gc.Now = func() time.Time { return clock }

	for _, step := range []struct {
		at      time.Duration
		deleted bool
	}{
		{0, false},                // first sighting
		{30 * time.Minute, false}, // second run, grace not over
		{59 * time.Minute, false}, // still inside the grace period
		{60 * time.Minute, true},  // third run, a full hour after the first sighting
	} {
		clock = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).Add(step.at)
		if err := e.gc.RunOnce(t.Context()); err != nil {
			t.Fatal(err)
		}
		if gone := e.kuma.Get(id) == nil; gone != step.deleted {
			t.Fatalf("at +%v: deleted = %v, want %v", step.at, gone, step.deleted)
		}
	}
}

func TestGCOrphanGraceResetsWhenResourceReturns(t *testing.T) {
	e := newGCEnv(t, nil)
	id := owned(t, e, "default/a")
	e.gc.OrphanGrace = time.Hour
	base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
	var clock time.Time
	e.gc.Now = func() time.Time { return clock }
	run := func(at time.Duration) {
		t.Helper()
		clock = base.Add(at)
		if err := e.gc.RunOnce(t.Context()); err != nil {
			t.Fatal(err)
		}
	}

	run(0) // orphan seen
	km := httpMonitor("a")
	if err := e.gc.Client.Create(t.Context(), km); err != nil {
		t.Fatal(err)
	}
	run(40 * time.Minute) // resource is back: history dropped
	if err := e.gc.Client.Delete(t.Context(), km); err != nil {
		t.Fatal(err)
	}
	run(61 * time.Minute) // orphan again, first sighting of the new streak
	run(90 * time.Minute) // only 29 minutes into the new streak
	if e.kuma.Get(id) == nil {
		t.Fatal("deleted although the grace period restarted when the resource returned")
	}
	run(122 * time.Minute) // 61 minutes into the new streak
	if e.kuma.Get(id) != nil {
		t.Error("orphan survived a full grace period")
	}
}

func TestGCSkipsMissingNamespace(t *testing.T) {
	e := newGCEnv(t, nil)
	id := owned(t, e, "ghost/x")
	if err := e.runs(t, 3); err != nil {
		t.Fatal(err)
	}
	if e.kuma.Get(id) == nil || e.cnt.deletes != 0 {
		t.Fatal("monitor of a missing namespace was deleted")
	}
	if err := e.gc.Client.Create(t.Context(), namespace("ghost")); err != nil {
		t.Fatal(err)
	}
	if err := e.runs(t, 2); err != nil {
		t.Fatal(err)
	}
	if e.kuma.Get(id) != nil {
		t.Error("orphan in an existing namespace survived")
	}
}

func TestGCStartFailsWhenCacheNeverSyncs(t *testing.T) {
	e := newGCEnv(t, nil)
	owned(t, e, "default/gone")
	released := make(chan struct{})
	close(released)
	e.gc.Cache = &waiter{release: released, result: false}
	if err := e.gc.Start(t.Context()); err == nil {
		t.Fatal("expected an error when the cache does not sync")
	}
	if e.cnt.deletes != 0 {
		t.Error("GC deleted without a synced cache")
	}
}