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

package controller

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strings"
	"sync"
	"time"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/manager"

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

// +kubebuilder:rbac:groups="",resources=namespaces,verbs=get

// CacheSyncer is satisfied by the manager's cache.
type CacheSyncer interface {
	WaitForCacheSync(ctx context.Context) bool
}

// errListFailed marks a run that stopped before deleting anything because a
// list failed.
var errListFailed = errors.New("list failed, nothing deleted")

// minOrphanRuns is how many consecutive runs must see a monitor orphaned
// before it is deleted, whatever the grace period.
const minOrphanRuns = 2

// GC deletes owned Kuma monitors that no KumaMonitor accounts for, and
// duplicates of one owner key beyond the lowest ID.
//
// A run deletes nothing unless both lists succeeded completely: any error from
// ListOwned (conflicts and partial results included) or from listing
// KumaMonitors aborts it. Start waits for Cache to sync first, so an empty
// cache at startup never looks like "every resource was deleted".
//
// An orphan, confirmed absent on the API server, is deleted only after it was
// seen orphaned in at least two consecutive runs spanning OrphanGrace; this
// keeps a cluster rebuild, where the resources are applied long after the
// controller starts, from wiping their monitors. The first-seen times live in
// memory, so a restart starts the grace period over. Orphans in a namespace
// that does not exist are never deleted. Duplicate extras are deleted at once.
type GC struct {
	Client   client.Client
	Kuma     kumaclient.Client
	Interval time.Duration
	// OrphanGrace is how long a monitor must have been seen orphaned.
	OrphanGrace time.Duration
	// Cache is waited on before the first run; nil skips the wait.
	Cache CacheSyncer
	// APIReader reads straight from the API server. Every orphan is confirmed
	// against it before it is deleted.
	APIReader client.Reader
	// Now is the clock; nil means time.Now.
	Now func() time.Time

	mu      sync.Mutex
	pending map[int64]orphanSeen
}

type orphanSeen struct {
	since time.Time
	runs  int
}

var (
	_ manager.Runnable               = (*GC)(nil)
	_ manager.LeaderElectionRunnable = (*GC)(nil)
)

// NeedLeaderElection is false: there is a single replica, no election.
func (g *GC) NeedLeaderElection() bool { return false }

func (g *GC) now() time.Time {
	if g.Now != nil {
		return g.Now()
	}
	return time.Now()
}

func (g *GC) Start(ctx context.Context) error {
	logger := log.FromContext(ctx).WithName("gc")
	if g.Cache != nil && !g.Cache.WaitForCacheSync(ctx) {
		if ctx.Err() != nil {
			return nil
		}
		return errors.New("gc: cache did not sync")
	}
	interval := g.Interval
	if interval <= 0 {
		interval = 5 * time.Minute
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		if err := g.RunOnce(ctx); err != nil {
			if errors.Is(err, errListFailed) {
				logger.Error(err, "gc run aborted")
			} else {
				logger.Error(err, "gc run finished with errors")
			}
		}
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}
	}
}

// existsInAPI asks the API server, not the cache, whether the resource behind
// an owner key exists. A malformed key is an error, never "absent".
func (g *GC) existsInAPI(ctx context.Context, ownerKey string) (bool, error) {
	ns, name, ok := strings.Cut(ownerKey, "/")
	if !ok || ns == "" || name == "" || strings.Contains(name, "/") {
		return false, fmt.Errorf("malformed owner key %q", ownerKey)
	}
	err := g.APIReader.Get(ctx, types.NamespacedName{Namespace: ns, Name: name}, &v1.KumaMonitor{})
	switch {
	case err == nil:
		return true, nil
	case apierrors.IsNotFound(err):
		return false, nil
	default:
		return false, err
	}
}

