fancy-dns/internal/kubesource/source.go

package kubesource

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"sync"
	"sync/atomic"
	"time"

	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/dynamic/dynamicinformer"
	"k8s.io/client-go/tools/cache"

	"git.k8s-git-k6n.net/mats/fancy-dns/internal/config"
)

// Problem is one object the source could not use.
type Problem struct {
	Kind      string `json:"kind"`
	Namespace string `json:"namespace"`
	Name      string `json:"name"`
	Message   string `json:"message"`
	// KeptLastGood is true when an earlier valid version of the object is
	// still in force.
	KeptLastGood bool `json:"kept_last_good"`
}

// Snapshot is everything the cluster currently declares, already converted and
// validated per object. It is immutable once published.
type Snapshot struct {
	// Revision increases with every rebuild that changed content.
	Revision uint64 `json:"revision"`
	// Groups are in evaluation order.
	Groups []config.Group      `json:"groups"`
	Lists  []config.ListSource `json:"lists"`
	// Zones maps an apex to its rendered master file.
	Zones    map[string]string `json:"zones"`
	Problems []Problem         `json:"problems,omitempty"`
}

// Options configure a Source.
type Options struct {
	// Client is the dynamic client; build one with NewClient.
	Client dynamic.Interface
	// Namespace restricts the watch; "" watches every namespace.
	Namespace string
	// Debounce coalesces bursts of events into one rebuild.
	Debounce time.Duration
	Resync   time.Duration
	Logger   *slog.Logger
}

// Source watches the custom resources and publishes Snapshots.
type Source struct {
	opt      Options
	log      *slog.Logger
	cur      atomic.Pointer[Snapshot]
	trigger  chan struct{}
	synced   chan struct{}
	syncOnce sync.Once

	mu       sync.Mutex
	onChange []func(*Snapshot)
	// good holds the last valid conversion of each object, so one edit that
	// breaks an object does not remove its policy from a running resolver.
	good map[string]any
	sig  string
	rev  uint64
}

// New returns a Source. Nothing runs until Run.
func New(opt Options) (*Source, error) {
	if opt.Client == nil {
		return nil, errors.New("kubesource: a client is required")
	}
	if opt.Debounce <= 0 {
		opt.Debounce = 250 * time.Millisecond
	}
	if opt.Resync <= 0 {
		opt.Resync = 10 * time.Minute
	}
	if opt.Logger == nil {
		opt.Logger = slog.Default()
	}
	s := &Source{
		opt:     opt,
		log:     opt.Logger.With("component", "kubesource"),
		trigger: make(chan struct{}, 1),
		synced:  make(chan struct{}),
		good:    map[string]any{},
	}
	return s, nil
}

// Snapshot is the latest published snapshot, or nil before the first sync.
func (s *Source) Snapshot() *Snapshot { return s.cur.Load() }

// OnChange registers fn to run, in order, after each snapshot that differs
// from the previous one.
func (s *Source) OnChange(fn func(*Snapshot)) {
	s.mu.Lock()
	s.onChange = append(s.onChange, fn)
	s.mu.Unlock()
}

