From 4318dfbf02ca4248dab29fc4fb730e2515a05754 Mon Sep 17 00:00:00 2001
From: John Belamaric <jbelamaric@infoblox.com>
Date: Mon, 7 Nov 2016 11:21:24 -0500
Subject: [PATCH] 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.
---
 middleware/kubernetes/controller.go | 41 +++++++++++++++++++++++++++--
 1 file 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
 	}
 }