Skip to content

K
k3s
Commit 29580020 authored by Eric Tune's avatar Eric Tune
Browse files
Options

Make pkg/proxy/config more like pkg/kubelet/config

Split SourceAPI into two subobjects. Parallel structure for endpoints, services will allow changing to use generic code in pkg/client/cache/reflector.go. Rename some funcs to be more like pkg/client/cache.
parent fa152ab3
Hide whitespace changes
Inline Side-by-side
Showing with 101 additions and 58 deletions
pkg/proxy/config/api.go
View file @ 29580020
...@@ -28,6 +28,7 @@ import ( ...@@ -28,6 +28,7 @@ import (
"github.com/golang/glog" "github.com/golang/glog"
) )
// TODO: to use Reflector, need to change the ServicesWatcher to a generic ListerWatcher.
// ServicesWatcher is capable of listing and watching for changes to services across ALL namespaces // ServicesWatcher is capable of listing and watching for changes to services across ALL namespaces
type ServicesWatcher interface { type ServicesWatcher interface {
List(label labels.Selector) (*api.ServiceList, error) List(label labels.Selector) (*api.ServiceList, error)
...@@ -43,12 +44,22 @@ type EndpointsWatcher interface { ...@@ -43,12 +44,22 @@ type EndpointsWatcher interface {
// SourceAPI implements a configuration source for services and endpoints that // SourceAPI implements a configuration source for services and endpoints that
// uses the client watch API to efficiently detect changes. // uses the client watch API to efficiently detect changes.
type SourceAPI struct { type SourceAPI struct {
servicesWatcher ServicesWatcher s servicesReflector
endpointsWatcher EndpointsWatcher e endpointsReflector
}
services chan<- ServiceUpdate type servicesReflector struct {
endpoints chan<- EndpointsUpdate watcher ServicesWatcher
services chan<- ServiceUpdate
resourceVersion string
waitDuration time.Duration
reconnectDuration time.Duration
}
type endpointsReflector struct {
watcher EndpointsWatcher
endpoints chan<- EndpointsUpdate
resourceVersion string
waitDuration time.Duration waitDuration time.Duration
reconnectDuration time.Duration reconnectDuration time.Duration
} }
...@@ -56,42 +67,54 @@ type SourceAPI struct { ...@@ -56,42 +67,54 @@ type SourceAPI struct {
// NewSourceAPI creates a config source that watches for changes to the services and endpoints. // NewSourceAPI creates a config source that watches for changes to the services and endpoints.
func NewSourceAPI(servicesWatcher ServicesWatcher, endpointsWatcher EndpointsWatcher, period time.Duration, services chan<- ServiceUpdate, endpoints chan<- EndpointsUpdate) *SourceAPI { func NewSourceAPI(servicesWatcher ServicesWatcher, endpointsWatcher EndpointsWatcher, period time.Duration, services chan<- ServiceUpdate, endpoints chan<- EndpointsUpdate) *SourceAPI {
config := &SourceAPI{ config := &SourceAPI{
servicesWatcher: servicesWatcher, s: servicesReflector{
endpointsWatcher: endpointsWatcher, watcher: servicesWatcher,
services: services, services: services,
endpoints: endpoints, resourceVersion: "",
waitDuration: period,
waitDuration: period, // prevent hot loops if the server starts to misbehave
// prevent hot loops if the server starts to misbehave reconnectDuration: time.Second * 1,
reconnectDuration: time.Second * 1, },
e: endpointsReflector{
watcher: endpointsWatcher,
endpoints: endpoints,
resourceVersion: "",
waitDuration: period,
// prevent hot loops if the server starts to misbehave
reconnectDuration: time.Second * 1,
},
} }
serviceVersion := "" go util.Forever(func() { config.s.listAndWatch() }, period)
go util.Forever(func() { go util.Forever(func() { config.e.listAndWatch() }, period)
config.runServices(&serviceVersion)
time.Sleep(wait.Jitter(config.reconnectDuration, 0.0))
}, period)
endpointVersion := ""
go util.Forever(func() {
config.runEndpoints(&endpointVersion)
time.Sleep(wait.Jitter(config.reconnectDuration, 0.0))
}, period)
return config return config
} }
// runServices loops forever looking for changes to services. func (r *servicesReflector) listAndWatch() {
func (s *SourceAPI) runServices(resourceVersion *string) { r.run(&r.resourceVersion)
time.Sleep(wait.Jitter(r.reconnectDuration, 0.0))
}
func (r *endpointsReflector) listAndWatch() {
r.run(&r.resourceVersion)
time.Sleep(wait.Jitter(r.reconnectDuration, 0.0))
}
// run loops forever looking for changes to services.
func (s *servicesReflector) run(resourceVersion *string) {
if len(*resourceVersion) == 0 { if len(*resourceVersion) == 0 {
services, err := s.servicesWatcher.List(labels.Everything()) services, err := s.watcher.List(labels.Everything())
if err != nil { if err != nil {
glog.Errorf("Unable to load services: %v", err) glog.Errorf("Unable to load services: %v", err)
// TODO: reconcile with pkg/client/cache which doesn't use reflector.
time.Sleep(wait.Jitter(s.waitDuration, 0.0)) time.Sleep(wait.Jitter(s.waitDuration, 0.0))
return return
} }
*resourceVersion = services.ResourceVersion *resourceVersion = services.ResourceVersion
// TODO: replace with code to update the
s.services <- ServiceUpdate{Op: SET, Services: services.Items} s.services <- ServiceUpdate{Op: SET, Services: services.Items}
} }
watcher, err := s.servicesWatcher.Watch(labels.Everything(), labels.Everything(), *resourceVersion) watcher, err := s.watcher.Watch(labels.Everything(), labels.Everything(), *resourceVersion)
if err != nil { if err != nil {
glog.Errorf("Unable to watch for services changes: %v", err) glog.Errorf("Unable to watch for services changes: %v", err)
if !client.IsTimeout(err) { if !client.IsTimeout(err) {
...@@ -104,11 +127,11 @@ func (s *SourceAPI) runServices(resourceVersion *string) { ...@@ -104,11 +127,11 @@ func (s *SourceAPI) runServices(resourceVersion *string) {
defer watcher.Stop() defer watcher.Stop()
ch := watcher.ResultChan() ch := watcher.ResultChan()
handleServicesWatch(resourceVersion, ch, s.services) s.watchHandler(resourceVersion, ch, s.services)
} }
// handleServicesWatch loops over an event channel and delivers config changes to an update channel. // watchHandler loops over an event channel and delivers config changes to an update channel.
func handleServicesWatch(resourceVersion *string, ch <-chan watch.Event, updates chan<- ServiceUpdate) { func (s *servicesReflector) watchHandler(resourceVersion *string, ch <-chan watch.Event, updates chan<- ServiceUpdate) {
for { for {
select { select {
case event, ok := <-ch: case event, ok := <-ch:
...@@ -146,10 +169,10 @@ func handleServicesWatch(resourceVersion *string, ch <-chan watch.Event, updates ...@@ -146,10 +169,10 @@ func handleServicesWatch(resourceVersion *string, ch <-chan watch.Event, updates
} }
} }
// runEndpoints loops forever looking for changes to endpoints. // run loops forever looking for changes to endpoints.
func (s *SourceAPI) runEndpoints(resourceVersion *string) { func (s *endpointsReflector) run(resourceVersion *string) {
if len(*resourceVersion) == 0 { if len(*resourceVersion) == 0 {
endpoints, err := s.endpointsWatcher.List(labels.Everything()) endpoints, err := s.watcher.List(labels.Everything())
if err != nil { if err != nil {
glog.Errorf("Unable to load endpoints: %v", err) glog.Errorf("Unable to load endpoints: %v", err)
time.Sleep(wait.Jitter(s.waitDuration, 0.0)) time.Sleep(wait.Jitter(s.waitDuration, 0.0))
...@@ -159,7 +182,7 @@ func (s *SourceAPI) runEndpoints(resourceVersion *string) { ...@@ -159,7 +182,7 @@ func (s *SourceAPI) runEndpoints(resourceVersion *string) {
s.endpoints <- EndpointsUpdate{Op: SET, Endpoints: endpoints.Items} s.endpoints <- EndpointsUpdate{Op: SET, Endpoints: endpoints.Items}
} }
watcher, err := s.endpointsWatcher.Watch(labels.Everything(), labels.Everything(), *resourceVersion) watcher, err := s.watcher.Watch(labels.Everything(), labels.Everything(), *resourceVersion)
if err != nil { if err != nil {
glog.Errorf("Unable to watch for endpoints changes: %v", err) glog.Errorf("Unable to watch for endpoints changes: %v", err)
if !client.IsTimeout(err) { if !client.IsTimeout(err) {
...@@ -173,11 +196,11 @@ func (s *SourceAPI) runEndpoints(resourceVersion *string) { ...@@ -173,11 +196,11 @@ func (s *SourceAPI) runEndpoints(resourceVersion *string) {
defer watcher.Stop() defer watcher.Stop()
ch := watcher.ResultChan() ch := watcher.ResultChan()
handleEndpointsWatch(resourceVersion, ch, s.endpoints) s.watchHandler(resourceVersion, ch, s.endpoints)
} }
// handleEndpointsWatch loops over an event channel and delivers config changes to an update channel. // watchHandler loops over an event channel and delivers config changes to an update channel.
func handleEndpointsWatch(resourceVersion *string, ch <-chan watch.Event, updates chan<- EndpointsUpdate) { func (s *endpointsReflector) watchHandler(resourceVersion *string, ch <-chan watch.Event, updates chan<- EndpointsUpdate) {
for { for {
select { select {
case event, ok := <-ch: case event, ok := <-ch:
......
pkg/proxy/config/api_test.go
View file @ 29580020
...@@ -32,12 +32,14 @@ func TestServices(t *testing.T) { ...@@ -32,12 +32,14 @@ func TestServices(t *testing.T) {
fakeWatch := watch.NewFake() fakeWatch := watch.NewFake()
fakeClient := &client.Fake{Watch: fakeWatch} fakeClient := &client.Fake{Watch: fakeWatch}
services := make(chan ServiceUpdate) services := make(chan ServiceUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), services: services} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll), services: services},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll)}}
resourceVersion := "1" resourceVersion := "1"
go func() { go func() {
// called twice // called twice
source.runServices(&resourceVersion) source.s.run(&resourceVersion)
source.runServices(&resourceVersion) source.s.run(&resourceVersion)
}() }()
// test adding a service to the watch // test adding a service to the watch
...@@ -84,11 +86,13 @@ func TestServicesFromZero(t *testing.T) { ...@@ -84,11 +86,13 @@ func TestServicesFromZero(t *testing.T) {
}, },
} }
services := make(chan ServiceUpdate) services := make(chan ServiceUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), services: services} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll), services: services},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll)}}
resourceVersion := "" resourceVersion := ""
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
source.runServices(&resourceVersion) source.s.run(&resourceVersion)
close(ch) close(ch)
}() }()
...@@ -112,11 +116,13 @@ func TestServicesFromZero(t *testing.T) { ...@@ -112,11 +116,13 @@ func TestServicesFromZero(t *testing.T) {
func TestServicesError(t *testing.T) { func TestServicesError(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("test")} fakeClient := &client.Fake{Err: errors.New("test")}
services := make(chan ServiceUpdate) services := make(chan ServiceUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), services: services} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll), services: services},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll)}}
resourceVersion := "1" resourceVersion := "1"
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
source.runServices(&resourceVersion) source.s.run(&resourceVersion)
close(ch) close(ch)
}() }()
...@@ -133,11 +139,13 @@ func TestServicesError(t *testing.T) { ...@@ -133,11 +139,13 @@ func TestServicesError(t *testing.T) {
func TestServicesErrorTimeout(t *testing.T) { func TestServicesErrorTimeout(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("use of closed network connection")} fakeClient := &client.Fake{Err: errors.New("use of closed network connection")}
services := make(chan ServiceUpdate) services := make(chan ServiceUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), services: services} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll), services: services},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll)}}
resourceVersion := "1" resourceVersion := "1"
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
source.runServices(&resourceVersion) source.s.run(&resourceVersion)
close(ch) close(ch)
}() }()
...@@ -154,11 +162,13 @@ func TestServicesErrorTimeout(t *testing.T) { ...@@ -154,11 +162,13 @@ func TestServicesErrorTimeout(t *testing.T) {
func TestServicesFromZeroError(t *testing.T) { func TestServicesFromZeroError(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("test")} fakeClient := &client.Fake{Err: errors.New("test")}
services := make(chan ServiceUpdate) services := make(chan ServiceUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), services: services} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll), services: services},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll)}}
resourceVersion := "" resourceVersion := ""
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
source.runServices(&resourceVersion) source.s.run(&resourceVersion)
close(ch) close(ch)
}() }()
...@@ -178,12 +188,14 @@ func TestEndpoints(t *testing.T) { ...@@ -178,12 +188,14 @@ func TestEndpoints(t *testing.T) {
fakeWatch := watch.NewFake() fakeWatch := watch.NewFake()
fakeClient := &client.Fake{Watch: fakeWatch} fakeClient := &client.Fake{Watch: fakeWatch}
endpoints := make(chan EndpointsUpdate) endpoints := make(chan EndpointsUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll)},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints}}
resourceVersion := "1" resourceVersion := "1"
go func() { go func() {
// called twice // called twice
source.runEndpoints(&resourceVersion) source.e.run(&resourceVersion)
source.runEndpoints(&resourceVersion) source.e.run(&resourceVersion)
}() }()
// test adding an endpoint to the watch // test adding an endpoint to the watch
...@@ -230,11 +242,13 @@ func TestEndpointsFromZero(t *testing.T) { ...@@ -230,11 +242,13 @@ func TestEndpointsFromZero(t *testing.T) {
}, },
} }
endpoints := make(chan EndpointsUpdate) endpoints := make(chan EndpointsUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll)},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints}}
resourceVersion := "" resourceVersion := ""
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
source.runEndpoints(&resourceVersion) source.e.run(&resourceVersion)
close(ch) close(ch)
}() }()
...@@ -258,11 +272,13 @@ func TestEndpointsFromZero(t *testing.T) { ...@@ -258,11 +272,13 @@ func TestEndpointsFromZero(t *testing.T) {
func TestEndpointsError(t *testing.T) { func TestEndpointsError(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("test")} fakeClient := &client.Fake{Err: errors.New("test")}
endpoints := make(chan EndpointsUpdate) endpoints := make(chan EndpointsUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll)},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints}}
resourceVersion := "1" resourceVersion := "1"
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
source.runEndpoints(&resourceVersion) source.e.run(&resourceVersion)
close(ch) close(ch)
}() }()
...@@ -279,11 +295,13 @@ func TestEndpointsError(t *testing.T) { ...@@ -279,11 +295,13 @@ func TestEndpointsError(t *testing.T) {
func TestEndpointsErrorTimeout(t *testing.T) { func TestEndpointsErrorTimeout(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("use of closed network connection")} fakeClient := &client.Fake{Err: errors.New("use of closed network connection")}
endpoints := make(chan EndpointsUpdate) endpoints := make(chan EndpointsUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll)},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints}}
resourceVersion := "1" resourceVersion := "1"
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
source.runEndpoints(&resourceVersion) source.e.run(&resourceVersion)
close(ch) close(ch)
}() }()
...@@ -300,11 +318,13 @@ func TestEndpointsErrorTimeout(t *testing.T) { ...@@ -300,11 +318,13 @@ func TestEndpointsErrorTimeout(t *testing.T) {
func TestEndpointsFromZeroError(t *testing.T) { func TestEndpointsFromZeroError(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("test")} fakeClient := &client.Fake{Err: errors.New("test")}
endpoints := make(chan EndpointsUpdate) endpoints := make(chan EndpointsUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints} source := SourceAPI{
s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll)},
e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints}}
resourceVersion := "" resourceVersion := ""
ch := make(chan struct{}) ch := make(chan struct{})
go func() { go func() {
source.runEndpoints(&resourceVersion) source.e.run(&resourceVersion)
close(ch) close(ch)
}() }()
......
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