// namespaceExists reports whether the namespace of an owner key exists.
func (g *GC) namespaceExists(ctx context.Context, ownerKey string) (bool, error) {
	ns, _, _ := strings.Cut(ownerKey, "/")
	err := g.APIReader.Get(ctx, types.NamespacedName{Name: ns}, &corev1.Namespace{})
	switch {
	case err == nil:
		return true, nil
	case apierrors.IsNotFound(err):
		return false, nil
	default:
		return false, err
	}
}

// RunOnce performs one collection pass.
func (g *GC) RunOnce(ctx context.Context) error {
	logger := log.FromContext(ctx).WithName("gc")
	if g.APIReader == nil {
		return errors.New("gc: APIReader is required to confirm orphans")
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.pending == nil {
		g.pending = map[int64]orphanSeen{}
	}

	// Kuma list must precede the cache list.
	owned, err := g.Kuma.ListOwned(ctx)
	if err != nil {
		clear(g.pending)
		return fmt.Errorf("%w: list owned monitors: %w", errListFailed, err)
	}
	var resources v1.KumaMonitorList
	if err := g.Client.List(ctx, &resources); err != nil {
		clear(g.pending)
		return fmt.Errorf("%w: list KumaMonitors: %w", errListFailed, err)
	}
	live := make(map[string]bool, len(resources.Items))
	for i := range resources.Items {
		live[OwnerKey(&resources.Items[i])] = true
	}

	now := g.now()
	slices.SortFunc(owned, func(a, b kumaclient.Monitor) int { return int(a.ID - b.ID) })
	seen := map[string]bool{}
	candidates := map[int64]bool{}
	var errs []error
	del := func(m kumaclient.Monitor, reason string) bool {
		logger.Info("deleting monitor", "monitorID", m.ID, "ownerKey", m.OwnerKey, "reason", reason)
		if err := g.Kuma.Delete(ctx, m.ID); err != nil && !errors.Is(err, kumaclient.ErrNotFound) {
			errs = append(errs, fmt.Errorf("delete monitor %d (%s): %w", m.ID, m.OwnerKey, err))
			return false
		}
		return true
	}
	for _, m := range owned {
		keep := live[m.OwnerKey] && !seen[m.OwnerKey]
		seen[m.OwnerKey] = true
		if keep {
			continue
		}
		if live[m.OwnerKey] {
			del(m, "duplicate of a lower monitor ID")
			continue
		}

		exists, err := g.existsInAPI(ctx, m.OwnerKey)
		if err != nil {
			errs = append(errs, fmt.Errorf("confirm orphan monitor %d (%s): %w", m.ID, m.OwnerKey, err))
			continue
		}
		if exists {
			logger.Info("KumaMonitor exists on the API server, not an orphan", "monitorID", m.ID, "ownerKey", m.OwnerKey)
			continue
		}
		nsOK, err := g.namespaceExists(ctx, m.OwnerKey)
		if err != nil {
			errs = append(errs, fmt.Errorf("check namespace of monitor %d (%s): %w", m.ID, m.OwnerKey, err))
			continue
		}
		if !nsOK {
			logger.Info("namespace does not exist, leaving the monitor alone", "monitorID", m.ID, "ownerKey", m.OwnerKey)
			continue
		}

		candidates[m.ID] = true
		st, ok := g.pending[m.ID]
		if !ok {
			st.since = now
		}
		st.runs++
		g.pending[m.ID] = st
		if st.runs < minOrphanRuns || now.Sub(st.since) < g.OrphanGrace {
			logger.Info("orphan candidate, waiting out the grace period", "monitorID", m.ID, "ownerKey", m.OwnerKey,
				"runs", st.runs, "orphanFor", now.Sub(st.since).String(), "grace", g.OrphanGrace.String())
			continue
		}
		if del(m, "no matching KumaMonitor") {
			delete(g.pending, m.ID)
		}
	}
	// Whatever stopped being an orphan, or vanished, loses its history.
	for id := range g.pending {
		if !candidates[id] {
			delete(g.pending, id)
		}
	}
	return errors.Join(errs...)
}