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 }