Add pod cache and verified pod responses (#483)

* Add pod cache and verified pod responses

* add ip indexing for pod cache

1 files changed, 71 insertions, 1 deletions

diff --git a/middleware/kubernetes/controller.go b/middleware/kubernetes/controller.go
index 150cc843f..e6e79366a 100644
--- a/middleware/kubernetes/controller.go
+++ b/middleware/kubernetes/controller.go
@@ -1,6 +1,7 @@
 package kubernetes
 
 import (
+	"errors"
 	"fmt"
 	"log"
 	"sync"
@@ -24,6 +25,8 @@ type storeToNamespaceLister struct {
 	cache.Store
 }
 
+const podIPIndex = "PodIP"
+
 // List lists all Namespaces in the store.
 func (s *storeToNamespaceLister) List() (ns api.NamespaceList, err error) {
 	for _, m := range s.Store.List() {
@@ -38,10 +41,12 @@ type dnsController struct {
 	selector *labels.Selector
 
 	svcController *cache.Controller
+	podController *cache.Controller
 	nsController  *cache.Controller
 	epController  *cache.Controller
 
 	svcLister cache.StoreToServiceLister
+	podLister cache.StoreToPodLister
 	nsLister  storeToNamespaceLister
 	epLister  cache.StoreToEndpointsLister
 
@@ -54,7 +59,7 @@ type dnsController struct {
 }
 
 // newDNSController creates a controller for CoreDNS.
-func newdnsController(kubeClient *kubernetes.Clientset, resyncPeriod time.Duration, lselector *labels.Selector) *dnsController {
+func newdnsController(kubeClient *kubernetes.Clientset, resyncPeriod time.Duration, lselector *labels.Selector, initPodCache bool) *dnsController {
 	dns := dnsController{
 		client:   kubeClient,
 		selector: lselector,
@@ -71,6 +76,18 @@ func newdnsController(kubeClient *kubernetes.Clientset, resyncPeriod time.Durati
 		cache.ResourceEventHandlerFuncs{},
 		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
 
+	if initPodCache {
+		dns.podLister.Indexer, dns.podController = cache.NewIndexerInformer(
+			&cache.ListWatch{
+				ListFunc:  podListFunc(dns.client, namespace, dns.selector),
+				WatchFunc: podWatchFunc(dns.client, namespace, dns.selector),
+			},
+			&api.Pod{}, // TODO replace with a lighter-weight custom struct
+			resyncPeriod,
+			cache.ResourceEventHandlerFuncs{},
+			cache.Indexers{podIPIndex: podIPIndexFunc})
+	}
+
 	dns.nsLister.Store, dns.nsController = cache.NewInformer(
 		&cache.ListWatch{
 			ListFunc:  namespaceListFunc(dns.client, dns.selector),
@@ -88,6 +105,14 @@ func newdnsController(kubeClient *kubernetes.Clientset, resyncPeriod time.Durati
 	return &dns
 }
 
+func podIPIndexFunc(obj interface{}) ([]string, error) {
+	p, ok := obj.(*api.Pod)
+	if !ok {
+		return nil, errors.New("obj was not an *api.Pod")
+	}
+	return []string{p.Status.PodIP}, nil
+}
+
 func serviceListFunc(c *kubernetes.Clientset, ns string, s *labels.Selector) func(api.ListOptions) (runtime.Object, error) {
 	return func(opts api.ListOptions) (runtime.Object, error) {
 		if s != nil {
@@ -107,6 +132,26 @@ func serviceListFunc(c *kubernetes.Clientset, ns string, s *labels.Selector) fun
 	}
 }
 
+func podListFunc(c *kubernetes.Clientset, ns string, s *labels.Selector) func(api.ListOptions) (runtime.Object, error) {
+	return func(opts api.ListOptions) (runtime.Object, error) {
+		if s != nil {
+			opts.LabelSelector = *s
+		}
+		listV1, err := c.Core().Pods(ns).List(opts)
+
+		if err != nil {
+			return nil, err
+		}
+		var listAPI api.PodList
+		err = v1.Convert_v1_PodList_To_api_PodList(listV1, &listAPI, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		return &listAPI, err
+	}
+}
+
 func v1ToAPIFilter(in watch.Event) (out watch.Event, keep bool) {
 	if in.Type == watch.Error {
 		return in, true
@@ -121,6 +166,14 @@ func v1ToAPIFilter(in watch.Event) (out watch.Event, keep bool) {
 			return in, true
 		}
 		return watch.Event{Type: in.Type, Object: &apiObj}, true
+	case *v1.Pod:
+		var apiObj api.Pod
+		err := v1.Convert_v1_Pod_To_api_Pod(v1Obj, &apiObj, nil)
+		if err != nil {
+			log.Printf("[ERROR] Could not convert v1.Pod: %s", err)
+			return in, true
+		}
+		return watch.Event{Type: in.Type, Object: &apiObj}, true
 	case *v1.Namespace:
 		var apiObj api.Namespace
 		err := v1.Convert_v1_Namespace_To_api_Namespace(v1Obj, &apiObj, nil)
@@ -156,6 +209,20 @@ func serviceWatchFunc(c *kubernetes.Clientset, ns string, s *labels.Selector) fu
 	}
 }
 
+func podWatchFunc(c *kubernetes.Clientset, ns string, s *labels.Selector) func(options api.ListOptions) (watch.Interface, error) {
+	return func(options api.ListOptions) (watch.Interface, error) {
+		if s != nil {
+			options.LabelSelector = *s
+		}
+		w, err := c.Core().Pods(ns).Watch(options)
+
+		if err != nil {
+			return nil, err
+		}
+		return watch.Filter(w, v1ToAPIFilter), nil
+	}
+}
+
 func namespaceListFunc(c *kubernetes.Clientset, s *labels.Selector) func(api.ListOptions) (runtime.Object, error) {
 	return func(opts api.ListOptions) (runtime.Object, error) {
 		if s != nil {
@@ -244,6 +311,9 @@ func (dns *dnsController) Run() {
 	go dns.svcController.Run(dns.stopCh)
 	go dns.nsController.Run(dns.stopCh)
 	go dns.epController.Run(dns.stopCh)
+	if dns.podController != nil {
+		go dns.podController.Run(dns.stopCh)
+	}
 	<-dns.stopCh
 }