plugin/kubernetes: allow trimming down of cached items. (#2128)

* Convert to runtime.Object to smaller structs

This adds conversion for all the objects we want to keep in the cache.
It keeps the minimum for CoreDNS to function and throws away the rest.

The conversion:
api.Endpoints -> object.Endpoints
api.Pod       -> object.Pod
api.Serivce   -> object.Service

We needed to copy some client-go stuff to insert a conversion function
into NewIndexInformers.

Some unrelated cleanups in the watch functionality as that needed to be
touched because of the above translation of objects.

Signed-off-by: Miek Gieben <miek@miek.nl>

* Reduce test line-count

Signed-off-by: Miek Gieben <miek@miek.nl>

* ....and fix test

Signed-off-by: Miek Gieben <miek@miek.nl>

* Drop use of append

Signed-off-by: Miek Gieben <miek@miek.nl>

* cosmetic changes

Signed-off-by: Miek Gieben <miek@miek.nl>

* that was a typo

Signed-off-by: Miek Gieben <miek@miek.nl>

* re-introduce append here

We can't really use len() here because we don't know the number before
hand.

Signed-off-by: Miek Gieben <miek@miek.nl>

* comment in better place

Signed-off-by: Miek Gieben <miek@miek.nl>

* Make the timestamp a bool; thats where it is used for

Signed-off-by: Miek Gieben <miek@miek.nl>

* Set incoming object to nil

Explicataliy discard the converted object; we did a deep copy it's
not needed anymore.

Signed-off-by: Miek Gieben <miek@miek.nl>

* Per Chris's comment

Signed-off-by: Miek Gieben <miek@miek.nl>

plugin/kubernetes/object/endpoint.go

@@ -0,0 +1,162 @@
+package object
+
+import (
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+// Endpoints is a stripped down api.Endpoints with only the items we need for CoreDNS.
+type Endpoints struct {
+	Version   string
+	Name      string
+	Namespace string
+	Index     string
+	IndexIP   []string
+	Subsets   []EndpointSubset
+
+	*Empty
+}
+
+// EndpointSubset is a group of addresses with a common set of ports. The
+// expanded set of endpoints is the Cartesian product of Addresses x Ports.
+type EndpointSubset struct {
+	Addresses []EndpointAddress
+	Ports     []EndpointPort
+}
+
+// EndpointAddress is a tuple that describes single IP address.
+type EndpointAddress struct {
+	IP            string
+	Hostname      string
+	NodeName      string
+	TargetRefName string
+}
+
+// EndpointPort is a tuple that describes a single port.
+type EndpointPort struct {
+	Port     int32
+	Name     string
+	Protocol string
+}
+
+// EndpointsKey return a string using for the index.
+func EndpointsKey(name, namespace string) string { return name + "." + namespace }
+
+// ToEndpoints converts an api.Service to a *Service.
+func ToEndpoints(obj interface{}) interface{} {
+	end, ok := obj.(*api.Endpoints)
+	if !ok {
+		return nil
+	}
+
+	e := &Endpoints{
+		Version:   end.GetResourceVersion(),
+		Name:      end.GetName(),
+		Namespace: end.GetNamespace(),
+		Index:     EndpointsKey(end.GetName(), end.GetNamespace()),
+		Subsets:   make([]EndpointSubset, len(end.Subsets)),
+	}
+	for i, eps := range end.Subsets {
+		sub := EndpointSubset{
+			Addresses: make([]EndpointAddress, len(eps.Addresses)),
+		}
+		if len(eps.Ports) == 0 {
+			// Add sentinal if there are no ports.
+			sub.Ports = []EndpointPort{{Port: -1}}
+		} else {
+			sub.Ports = make([]EndpointPort, len(eps.Ports))
+		}
+
+		for j, a := range eps.Addresses {
+			ea := EndpointAddress{IP: a.IP, Hostname: a.Hostname}
+			if a.NodeName != nil {
+				ea.NodeName = *a.NodeName
+			}
+			if a.TargetRef != nil {
+				ea.TargetRefName = a.TargetRef.Name
+			}
+			sub.Addresses[j] = ea
+		}
+
+		for k, p := range eps.Ports {
+			ep := EndpointPort{Port: p.Port, Name: p.Name, Protocol: string(p.Protocol)}
+			sub.Ports[k] = ep
+		}
+
+		e.Subsets[i] = sub
+	}
+
+	for _, eps := range end.Subsets {
+		for _, a := range eps.Addresses {
+			e.IndexIP = append(e.IndexIP, a.IP)
+		}
+	}
+
+	*end = api.Endpoints{}
+
+	return e
+}
+
+// CopyWithoutSubsets copies e, without the subsets.
+func (e *Endpoints) CopyWithoutSubsets() *Endpoints {
+	e1 := &Endpoints{
+		Version:   e.Version,
+		Name:      e.Name,
+		Namespace: e.Namespace,
+		Index:     e.Index,
+		IndexIP:   make([]string, len(e.IndexIP)),
+	}
+	copy(e1.IndexIP, e.IndexIP)
+	return e1
+}
+
+var _ runtime.Object = &Endpoints{}
+
+// DeepCopyObject implements the ObjectKind interface.
+func (e *Endpoints) DeepCopyObject() runtime.Object {
+	e1 := &Endpoints{
+		Version:   e.Version,
+		Name:      e.Name,
+		Namespace: e.Namespace,
+		Index:     e.Index,
+		IndexIP:   make([]string, len(e.IndexIP)),
+		Subsets:   make([]EndpointSubset, len(e.Subsets)),
+	}
+	copy(e1.IndexIP, e.IndexIP)
+
+	for i, eps := range e.Subsets {
+		sub := EndpointSubset{
+			Addresses: make([]EndpointAddress, len(eps.Addresses)),
+			Ports:     make([]EndpointPort, len(eps.Ports)),
+		}
+		for j, a := range eps.Addresses {
+			ea := EndpointAddress{IP: a.IP, Hostname: a.Hostname, NodeName: a.NodeName, TargetRefName: a.TargetRefName}
+			sub.Addresses[j] = ea
+		}
+		for k, p := range eps.Ports {
+			ep := EndpointPort{Port: p.Port, Name: p.Name, Protocol: p.Protocol}
+			sub.Ports[k] = ep
+		}
+
+		e1.Subsets[i] = sub
+	}
+	return e1
+}
+
+// GetNamespace implements the metav1.Object interface.
+func (e *Endpoints) GetNamespace() string { return e.Namespace }
+
+// SetNamespace implements the metav1.Object interface.
+func (e *Endpoints) SetNamespace(namespace string) {}
+
+// GetName implements the metav1.Object interface.
+func (e *Endpoints) GetName() string { return e.Name }
+
+// SetName implements the metav1.Object interface.
+func (e *Endpoints) SetName(name string) {}
+
+// GetResourceVersion implements the metav1.Object interface.
+func (e *Endpoints) GetResourceVersion() string { return e.Version }
+
+// SetResourceVersion implements the metav1.Object interface.
+func (e *Endpoints) SetResourceVersion(version string) {}

plugin/kubernetes/object/informer.go

@@ -0,0 +1,51 @@
+package object
+
+import (
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/cache"
+)
+
+// NewIndexerInformer is a copy of the cache.NewIndexInformer function, but allows Process to have a conversion function (ToFunc).
+func NewIndexerInformer(lw cache.ListerWatcher, objType runtime.Object, resyncPeriod time.Duration, h cache.ResourceEventHandler, indexers cache.Indexers, convert ToFunc) (cache.Indexer, cache.Controller) {
+	clientState := cache.NewIndexer(cache.DeletionHandlingMetaNamespaceKeyFunc, indexers)
+
+	fifo := cache.NewDeltaFIFO(cache.MetaNamespaceKeyFunc, clientState)
+
+	cfg := &cache.Config{
+		Queue:            fifo,
+		ListerWatcher:    lw,
+		ObjectType:       objType,
+		FullResyncPeriod: resyncPeriod,
+		RetryOnError:     false,
+		Process: func(obj interface{}) error {
+			for _, d := range obj.(cache.Deltas) {
+
+				obj := convert(d.Object)
+
+				switch d.Type {
+				case cache.Sync, cache.Added, cache.Updated:
+					if old, exists, err := clientState.Get(obj); err == nil && exists {
+						if err := clientState.Update(obj); err != nil {
+							return err
+						}
+						h.OnUpdate(old, obj)
+					} else {
+						if err := clientState.Add(obj); err != nil {
+							return err
+						}
+						h.OnAdd(obj)
+					}
+				case cache.Deleted:
+					if err := clientState.Delete(obj); err != nil {
+						return err
+					}
+					h.OnDelete(obj)
+				}
+			}
+			return nil
+		},
+	}
+	return clientState, cache.New(cfg)
+}

plugin/kubernetes/object/object.go

@@ -0,0 +1,94 @@
+package object
+
+import (
+	"k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// ToFunc converts one empty interface to another.
+type ToFunc func(interface{}) interface{}
+
+// Empty is an empty struct.
+type Empty struct{}
+
+// GetObjectKind implementss the ObjectKind interface as a noop.
+func (e *Empty) GetObjectKind() schema.ObjectKind { return schema.EmptyObjectKind }
+
+// GetGenerateName implements the metav1.Object interface.
+func (e *Empty) GetGenerateName() string { return "" }
+
+// SetGenerateName implements the metav1.Object interface.
+func (e *Empty) SetGenerateName(name string) {}
+
+// GetUID implements the metav1.Object interface.
+func (e *Empty) GetUID() types.UID { return "" }
+
+// SetUID implements the metav1.Object interface.
+func (e *Empty) SetUID(uid types.UID) {}
+
+// GetGeneration implements the metav1.Object interface.
+func (e *Empty) GetGeneration() int64 { return 0 }
+
+// SetGeneration implements the metav1.Object interface.
+func (e *Empty) SetGeneration(generation int64) {}
+
+// GetSelfLink implements the metav1.Object interface.
+func (e *Empty) GetSelfLink() string { return "" }
+
+// SetSelfLink implements the metav1.Object interface.
+func (e *Empty) SetSelfLink(selfLink string) {}
+
+// GetCreationTimestamp implements the metav1.Object interface.
+func (e *Empty) GetCreationTimestamp() v1.Time { return v1.Time{} }
+
+// SetCreationTimestamp implements the metav1.Object interface.
+func (e *Empty) SetCreationTimestamp(timestamp v1.Time) {}
+
+// GetDeletionTimestamp implements the metav1.Object interface.
+func (e *Empty) GetDeletionTimestamp() *v1.Time { return &v1.Time{} }
+
+// SetDeletionTimestamp implements the metav1.Object interface.
+func (e *Empty) SetDeletionTimestamp(timestamp *v1.Time) {}
+
+// GetDeletionGracePeriodSeconds implements the metav1.Object interface.
+func (e *Empty) GetDeletionGracePeriodSeconds() *int64 { return nil }
+
+// SetDeletionGracePeriodSeconds implements the metav1.Object interface.
+func (e *Empty) SetDeletionGracePeriodSeconds(*int64) {}
+
+// GetLabels implements the metav1.Object interface.
+func (e *Empty) GetLabels() map[string]string { return nil }
+
+// SetLabels implements the metav1.Object interface.
+func (e *Empty) SetLabels(labels map[string]string) {}
+
+// GetAnnotations implements the metav1.Object interface.
+func (e *Empty) GetAnnotations() map[string]string { return nil }
+
+// SetAnnotations implements the metav1.Object interface.
+func (e *Empty) SetAnnotations(annotations map[string]string) {}
+
+// GetInitializers implements the metav1.Object interface.
+func (e *Empty) GetInitializers() *v1.Initializers { return nil }
+
+// SetInitializers implements the metav1.Object interface.
+func (e *Empty) SetInitializers(initializers *v1.Initializers) {}
+
+// GetFinalizers implements the metav1.Object interface.
+func (e *Empty) GetFinalizers() []string { return nil }
+
+// SetFinalizers implements the metav1.Object interface.
+func (e *Empty) SetFinalizers(finalizers []string) {}
+
+// GetOwnerReferences implements the metav1.Object interface.
+func (e *Empty) GetOwnerReferences() []v1.OwnerReference { return nil }
+
+// SetOwnerReferences implements the metav1.Object interface.
+func (e *Empty) SetOwnerReferences([]v1.OwnerReference) {}
+
+// GetClusterName implements the metav1.Object interface.
+func (e *Empty) GetClusterName() string { return "" }
+
+// SetClusterName implements the metav1.Object interface.
+func (e *Empty) SetClusterName(clusterName string) {}

plugin/kubernetes/object/pod.go

@@ -0,0 +1,72 @@
+package object
+
+import (
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+// Pod is a stripped down api.Pod with only the items we need for CoreDNS.
+type Pod struct {
+	Version   string
+	PodIP     string
+	Name      string
+	Namespace string
+	Deleting  bool
+
+	*Empty
+}
+
+// ToPod converts an api.Pod to a *Pod.
+func ToPod(obj interface{}) interface{} {
+	pod, ok := obj.(*api.Pod)
+	if !ok {
+		return nil
+	}
+
+	p := &Pod{
+		Version:   pod.GetResourceVersion(),
+		PodIP:     pod.Status.PodIP,
+		Namespace: pod.GetNamespace(),
+		Name:      pod.GetName(),
+	}
+	t := pod.ObjectMeta.DeletionTimestamp
+	if t != nil {
+		p.Deleting = !(*t).Time.IsZero()
+	}
+
+	*pod = api.Pod{}
+
+	return p
+}
+
+var _ runtime.Object = &Pod{}
+
+// DeepCopyObject implements the ObjectKind interface.
+func (p *Pod) DeepCopyObject() runtime.Object {
+	p1 := &Pod{
+		Version:   p.Version,
+		PodIP:     p.PodIP,
+		Namespace: p.Namespace,
+		Name:      p.Name,
+		Deleting:  p.Deleting,
+	}
+	return p1
+}
+
+// GetNamespace implements the metav1.Object interface.
+func (p *Pod) GetNamespace() string { return p.Namespace }
+
+// SetNamespace implements the metav1.Object interface.
+func (p *Pod) SetNamespace(namespace string) {}
+
+// GetName implements the metav1.Object interface.
+func (p *Pod) GetName() string { return p.Name }
+
+// SetName implements the metav1.Object interface.
+func (p *Pod) SetName(name string) {}
+
+// GetResourceVersion implements the metav1.Object interface.
+func (p *Pod) GetResourceVersion() string { return p.Version }
+
+// SetResourceVersion implements the metav1.Object interface.
+func (p *Pod) SetResourceVersion(version string) {}

plugin/kubernetes/object/service.go

@@ -0,0 +1,89 @@
+package object
+
+import (
+	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+// Service is a stripped down api.Service with only the items we need for CoreDNS.
+type Service struct {
+	Version      string
+	Name         string
+	Namespace    string
+	Index        string
+	ClusterIP    string
+	Type         api.ServiceType
+	ExternalName string
+	Ports        []api.ServicePort
+
+	*Empty
+}
+
+// ServiceKey return a string using for the index.
+func ServiceKey(name, namespace string) string { return name + "." + namespace }
+
+// ToService converts an api.Service to a *Service.
+func ToService(obj interface{}) interface{} {
+	svc, ok := obj.(*api.Service)
+	if !ok {
+		return nil
+	}
+
+	s := &Service{
+		Version:      svc.GetResourceVersion(),
+		Name:         svc.GetName(),
+		Namespace:    svc.GetNamespace(),
+		Index:        ServiceKey(svc.GetName(), svc.GetNamespace()),
+		ClusterIP:    svc.Spec.ClusterIP,
+		Type:         svc.Spec.Type,
+		ExternalName: svc.Spec.ExternalName,
+	}
+
+	if len(svc.Spec.Ports) == 0 {
+		// Add sentinal if there are no ports.
+		s.Ports = []api.ServicePort{{Port: -1}}
+	} else {
+		s.Ports = make([]api.ServicePort, len(svc.Spec.Ports))
+		copy(s.Ports, svc.Spec.Ports)
+	}
+
+	*svc = api.Service{}
+
+	return s
+}
+
+var _ runtime.Object = &Service{}
+
+// DeepCopyObject implements the ObjectKind interface.
+func (s *Service) DeepCopyObject() runtime.Object {
+	s1 := &Service{
+		Version:      s.Version,
+		Name:         s.Name,
+		Namespace:    s.Namespace,
+		Index:        s.Index,
+		ClusterIP:    s.ClusterIP,
+		Type:         s.Type,
+		ExternalName: s.ExternalName,
+		Ports:        make([]api.ServicePort, len(s.Ports)),
+	}
+	copy(s1.Ports, s.Ports)
+	return s1
+}
+
+// GetNamespace implements the metav1.Object interface.
+func (s *Service) GetNamespace() string { return s.Namespace }
+
+// SetNamespace implements the metav1.Object interface.
+func (s *Service) SetNamespace(namespace string) {}
+
+// GetName implements the metav1.Object interface.
+func (s *Service) GetName() string { return s.Name }
+
+// SetName implements the metav1.Object interface.
+func (s *Service) SetName(name string) {}
+
+// GetResourceVersion implements the metav1.Object interface.
+func (s *Service) GetResourceVersion() string { return s.Version }
+
+// SetResourceVersion implements the metav1.Object interface.
+func (s *Service) SetResourceVersion(version string) {}