Fix Kubernetes Watches (#392)

The watchers were still trying to process raw v1 objects which
failed to be added to the store. This meant new services and
namespaces created after CoreDNS started would not be discoverable.

Add a filter function that converts watch events with v1 objects
to events with api objects.

1 files changed, 39 insertions, 2 deletions

diff --git a/middleware/kubernetes/controller.go b/middleware/kubernetes/controller.go
index 33d197a84..edf97c31b 100644
--- a/middleware/kubernetes/controller.go
+++ b/middleware/kubernetes/controller.go
@@ -2,6 +2,7 @@ package kubernetes
 
 import (
 	"fmt"
+	"log"
 	"sync"
 	"time"
 
@@ -96,12 +97,44 @@ func serviceListFunc(c *kubernetes.Clientset, ns string, s *labels.Selector) fun
 	}
 }
 
+func v1ToApiFilter(in watch.Event) (out watch.Event, keep bool) {
+	if in.Type == watch.Error {
+		return in, true
+	}
+
+	switch v1Obj := in.Object.(type) {
+		case *v1.Service:
+			var apiObj api.Service
+			err := v1.Convert_v1_Service_To_api_Service(v1Obj, &apiObj, nil)
+			if err != nil {
+				log.Printf("[ERROR] Could not convert v1.Service: %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)
+			if err != nil {
+				log.Printf("[ERROR] Could not convert v1.Namespace: %s", err)
+				return in, true
+			}
+			return watch.Event{Type: in.Type, Object: &apiObj}, true
+	}
+
+	log.Printf("[WARN] Unhandled v1 type in event: %v", in)
+	return in, true
+}
+
 func serviceWatchFunc(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
 		}
-		return c.Core().Services(ns).Watch(options)
+		w, err := c.Core().Services(ns).Watch(options)
+		if err != nil {
+			return nil, err
+		}
+		return watch.Filter(w, v1ToApiFilter), nil
 	}
 }
 
@@ -128,7 +161,11 @@ func namespaceWatchFunc(c *kubernetes.Clientset, s *labels.Selector) func(option
 		if s != nil {
 			options.LabelSelector = *s
 		}
-		return c.Core().Namespaces().Watch(options)
+		w, err := c.Core().Namespaces().Watch(options)
+		if err != nil {
+			return nil, err
+		}
+		return watch.Filter(w, v1ToApiFilter), nil
 	}
 }