Fix k8s client (#379)

* Fix k8s client to use client-go

* Fix Kubernetes Build Issue

The client-go code requires you to vendor. I have done a hack here
in the Makefile to vendor it to version 1.5. But looks like we
will need to do this the 'right' way soon.

* Convert v1 to api Objects in List Functions

Also removed the endpoint controller which was not used for anything.
The Watch functions may still need the same treatment.

* Vendor client-go release-1.5

* Fix basic SRV feature

This is actually not serving SRV records correctly, but this should
get it to work as it did prior to the k8s client changes. Another
fix will be needed to serve SRV records as defined in the spec.

* Add additional output in test result

Add the response to the test output.

* Fix erroneous test data

4 files changed, 65 insertions, 68 deletions

diff --git a/middleware/kubernetes/controller.go b/middleware/kubernetes/controller.go
index 993078318..de6b98274 100644
--- a/middleware/kubernetes/controller.go
+++ b/middleware/kubernetes/controller.go
@@ -5,12 +5,13 @@ import (
 	"sync"
 	"time"
 
-	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/client/cache"
-	clientset_generated "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4"
-	"k8s.io/kubernetes/pkg/labels"
-	"k8s.io/kubernetes/pkg/runtime"
-	"k8s.io/kubernetes/pkg/watch"
+	"k8s.io/client-go/1.5/pkg/api"
+	"k8s.io/client-go/1.5/pkg/api/v1"
+	"k8s.io/client-go/1.5/tools/cache"
+	"k8s.io/client-go/1.5/kubernetes"
+	"k8s.io/client-go/1.5/pkg/labels"
+	"k8s.io/client-go/1.5/pkg/runtime"
+	"k8s.io/client-go/1.5/pkg/watch"
 )
 
 var (
@@ -31,16 +32,14 @@ func (s *storeToNamespaceLister) List() (ns api.NamespaceList, err error) {
 }
 
 type dnsController struct {
-	client *clientset_generated.Clientset
+	client *kubernetes.Clientset
 
 	selector *labels.Selector
 
-	endpController *cache.Controller
 	svcController  *cache.Controller
 	nsController   *cache.Controller
 
 	svcLister  cache.StoreToServiceLister
-	endpLister cache.StoreToEndpointsLister
 	nsLister   storeToNamespaceLister
 
 	// stopLock is used to enforce only a single call to Stop is active.
@@ -52,20 +51,13 @@ type dnsController struct {
 }
 
 // newDNSController creates a controller for CoreDNS.
-func newdnsController(kubeClient *clientset_generated.Clientset, resyncPeriod time.Duration, lselector *labels.Selector) *dnsController {
+func newdnsController(kubeClient *kubernetes.Clientset, resyncPeriod time.Duration, lselector *labels.Selector) *dnsController {
 	dns := dnsController{
 		client:   kubeClient,
 		selector: lselector,
 		stopCh:   make(chan struct{}),
 	}
 
-	dns.endpLister.Store, dns.endpController = cache.NewInformer(
-		&cache.ListWatch{
-			ListFunc:  endpointsListFunc(dns.client, namespace, dns.selector),
-			WatchFunc: endpointsWatchFunc(dns.client, namespace, dns.selector),
-		},
-		&api.Endpoints{}, resyncPeriod, cache.ResourceEventHandlerFuncs{})
-
 	dns.svcLister.Indexer, dns.svcController = cache.NewIndexerInformer(
 		&cache.ListWatch{
 			ListFunc:  serviceListFunc(dns.client, namespace, dns.selector),
@@ -86,62 +78,62 @@ func newdnsController(kubeClient *clientset_generated.Clientset, resyncPeriod ti
 	return &dns
 }
 
-func serviceListFunc(c *clientset_generated.Clientset, ns string, s *labels.Selector) func(api.ListOptions) (runtime.Object, error) {
+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 {
 			opts.LabelSelector = *s
 		}
-		return c.Services(ns).List(opts)
-	}
-}
-
-func serviceWatchFunc(c *clientset_generated.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
+		list_v1, err := c.Core().Services(ns).List(opts)
+		if err != nil {
+			return nil, err
 		}
-		return c.Services(ns).Watch(options)
-	}
-}
-
-func endpointsListFunc(c *clientset_generated.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
+		var list_api api.ServiceList
+		err = v1.Convert_v1_ServiceList_To_api_ServiceList(list_v1, &list_api, nil)
+		if err != nil {
+			return nil, err
 		}
-		return c.Endpoints(ns).List(opts)
+		return &list_api, err
 	}
 }
 
-func endpointsWatchFunc(c *clientset_generated.Clientset, ns string, s *labels.Selector) func(options api.ListOptions) (watch.Interface, error) {
+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.Endpoints(ns).Watch(options)
+		return c.Core().Services(ns).Watch(options)
 	}
 }
 
-func namespaceListFunc(c *clientset_generated.Clientset, s *labels.Selector) func(api.ListOptions) (runtime.Object, error) {
+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 {
 			opts.LabelSelector = *s
 		}
-		return c.Namespaces().List(opts)
+		list_v1, err := c.Core().Namespaces().List(opts)
+		if err != nil {
+			return nil, err
+		}
+		var list_api api.NamespaceList
+		err = v1.Convert_v1_NamespaceList_To_api_NamespaceList(list_v1, &list_api, nil)
+		if err != nil {
+			return nil, err
+		}
+		return &list_api, err
 	}
 }
 
-func namespaceWatchFunc(c *clientset_generated.Clientset, s *labels.Selector) func(options api.ListOptions) (watch.Interface, error) {
+func namespaceWatchFunc(c *kubernetes.Clientset, 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.Namespaces().Watch(options)
+		return c.Core().Namespaces().Watch(options)
 	}
 }
 
 func (dns *dnsController) controllersInSync() bool {
-	return dns.svcController.HasSynced() && dns.endpController.HasSynced()
+	return dns.svcController.HasSynced()
 }
 
 // Stop stops the  controller.
@@ -162,7 +154,6 @@ func (dns *dnsController) Stop() error {
 
 // Run starts the controller.
 func (dns *dnsController) Run() {
-	go dns.endpController.Run(dns.stopCh)
 	go dns.svcController.Run(dns.stopCh)
 	go dns.nsController.Run(dns.stopCh)
 	<-dns.stopCh
diff --git a/middleware/kubernetes/kubernetes.go b/middleware/kubernetes/kubernetes.go
index 43d98e079..05359dc69 100644
--- a/middleware/kubernetes/kubernetes.go
+++ b/middleware/kubernetes/kubernetes.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"strconv"
 	"strings"
 	"time"
 
@@ -17,13 +18,13 @@ import (
 	"github.com/miekg/coredns/request"
 
 	"github.com/miekg/dns"
-	"k8s.io/kubernetes/pkg/api"
-	unversionedapi "k8s.io/kubernetes/pkg/api/unversioned"
-	clientset_generated "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4"
-	"k8s.io/kubernetes/pkg/client/restclient"
-	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
-	clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
-	"k8s.io/kubernetes/pkg/labels"
+	"k8s.io/client-go/1.5/pkg/api"
+	unversionedapi "k8s.io/client-go/1.5/pkg/api/unversioned"
+	"k8s.io/client-go/1.5/kubernetes"
+	"k8s.io/client-go/1.5/rest"
+	"k8s.io/client-go/1.5/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/1.5/tools/clientcmd/api"
+	"k8s.io/client-go/1.5/pkg/labels"
 )
 
 // Kubernetes implements a middleware that connects to a Kubernetes cluster.
@@ -65,7 +66,7 @@ func (k *Kubernetes) Debug() string {
 	return "debug"
 }
 
-func (k *Kubernetes) getClientConfig() (*restclient.Config, error) {
+func (k *Kubernetes) getClientConfig() (*rest.Config, error) {
 	// For a custom api server or running outside a k8s cluster
 	// set URL in env.KUBERNETES_MASTER or set endpoint in Corefile
 	loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
@@ -75,7 +76,7 @@ func (k *Kubernetes) getClientConfig() (*restclient.Config, error) {
 	if len(k.APIEndpoint) > 0 {
 		clusterinfo.Server = k.APIEndpoint
 	} else {
-		cc, err := restclient.InClusterConfig()
+		cc, err := rest.InClusterConfig()
 		if err != nil {
 			return nil, err
 		}
@@ -104,7 +105,7 @@ func (k *Kubernetes) InitKubeCache() error {
 		return err
 	}
 
-	kubeClient, err := clientset_generated.NewForConfig(config)
+	kubeClient, err := kubernetes.NewForConfig(config)
 	if err != nil {
 		return fmt.Errorf("Failed to create kubernetes notification controller: %v", err)
 	}
@@ -198,7 +199,7 @@ func (k *Kubernetes) Records(name string, exact bool) ([]msg.Service, error) {
 		return nil, nil
 	}
 
-	k8sItems, err := k.Get(namespace, nsWildcard, serviceName, serviceWildcard)
+	k8sItems, err := k.Get(namespace, nsWildcard, serviceName, serviceWildcard, typeName)
 	if err != nil {
 		return nil, err
 	}
@@ -207,26 +208,22 @@ func (k *Kubernetes) Records(name string, exact bool) ([]msg.Service, error) {
 		return nil, nil
 	}
 
-	records := k.getRecordsForServiceItems(k8sItems, nametemplate.NameValues{TypeName: typeName, ServiceName: serviceName, Namespace: namespace, Zone: zone})
+	records := k.getRecordsForServiceItems(k8sItems, zone)
 	return records, nil
 }
 
-// TODO: assemble name from parts found in k8s data based on name template rather than reusing query string
-func (k *Kubernetes) getRecordsForServiceItems(serviceItems []*api.Service, values nametemplate.NameValues) []msg.Service {
+func (k *Kubernetes) getRecordsForServiceItems(serviceItems []*api.Service, zone string) []msg.Service {
 	var records []msg.Service
 
 	for _, item := range serviceItems {
 		clusterIP := item.Spec.ClusterIP
 
-		// Create records by constructing record name from template...
-		//values.Namespace = item.Metadata.Namespace
-		//values.ServiceName = item.Metadata.Name
-		//s := msg.Service{Host: g.NameTemplate.GetRecordNameFromNameValues(values)}
-		//records = append(records, s)
-
 		// Create records for each exposed port...
-		for _, p := range item.Spec.Ports {
-			s := msg.Service{Host: clusterIP, Port: int(p.Port)}
+		key := k.NameTemplate.RecordNameFromNameValues(nametemplate.NameValues{TypeName: "svc", ServiceName: item.ObjectMeta.Name, Namespace: item.ObjectMeta.Namespace, Zone: zone})
+		key = strings.Replace(key, ".", "/", -1)
+
+		for i, p := range item.Spec.Ports {
+			s := msg.Service{Key: msg.Path(strconv.Itoa(i) + "." + key, "coredns"), Host: clusterIP, Port: int(p.Port)}
 			records = append(records, s)
 		}
 	}
@@ -235,7 +232,16 @@ func (k *Kubernetes) getRecordsForServiceItems(serviceItems []*api.Service, valu
 }
 
 // Get performs the call to the Kubernetes http API.
-func (k *Kubernetes) Get(namespace string, nsWildcard bool, servicename string, serviceWildcard bool) ([]*api.Service, error) {
+func (k *Kubernetes) Get(namespace string, nsWildcard bool, servicename string, serviceWildcard bool, typeName string) ([]*api.Service, error) {
+	switch {
+	case typeName == "pod":
+		return nil, fmt.Errorf("pod not implemented")
+	default:
+		return k.getServices(namespace, nsWildcard, servicename, serviceWildcard)
+	}
+}
+
+func (k *Kubernetes) getServices(namespace string, nsWildcard bool, servicename string, serviceWildcard bool) ([]*api.Service, error) {
 	serviceList := k.APIConn.ServiceList()
 
 	var resultItems []*api.Service
diff --git a/middleware/kubernetes/setup.go b/middleware/kubernetes/setup.go
index 98afc8334..4d303b25c 100644
--- a/middleware/kubernetes/setup.go
+++ b/middleware/kubernetes/setup.go
@@ -11,7 +11,7 @@ import (
 	"github.com/miekg/coredns/middleware/kubernetes/nametemplate"
 
 	"github.com/mholt/caddy"
-	unversionedapi "k8s.io/kubernetes/pkg/api/unversioned"
+	unversionedapi "k8s.io/client-go/1.5/pkg/api/unversioned"
 )
 
 func init() {
diff --git a/middleware/kubernetes/setup_test.go b/middleware/kubernetes/setup_test.go
index 5d2154350..72b60e5ce 100644
--- a/middleware/kubernetes/setup_test.go
+++ b/middleware/kubernetes/setup_test.go
@@ -6,7 +6,7 @@ import (
 	"time"
 
 	"github.com/mholt/caddy"
-	unversionedapi "k8s.io/kubernetes/pkg/api/unversioned"
+	unversionedapi "k8s.io/client-go/1.5/pkg/api/unversioned"
 )
 
 func TestKubernetesParse(t *testing.T) {