Commit 4318dfbf authored by John Belamaric's avatar John Belamaric Committed by Miek Gieben

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.
parent 352901d8
...@@ -2,6 +2,7 @@ package kubernetes ...@@ -2,6 +2,7 @@ package kubernetes
import ( import (
"fmt" "fmt"
"log"
"sync" "sync"
"time" "time"
...@@ -96,12 +97,44 @@ func serviceListFunc(c *kubernetes.Clientset, ns string, s *labels.Selector) fun ...@@ -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) { 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) { return func(options api.ListOptions) (watch.Interface, error) {
if s != nil { if s != nil {
options.LabelSelector = *s 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 ...@@ -128,7 +161,11 @@ func namespaceWatchFunc(c *kubernetes.Clientset, s *labels.Selector) func(option
if s != nil { if s != nil {
options.LabelSelector = *s 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
} }
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment