osbase/net/dns/kubernetes/controller.go

package kubernetes

// Taken and modified from the Kubernetes plugin of CoreDNS, under Apache 2.0.

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

	api "k8s.io/api/core/v1"
	discovery "k8s.io/api/discovery/v1"
	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"

	"source.monogon.dev/osbase/net/dns/kubernetes/object"
)

const (
	svcIPIndex           = "ServiceIP"
	epNameNamespaceIndex = "EndpointNameNamespace"
	epIPIndex            = "EndpointsIP"
)

// epLabelSelector selects EndpointSlices that belong to headless services.
// Endpoint DNS records are only served for headless services,
// and we can save resources by not fetching other EndpointSlices.
const epLabelSelector = api.IsHeadlessService

type dnsController interface {
	GetSvc(string) *object.Service
	SvcIndexReverse(string) []*object.Service
	EpIndex(string) []*object.Endpoints
	EpIndexReverse(string) []*object.Endpoints
	NamespaceExists(string) bool

	Start(<-chan struct{})
	HasSynced() bool

	// Modified returns the timestamp of the most recent changes to services.
	Modified() int64
}

type dnsControl struct {
	// modified tracks timestamp of the most recent changes
	modified  atomic.Int64
	hasSynced atomic.Bool

	svcController cache.Controller
	epController  cache.Controller
	nsController  cache.Controller

	svcLister cache.Indexer
	epLister  cache.Indexer
	nsLister  cache.Store
}

// newdnsController creates a controller.
func newdnsController(ctx context.Context, kubeClient kubernetes.Interface) *dnsControl {
	dns := dnsControl{}

	dns.svcLister, dns.svcController = object.NewIndexerInformer(
		&cache.ListWatch{
			ListFunc:  serviceListFunc(ctx, kubeClient, api.NamespaceAll),
			WatchFunc: serviceWatchFunc(ctx, kubeClient, api.NamespaceAll),
		},
		&api.Service{},
		&dns,
		cache.Indexers{svcIPIndex: svcIPIndexFunc},
		object.DefaultProcessor(object.ToService),
	)

	dns.epLister, dns.epController = object.NewIndexerInformer(
		&cache.ListWatch{
			ListFunc:  endpointSliceListFunc(ctx, kubeClient, api.NamespaceAll),
			WatchFunc: endpointSliceWatchFunc(ctx, kubeClient, api.NamespaceAll),
		},
		&discovery.EndpointSlice{},
		&dns,
		cache.Indexers{epNameNamespaceIndex: epNameNamespaceIndexFunc, epIPIndex: epIPIndexFunc},
		object.DefaultProcessor(object.EndpointSliceToEndpoints),
	)

	dns.nsLister, dns.nsController = object.NewIndexerInformer(
		&cache.ListWatch{
			ListFunc:  namespaceListFunc(ctx, kubeClient),
			WatchFunc: namespaceWatchFunc(ctx, kubeClient),
		},
		&api.Namespace{},
		&dns,
		cache.Indexers{},
		object.DefaultProcessor(object.ToNamespace),
	)

	return &dns
}

func svcIPIndexFunc(obj interface{}) ([]string, error) {
	svc, ok := obj.(*object.Service)
	if !ok {
		return nil, errObj
	}
	idx := make([]string, len(svc.ClusterIPs))
	copy(idx, svc.ClusterIPs)
	return idx, nil
}

func epNameNamespaceIndexFunc(obj interface{}) ([]string, error) {
	s, ok := obj.(*object.Endpoints)
	if !ok {
		return nil, errObj
	}
	return []string{s.Index}, nil
}

func epIPIndexFunc(obj interface{}) ([]string, error) {
	ep, ok := obj.(*object.Endpoints)
	if !ok {
		return nil, errObj
	}
	idx := make([]string, len(ep.Addresses))
	for i, addr := range ep.Addresses {
		idx[i] = addr.IP
	}
	return idx, nil
}

func serviceListFunc(ctx context.Context, c kubernetes.Interface, ns string) func(meta.ListOptions) (runtime.Object, error) {
	return func(opts meta.ListOptions) (runtime.Object, error) {
		return c.CoreV1().Services(ns).List(ctx, opts)
	}
}

func endpointSliceListFunc(ctx context.Context, c kubernetes.Interface, ns string) func(meta.ListOptions) (runtime.Object, error) {
	return func(opts meta.ListOptions) (runtime.Object, error) {
		opts.LabelSelector = epLabelSelector
		return c.DiscoveryV1().EndpointSlices(ns).List(ctx, opts)
	}
}

func namespaceListFunc(ctx context.Context, c kubernetes.Interface) func(meta.ListOptions) (runtime.Object, error) {
	return func(opts meta.ListOptions) (runtime.Object, error) {
		return c.CoreV1().Namespaces().List(ctx, opts)
	}
}

func serviceWatchFunc(ctx context.Context, c kubernetes.Interface, ns string) func(options meta.ListOptions) (watch.Interface, error) {
	return func(options meta.ListOptions) (watch.Interface, error) {
		return c.CoreV1().Services(ns).Watch(ctx, options)
	}
}

