plugin/kubernetes: implement HasSynced() (#1155)

* plugin/kubernetes: wait until api is ready

Wait for HasSynced before allowing startup to avoid startup race.

Also do a small refactor in findServices() to pull a check out of the
loop - only needs to be done once.

* sigh

8 files changed, 29 insertions, 19 deletions

diff --git a/plugin/kubernetes/apiproxy.go b/plugin/kubernetes/apiproxy.go
index 196ca5e60..59a569885 100644
--- a/plugin/kubernetes/apiproxy.go
+++ b/plugin/kubernetes/apiproxy.go
@@ -65,7 +65,9 @@ func (p *proxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 func (p *apiProxy) Run() {
 	p.handler.Start()
-	p.Serve(p.listener)
+	go func() {
+		p.Serve(p.listener)
+	}()
 }
 
 func (p *apiProxy) Stop() {
diff --git a/plugin/kubernetes/controller.go b/plugin/kubernetes/controller.go
index a3145bf3d..3cad9e6c2 100644
--- a/plugin/kubernetes/controller.go
+++ b/plugin/kubernetes/controller.go
@@ -38,6 +38,7 @@ type dnsController interface {
 	GetNodeByName(string) (*api.Node, error)
 
 	Run()
+	HasSynced() bool
 	Stop() error
 }
 
@@ -229,17 +230,6 @@ func endpointsWatchFunc(c *kubernetes.Clientset, ns string, s *labels.Selector)
 	}
 }
 
-func (dns *dnsControl) controllersInSync() bool {
-	hs := dns.svcController.HasSynced() &&
-		dns.epController.HasSynced()
-
-	if dns.podController != nil {
-		hs = hs && dns.podController.HasSynced()
-	}
-
-	return hs
-}
-
 // Stop stops the  controller.
 func (dns *dnsControl) Stop() error {
 	dns.stopLock.Lock()
@@ -266,6 +256,17 @@ func (dns *dnsControl) Run() {
 	<-dns.stopCh
 }
 
+// HasSynced calls on all controllers.
+func (dns *dnsControl) HasSynced() bool {
+	a := dns.svcController.HasSynced()
+	b := dns.epController.HasSynced()
+	c := true
+	if dns.podController != nil {
+		c = dns.podController.HasSynced()
+	}
+	return a && b && c
+}
+
 func (dns *dnsControl) ServiceList() (svcs []*api.Service) {
 	os := dns.svcLister.List()
 	for _, o := range os {
diff --git a/plugin/kubernetes/handler_test.go b/plugin/kubernetes/handler_test.go
index 77edae236..5e9743b68 100644
--- a/plugin/kubernetes/handler_test.go
+++ b/plugin/kubernetes/handler_test.go
@@ -191,6 +191,7 @@ func TestServeDNS(t *testing.T) {
 
 type APIConnServeTest struct{}
 
+func (APIConnServeTest) HasSynced() bool                        { return true }
 func (APIConnServeTest) Run()                                   { return }
 func (APIConnServeTest) Stop() error                            { return nil }
 func (APIConnServeTest) EpIndexReverse(string) []*api.Endpoints { return nil }
diff --git a/plugin/kubernetes/kubernetes.go b/plugin/kubernetes/kubernetes.go
index f51ee8e3b..11b6cc882 100644
--- a/plugin/kubernetes/kubernetes.go
+++ b/plugin/kubernetes/kubernetes.go
@@ -338,10 +338,13 @@ func (k *Kubernetes) findServices(r recordRequest, zone string) (services []msg.
 	)
 	if wildcard(r.service) || wildcard(r.namespace) {
 		serviceList = k.APIConn.ServiceList()
+		endpointsList = k.APIConn.EndpointsList()
 	} else {
 		idx = r.service + "." + r.namespace
 		serviceList = k.APIConn.SvcIndex(idx)
+		endpointsList = k.APIConn.EpIndex(idx)
 	}
+
 	for _, svc := range serviceList {
 
 		if !(match(r.namespace, svc.Namespace) && match(r.service, svc.Name)) {
@@ -356,12 +359,6 @@ func (k *Kubernetes) findServices(r recordRequest, zone string) (services []msg.
 
 		// Endpoint query or headless service
 		if svc.Spec.ClusterIP == api.ClusterIPNone || r.endpoint != "" {
-			if wildcard(r.service) || wildcard(r.namespace) {
-				endpointsList = k.APIConn.EndpointsList()
-			} else {
-				idx = r.service + "." + r.namespace
-				endpointsList = k.APIConn.EpIndex(idx)
-			}
 			for _, ep := range endpointsList {
 				if ep.ObjectMeta.Name != svc.Name || ep.ObjectMeta.Namespace != svc.Namespace {
 					continue
diff --git a/plugin/kubernetes/kubernetes_test.go b/plugin/kubernetes/kubernetes_test.go
index fecc7c396..ae0bb13e5 100644
--- a/plugin/kubernetes/kubernetes_test.go
+++ b/plugin/kubernetes/kubernetes_test.go
@@ -51,6 +51,7 @@ func TestEndpointHostname(t *testing.T) {
 
 type APIConnServiceTest struct{}
 
+func (APIConnServiceTest) HasSynced() bool                        { return true }
 func (APIConnServiceTest) Run()                                   { return }
 func (APIConnServiceTest) Stop() error                            { return nil }
 func (APIConnServiceTest) PodIndex(string) []*api.Pod             { return nil }
diff --git a/plugin/kubernetes/ns_test.go b/plugin/kubernetes/ns_test.go
index f179231a9..70e297054 100644
--- a/plugin/kubernetes/ns_test.go
+++ b/plugin/kubernetes/ns_test.go
@@ -9,6 +9,7 @@ import (
 
 type APIConnTest struct{}
 
+func (APIConnTest) HasSynced() bool                       { return true }
 func (APIConnTest) Run()                                  { return }
 func (APIConnTest) Stop() error                           { return nil }
 func (APIConnTest) PodIndex(string) []*api.Pod            { return nil }
diff --git a/plugin/kubernetes/reverse_test.go b/plugin/kubernetes/reverse_test.go
index f9102f244..7a5435181 100644
--- a/plugin/kubernetes/reverse_test.go
+++ b/plugin/kubernetes/reverse_test.go
@@ -14,6 +14,7 @@ import (
 
 type APIConnReverseTest struct{}
 
+func (APIConnReverseTest) HasSynced() bool                       { return true }
 func (APIConnReverseTest) Run()                                  { return }
 func (APIConnReverseTest) Stop() error                           { return nil }
 func (APIConnReverseTest) PodIndex(string) []*api.Pod            { return nil }
diff --git a/plugin/kubernetes/setup.go b/plugin/kubernetes/setup.go
index cc823d54e..13e6c810f 100644
--- a/plugin/kubernetes/setup.go
+++ b/plugin/kubernetes/setup.go
@@ -39,8 +39,14 @@ func setup(c *caddy.Controller) error {
 	c.OnStartup(func() error {
 		go kubernetes.APIConn.Run()
 		if kubernetes.APIProxy != nil {
-			go kubernetes.APIProxy.Run()
+			kubernetes.APIProxy.Run()
 		}
+		synced := false
+		for synced == false {
+			synced = kubernetes.APIConn.HasSynced()
+			time.Sleep(100 * time.Millisecond)
+		}
+
 		return nil
 	})