fancy-dns/internal/app/kube.go

package app

import (
	"context"
	"path/filepath"
	"time"

	"k8s.io/client-go/dynamic"

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

// KubeOptions turn on the Kubernetes source: ClientGroup, BlockList and
// DNSZone objects are watched and merged into the configuration. The runtime
// only reads; it works the same in a pod and on a workstation.
type KubeOptions struct {
	Enabled bool
	// Kubeconfig and Context pick credentials; both empty means the pod's
	// service account when there is one, else the default kubeconfig.
	Kubeconfig string
	Context    string
	// Namespace restricts the watch; empty means the pod's own namespace
	// in-cluster, else every namespace.
	Namespace string
	// SyncTimeout bounds how long startup waits for the first listing before
	// falling back to the last saved snapshot. Zero means 30 seconds.
	SyncTimeout time.Duration
	// Client replaces the one built from the credentials; for tests.
	Client dynamic.Interface
}

const (
	kubeDegraded  = "kubernetes: not synced, serving the last saved cluster configuration"
	kubeRejected  = "kubernetes: objects rejected"
	kubeNoSnap    = "kubernetes: not synced and no saved snapshot, cluster configuration is empty"
	kubeInvalid   = "kubernetes: the merged configuration is invalid, cluster objects are ignored"
	kubeZoneFail  = "kubernetes: cluster zones could not be written, serving the previous ones"
	defaultKubeTO = 30 * time.Second
)

func (a *App) kubeZoneDir() string {
	if a.kube == nil {
		return ""
	}
	return filepath.Join(a.DataDir(), "kube-zones")
}

func (a *App) kubeSnapshotPath() string { return filepath.Join(a.DataDir(), "kube-snapshot.json") }

// startKube connects to the cluster and installs the overlay. It runs before
// the components that read the configuration, so the first blocklist compile
// and the first zone load already include the cluster's objects.
func (a *App) startKube(ctx context.Context) error {
	o := a.opt.Kube
	if !o.Enabled {
		return nil
	}
	ns := o.Namespace
	if ns == "" {
		ns = kubesource.Namespace()
	}
	client, from := o.Client, "injected client"
	if client == nil {
		var err error
		client, from, err = kubesource.NewClient(kubesource.ClientOptions{Kubeconfig: o.Kubeconfig, Context: o.Context})
		if err != nil {
			return err
		}
	}
	src, err := kubesource.New(kubesource.Options{Client: client, Namespace: ns, Logger: a.log})
	if err != nil {
		return err
	}
	a.kube = src
	a.log.Info("watching Kubernetes for fancy-dns objects", "credentials", from, "namespace", ns)

	a.cfg.SetOverlay(a.kubeOverlay)
	go func() { _ = src.Run(ctx) }()

	wait := o.SyncTimeout
	if wait <= 0 {
		wait = defaultKubeTO
	}
	syncCtx, cancel := context.WithTimeout(ctx, wait)
	defer cancel()
	if err := src.WaitSynced(syncCtx); err != nil {
		// A cluster that cannot be reached must not stop DNS from coming up:
		// resolve from what was last known and keep watching in the background.
		snap, lerr := kubesource.Load(a.kubeSnapshotPath())
		switch {
		case lerr != nil:
			a.log.Error("saved cluster snapshot is unreadable", "error", lerr)
			a.markDegraded(kubeNoSnap)
		case snap == nil:
			a.markDegraded(kubeNoSnap)
		default:
			a.kubeSnap.Store(snap)
			a.markDegraded(kubeDegraded)
		}
		a.log.Warn("Kubernetes did not sync in time, continuing without it", "waited", wait.String(), "error", err)
		a.applyKube(a.kubeSnap.Load(), false)
	} else {
		a.applyKube(src.Snapshot(), true)
	}
	// Later changes arrive here. Registered after the first apply so the
	// startup path above is the only one that runs before the listeners exist.
	src.OnChange(func(sn *kubesource.Snapshot) { a.applyKube(sn, true) })
	return nil
}

// applyKube makes snap the cluster's contribution to the configuration.
func (a *App) applyKube(snap *kubesource.Snapshot, live bool) {
	if snap == nil {
		a.cfg.Reapply()
		return
	}
	prev := a.kubeSnap.Load()
	a.kubeSnap.Store(snap)
	if live {
		a.clearDegraded(kubeDegraded)
		a.clearDegraded(kubeNoSnap)
		if err := kubesource.Save(a.kubeSnapshotPath(), snap); err != nil {
			a.log.Warn("could not save the cluster snapshot", "error", err)
		}
		if len(snap.Problems) > 0 {
			a.markDegraded(kubeRejected)
		} else {
			a.clearDegraded(kubeRejected)
		}
	}
	if _, err := kubesource.SyncZoneDir(a.kubeZoneDir(), snap.Zones); err != nil {
		a.markDegraded(kubeZoneFail)
		a.log.Error("could not write cluster zones, keeping the previous ones", "error", err)
	} else {
		a.clearDegraded(kubeZoneFail)
	}
	if a.blocks != nil {
		a.blocks.SetManagedRPZ(kubesource.RenderRPZ(snap.Overrides))
	}
	a.cfg.Reapply()
	if prev == nil {
		return
	}
	// Reapply reloaded the zones and rebuilt the ACL. Cached answers were
	// produced under the old policy, so they go, as they do for any ACL,
	// blocklist or zone edit.
	a.invalidateBlocklistCache()
	if a.blocks != nil {
		go func() {
			if _, err := a.blocks.RefreshAll(context.Background()); err != nil {
				a.log.Warn("blocklist refresh after cluster change failed", "error", err)
			}
		}()
	}
}

// kubeOverlay is the Store overlay: the file's configuration plus the
// cluster's objects. A merge that does not validate is dropped whole and the
// file's configuration stays in force, because half of a policy is worse than
// the last whole one.
func (a *App) kubeOverlay(base *config.Config) *config.Config {
	snap := a.kubeSnap.Load()
	if snap == nil {
		return base
	}
	merged, dropped := kubesource.Merge(base, snap)
	for _, d := range dropped {
		a.log.Warn("cluster object ignored", "object", d)
	}
	check := *merged
	if err := check.Validate(); err != nil {
		a.log.Error("cluster configuration does not validate, ignoring it", "error", err)
		a.markDegraded(kubeInvalid)
		return base
	}
	a.clearDegraded(kubeInvalid)
	return merged
}

// clusterOwned reports whether a group or list name comes from the cluster,
// and so cannot be edited through the console.
func (a *App) clusterOwned(kind, name string) bool {
	snap := a.kubeSnap.Load()
	if snap == nil {
		return false
	}
	switch kind {
	case "group":
		for _, g := range snap.Groups {
			if g.Name == name {
				return true
			}
		}
	case "list":
		for _, l := range snap.Lists {
			if l.Name == name {
				return true
			}
		}
	}
	return false
}