// WaitSynced blocks until every informer has listed once, or ctx ends. A
// cluster without the CRDs installed never syncs, so callers bound the wait.
func (s *Source) WaitSynced(ctx context.Context) error {
	select {
	case <-s.synced:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Run starts the informers and rebuilds snapshots until ctx ends.
func (s *Source) Run(ctx context.Context) error {
	factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(s.opt.Client, s.opt.Resync, s.opt.Namespace, nil)
	informers := make(map[schema.GroupVersionResource]cache.SharedIndexInformer, len(allGVRs))
	handler := cache.ResourceEventHandlerFuncs{
		AddFunc:    func(any) { s.poke() },
		UpdateFunc: func(_, _ any) { s.poke() },
		DeleteFunc: func(any) { s.poke() },
	}
	var syncs []cache.InformerSynced
	for _, gvr := range allGVRs {
		inf := factory.ForResource(gvr).Informer()
		if _, err := inf.AddEventHandler(handler); err != nil {
			return fmt.Errorf("watch %s: %w", gvr.Resource, err)
		}
		_ = inf.SetWatchErrorHandler(func(_ *cache.Reflector, err error) {
			s.log.Warn("watch failed, will retry", "resource", gvr.Resource, "error", err)
		})
		informers[gvr] = inf
		syncs = append(syncs, inf.HasSynced)
	}
	factory.Start(ctx.Done())

	go func() {
		if cache.WaitForCacheSync(ctx.Done(), syncs...) {
			s.rebuild(informers)
			s.syncOnce.Do(func() { close(s.synced) })
		}
	}()

	for {
		select {
		case <-ctx.Done():
			factory.Shutdown()
			return nil
		case <-s.trigger:
		}
		// Debounce: let a burst (a kubectl apply of a directory) land first.
		t := time.NewTimer(s.opt.Debounce)
		select {
		case <-ctx.Done():
			t.Stop()
			factory.Shutdown()
			return nil
		case <-t.C:
		}
		select {
		case <-s.synced:
			s.rebuild(informers)
		default:
			// Not synced yet: the sync goroutine does the first rebuild, and
			// a partial list must not be published as the cluster's state.
		}
	}
}

func (s *Source) poke() {
	select {
	case s.trigger <- struct{}{}:
	default:
	}
}

func (s *Source) objects(inf cache.SharedIndexInformer) []*unstructured.Unstructured {
	raw := inf.GetStore().List()
	out := make([]*unstructured.Unstructured, 0, len(raw))
	for _, o := range raw {
		if u, ok := o.(*unstructured.Unstructured); ok && u.GetDeletionTimestamp() == nil {
			out = append(out, u)
		}
	}
	// A stable order makes duplicate resolution and the snapshot signature
	// deterministic regardless of the informer's map iteration.
	sort.Slice(out, func(i, j int) bool {
		a, b := out[i], out[j]
		if at, bt := a.GetCreationTimestamp(), b.GetCreationTimestamp(); !at.Equal(&bt) {
			return at.Before(&bt)
		}
		if a.GetNamespace() != b.GetNamespace() {
			return a.GetNamespace() < b.GetNamespace()
		}
		return a.GetName() < b.GetName()
	})
	return out
}

func (s *Source) rebuild(informers map[schema.GroupVersionResource]cache.SharedIndexInformer) {
	s.mu.Lock()
	snap := s.build(
		s.objects(informers[GroupGVR]),
		s.objects(informers[ListGVR]),
		s.objects(informers[ZoneGVR]),
	)
	sig := signature(snap)
	changed := sig != s.sig
	if changed {
		s.sig = sig
		s.rev++
		snap.Revision = s.rev
		s.cur.Store(snap)
	} else if prev := s.cur.Load(); prev != nil {
		// Content is the same; problems may still have changed, so refresh
		// them without notifying anyone.
		snap.Revision = prev.Revision
		s.cur.Store(snap)
	}
	hooks := append([]func(*Snapshot){}, s.onChange...)
	s.mu.Unlock()
	if !changed {
		return
	}
	s.log.Info("cluster configuration changed", "revision", snap.Revision,
		"groups", len(snap.Groups), "lists", len(snap.Lists), "zones", len(snap.Zones),
		"rejected", len(snap.Problems))
	for _, fn := range hooks {
		fn(snap)
	}
}

// build converts every object. Caller holds s.mu.
func (s *Source) build(groups, lists, zones []*unstructured.Unstructured) *Snapshot {
	snap := &Snapshot{Zones: map[string]string{}}
	live := map[string]bool{}

	problem := func(kind string, u *unstructured.Unstructured, msg string, kept bool) {
		snap.Problems = append(snap.Problems, Problem{
			Kind: kind, Namespace: u.GetNamespace(), Name: u.GetName(), Message: msg, KeptLastGood: kept,
		})
		s.log.Warn("rejected object", "kind", kind, "namespace", u.GetNamespace(), "name", u.GetName(),
			"error", msg, "kept_last_good", kept)
	}
	// resolve converts one object, falling back to its last good version.
	resolve := func(kind string, u *unstructured.Unstructured, conv func() (any, error)) (any, bool) {
		key := kind + "/" + u.GetNamespace() + "/" + u.GetName()
		live[key] = true
		v, err := conv()
		if err == nil {
			s.good[key] = v
			return v, true
		}
		old, ok := s.good[key]
		problem(kind, u, err.Error(), ok)
		return old, ok
	}

	seenGroup := map[string]bool{}
	var gs []group
	for _, u := range groups {
		v, ok := resolve("ClientGroup", u, func() (any, error) { return convertGroup(u) })
		if !ok {
			continue
		}
		g := v.(group)
		if seenGroup[g.cfg.Name] {
			problem("ClientGroup", u, "another ClientGroup already uses this name", false)
			continue
		}
		seenGroup[g.cfg.Name] = true
		gs = append(gs, g)
	}
	sort.SliceStable(gs, func(i, j int) bool {
		if gs[i].priority != gs[j].priority {
			return gs[i].priority < gs[j].priority
		}
		return gs[i].cfg.Name < gs[j].cfg.Name
	})
	for _, g := range gs {
		snap.Groups = append(snap.Groups, g.cfg)
	}

	seenList := map[string]bool{}
	for _, u := range lists {
		v, ok := resolve("BlockList", u, func() (any, error) { return convertList(u) })
		if !ok {
			continue
		}
		l := v.(config.ListSource)
		if seenList[l.Name] {
			problem("BlockList", u, "another BlockList already uses this name", false)
			continue
		}
		seenList[l.Name] = true
		snap.Lists = append(snap.Lists, l)
	}

	for _, u := range zones {
		v, ok := resolve("DNSZone", u, func() (any, error) { return convertZone(u) })
		if !ok {
			continue
		}
		z := v.(zone)
		if _, dup := snap.Zones[z.apex]; dup {
			// Objects are ordered oldest first, so the earliest claim wins.
			problem("DNSZone", u, "another DNSZone already declares "+z.apex, false)
			continue
		}
		snap.Zones[z.apex] = z.text
	}

	for key := range s.good {
		if !live[key] {
			delete(s.good, key)
		}
	}
	return snap
}