func endpointSliceWatchFunc(ctx context.Context, c kubernetes.Interface, ns string) func(options meta.ListOptions) (watch.Interface, error) {
	return func(options meta.ListOptions) (watch.Interface, error) {
		options.LabelSelector = epLabelSelector
		return c.DiscoveryV1().EndpointSlices(ns).Watch(ctx, options)
	}
}

func namespaceWatchFunc(ctx context.Context, c kubernetes.Interface) func(options meta.ListOptions) (watch.Interface, error) {
	return func(options meta.ListOptions) (watch.Interface, error) {
		return c.CoreV1().Namespaces().Watch(ctx, options)
	}
}

// Start starts the controller.
func (dns *dnsControl) Start(stopCh <-chan struct{}) {
	go dns.svcController.Run(stopCh)
	go dns.epController.Run(stopCh)
	go dns.nsController.Run(stopCh)
}

// HasSynced returns true if the initial data has been
// completely loaded into memory.
func (dns *dnsControl) HasSynced() bool {
	if dns.hasSynced.Load() {
		return true
	}
	a := dns.svcController.HasSynced()
	b := dns.epController.HasSynced()
	c := dns.nsController.HasSynced()
	if a && b && c {
		dns.hasSynced.Store(true)
		return true
	}
	return false
}

func (dns *dnsControl) GetSvc(key string) *object.Service {
	o, exists, err := dns.svcLister.GetByKey(key)
	if err != nil || !exists {
		return nil
	}
	s, ok := o.(*object.Service)
	if ok {
		return s
	}
	return nil
}

func (dns *dnsControl) SvcIndexReverse(ip string) (svcs []*object.Service) {
	os, err := dns.svcLister.ByIndex(svcIPIndex, ip)
	if err != nil {
		return nil
	}
	svcs = make([]*object.Service, 0, len(os))
	for _, o := range os {
		s, ok := o.(*object.Service)
		if ok {
			svcs = append(svcs, s)
		}
	}
	return svcs
}

func (dns *dnsControl) EpIndex(idx string) (ep []*object.Endpoints) {
	os, err := dns.epLister.ByIndex(epNameNamespaceIndex, idx)
	if err != nil {
		return nil
	}
	ep = make([]*object.Endpoints, 0, len(os))
	for _, o := range os {
		e, ok := o.(*object.Endpoints)
		if ok {
			ep = append(ep, e)
		}
	}
	return ep
}

func (dns *dnsControl) EpIndexReverse(ip string) (ep []*object.Endpoints) {
	os, err := dns.epLister.ByIndex(epIPIndex, ip)
	if err != nil {
		return nil
	}
	ep = make([]*object.Endpoints, 0, len(os))
	for _, o := range os {
		e, ok := o.(*object.Endpoints)
		if ok {
			ep = append(ep, e)
		}
	}
	return ep
}

// NamespaceExists returns true if a namespace with this name exists.
func (dns *dnsControl) NamespaceExists(name string) bool {
	_, exists, _ := dns.nsLister.GetByKey(name)
	return exists
}

func (dns *dnsControl) OnAdd(obj interface{}, isInInitialList bool) {
	dns.updateModified()
	switch obj := obj.(type) {
	case *object.Endpoints:
		// Don't record latency during initial sync, because measuring latency only
		// makes sense for changes that happen while the service is running.
		if !isInInitialList {
			recordDNSProgrammingLatency(obj.LastChangeTriggerTime)
		}
	}
}

func (dns *dnsControl) OnDelete(obj interface{}) {
	dns.updateModified()
	// Note: We cannot record programming latency on deletes, because the trigger
	// time annotation is not updated when the object is deleted.
}

func (dns *dnsControl) OnUpdate(oldObj, newObj interface{}) {
	// If both objects have the same resource version, they are identical.
	if oldObj.(meta.Object).GetResourceVersion() == newObj.(meta.Object).GetResourceVersion() {
		return
	}

	switch newObj := newObj.(type) {
	case *object.Service:
		if object.ServiceModified(oldObj.(*object.Service), newObj) {
			dns.updateModified()
		}
	case *object.Endpoints:
		oldObj := oldObj.(*object.Endpoints)
		if object.EndpointsModified(oldObj, newObj) {
			dns.updateModified()
			// If the trigger time has not changed, the process that last updated the
			// object did not update the trigger time, so we can't know the latency.
			if oldObj.LastChangeTriggerTime != newObj.LastChangeTriggerTime {
				recordDNSProgrammingLatency(newObj.LastChangeTriggerTime)
			}
		}
	}
}

func (dns *dnsControl) Modified() int64 {
	return dns.modified.Load()
}

// updateModified set dns.modified to the current time.
func (dns *dnsControl) updateModified() {
	unix := time.Now().Unix()
	dns.modified.Store(unix)
}

var errObj = errors.New("obj was not of the correct type")