github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-09-09 21:21:14 +00:00
Code Issues Projects Releases Wiki Activity

Reset the resourceVersion so that we poll again for non-timeout errors.

Browse Source
This commit is contained in:
Brendan Burns
2015-01-06 11:36:03 -08:00
parent 4432ba06bd
commit 0f60d7bca3
3 changed files with 77 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/proxy/config/api.go
View File

@@ -20,6 +20,7 @@ import (
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util/wait"
@@ -93,6 +94,10 @@ func (s *SourceAPI) runServices(resourceVersion *string) {
watcher, err := s.servicesWatcher.Watch(labels.Everything(), labels.Everything(), *resourceVersion)
if err != nil {
glog.Errorf("Unable to watch for services changes: %v", err)
if !client.IsTimeout(err) {
// Reset so that we do a fresh get request
*resourceVersion = ""
}
time.Sleep(wait.Jitter(s.waitDuration, 0.0))
return
}
@@ -157,6 +162,11 @@ func (s *SourceAPI) runEndpoints(resourceVersion *string) {
watcher, err := s.endpointsWatcher.Watch(labels.Everything(), labels.Everything(), *resourceVersion)
if err != nil {
glog.Errorf("Unable to watch for endpoints changes: %v", err)
if !client.IsTimeout(err) {
// Reset so that we do a fresh get request
*resourceVersion = ""
}
time.Sleep(wait.Jitter(s.waitDuration, 0.0))
return
}

42
pkg/proxy/config/api_test.go
View File

@@ -120,6 +120,27 @@ func TestServicesError(t *testing.T) {
close(ch)
}()
// should have listed only
<-ch
if resourceVersion != "" {
t.Errorf("unexpected resource version, got %#v", resourceVersion)
}
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-services", "1"}}) {
t.Errorf("unexpected actions, got %#v", fakeClient)
}
}
func TestServicesErrorTimeout(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("use of closed network connection")}
services := make(chan ServiceUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), services: services}
resourceVersion := "1"
ch := make(chan struct{})
go func() {
source.runServices(&resourceVersion)
close(ch)
}()
// should have listed only
<-ch
if resourceVersion != "1" {
@@ -245,6 +266,27 @@ func TestEndpointsError(t *testing.T) {
close(ch)
}()
// should have listed only
<-ch
if resourceVersion != "" {
t.Errorf("unexpected resource version, got %#v", resourceVersion)
}
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-endpoints", "1"}}) {
t.Errorf("unexpected actions, got %#v", fakeClient)
}
}
func TestEndpointsErrorTimeout(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("use of closed network connection")}
endpoints := make(chan EndpointsUpdate)
source := SourceAPI{servicesWatcher: fakeClient.Services(api.NamespaceAll), endpointsWatcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints}
resourceVersion := "1"
ch := make(chan struct{})
go func() {
source.runEndpoints(&resourceVersion)
close(ch)
}()
// should have listed only
<-ch
if resourceVersion != "1" {