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

Services and Endpoints weren't properly sync'ing

Browse Source 
They need incremental changes and a resync on start.
This commit is contained in:
Clayton Coleman
2014-08-28 22:31:41 -04:00
parent 6c5568261e
commit 01e668187c
14 changed files with 326 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,6 +29,8 @@ import (
// Watcher is the interface needed to receive changes to services and endpoints.
type Watcher interface {
ListServices(label labels.Selector) (api.ServiceList, error)
ListEndpoints(label labels.Selector) (api.EndpointsList, error)
WatchServices(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error)
WatchEndpoints(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error)
}
@@ -70,6 +72,17 @@ func NewSourceAPI(client Watcher, period time.Duration, services chan<- ServiceU
// runServices loops forever looking for changes to services.
func (s *SourceAPI) runServices(resourceVersion *uint64) {
if *resourceVersion == 0 {
services, err := s.client.ListServices(labels.Everything())
if err != nil {
glog.Errorf("Unable to load services: %v", err)
time.Sleep(wait.Jitter(s.waitDuration, 0.0))
return
}
*resourceVersion = services.ResourceVersion
s.services <- ServiceUpdate{Op: SET, Services: services.Items}
}
watcher, err := s.client.WatchServices(labels.Everything(), labels.Everything(), *resourceVersion)
if err != nil {
glog.Errorf("Unable to watch for services changes: %v", err)
@@ -97,10 +110,10 @@ func handleServicesWatch(resourceVersion *uint64, ch <-chan watch.Event, updates
switch event.Type {
case watch.Added, watch.Modified:
updates <- ServiceUpdate{Op: SET, Services: []api.Service{*service}}
updates <- ServiceUpdate{Op: ADD, Services: []api.Service{*service}}
case watch.Deleted:
updates <- ServiceUpdate{Op: SET}
updates <- ServiceUpdate{Op: REMOVE, Services: []api.Service{*service}}
}
}
}
@@ -108,6 +121,17 @@ func handleServicesWatch(resourceVersion *uint64, ch <-chan watch.Event, updates
// runEndpoints loops forever looking for changes to endpoints.
func (s *SourceAPI) runEndpoints(resourceVersion *uint64) {
if *resourceVersion == 0 {
endpoints, err := s.client.ListEndpoints(labels.Everything())
if err != nil {
glog.Errorf("Unable to load endpoints: %v", err)
time.Sleep(wait.Jitter(s.waitDuration, 0.0))
return
}
*resourceVersion = endpoints.ResourceVersion
s.endpoints <- EndpointsUpdate{Op: SET, Endpoints: endpoints.Items}
}
watcher, err := s.client.WatchEndpoints(labels.Everything(), labels.Everything(), *resourceVersion)
if err != nil {
glog.Errorf("Unable to watch for endpoints changes: %v", err)
@@ -135,10 +159,10 @@ func handleEndpointsWatch(resourceVersion *uint64, ch <-chan watch.Event, update
switch event.Type {
case watch.Added, watch.Modified:
updates <- EndpointsUpdate{Op: SET, Endpoints: []api.Endpoints{*endpoints}}
updates <- EndpointsUpdate{Op: ADD, Endpoints: []api.Endpoints{*endpoints}}
case watch.Deleted:
updates <- EndpointsUpdate{Op: SET}
updates <- EndpointsUpdate{Op: REMOVE, Endpoints: []api.Endpoints{*endpoints}}
}
}
}

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

@@ -17,6 +17,7 @@ limitations under the License.
package config
import (
"errors"
"reflect"
"testing"
@@ -32,7 +33,7 @@ func TestServices(t *testing.T) {
fakeClient := &client.Fake{Watch: fakeWatch}
services := make(chan ServiceUpdate)
source := SourceAPI{client: fakeClient, services: services}
resourceVersion := uint64(0)
resourceVersion := uint64(1)
go func() {
// called twice
source.runServices(&resourceVersion)
@@ -41,12 +42,12 @@ func TestServices(t *testing.T) {
// test adding a service to the watch
fakeWatch.Add(&service)
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-services", uint64(0)}}) {
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-services", uint64(1)}}) {
t.Errorf("expected call to watch-services, got %#v", fakeClient)
}
actual := <-services
expected := ServiceUpdate{Op: SET, Services: []api.Service{service}}
expected := ServiceUpdate{Op: ADD, Services: []api.Service{service}}
if !reflect.DeepEqual(expected, actual) {
t.Errorf("expected %#v, got %#v", expected, actual)
}
@@ -54,7 +55,7 @@ func TestServices(t *testing.T) {
// verify that a delete results in a config change
fakeWatch.Delete(&service)
actual = <-services
expected = ServiceUpdate{Op: SET}
expected = ServiceUpdate{Op: REMOVE, Services: []api.Service{service}}
if !reflect.DeepEqual(expected, actual) {
t.Errorf("expected %#v, got %#v", expected, actual)
}
@@ -65,11 +66,91 @@ func TestServices(t *testing.T) {
fakeWatch.Stop()
newFakeWatch.Add(&service)
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-services", uint64(0)}, {"watch-services", uint64(3)}}) {
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-services", uint64(1)}, {"watch-services", uint64(3)}}) {
t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
}
}
func TestServicesFromZero(t *testing.T) {
service := api.Service{JSONBase: api.JSONBase{ID: "bar", ResourceVersion: uint64(2)}}
fakeWatch := watch.NewFake()
fakeWatch.Stop()
fakeClient := &client.Fake{Watch: fakeWatch}
fakeClient.ServiceList = api.ServiceList{
JSONBase: api.JSONBase{ResourceVersion: 2},
Items: []api.Service{
service,
},
}
services := make(chan ServiceUpdate)
source := SourceAPI{client: fakeClient, services: services}
resourceVersion := uint64(0)
ch := make(chan struct{})
go func() {
source.runServices(&resourceVersion)
close(ch)
}()
// should get services SET
actual := <-services
expected := ServiceUpdate{Op: SET, Services: []api.Service{service}}
if !reflect.DeepEqual(expected, actual) {
t.Errorf("expected %#v, got %#v", expected, actual)
}
// should have listed, then watched
<-ch
if resourceVersion != 2 {
t.Errorf("unexpected resource version, got %#v", resourceVersion)
}
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"list-services", nil}, {"watch-services", uint64(2)}}) {
t.Errorf("unexpected actions, got %#v", fakeClient)
}
}
func TestServicesError(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("test")}
services := make(chan ServiceUpdate)
source := SourceAPI{client: fakeClient, services: services}
resourceVersion := uint64(1)
ch := make(chan struct{})
go func() {
source.runServices(&resourceVersion)
close(ch)
}()
// should have listed only
<-ch
if resourceVersion != 1 {
t.Errorf("unexpected resource version, got %#v", resourceVersion)
}
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-services", uint64(1)}}) {
t.Errorf("unexpected actions, got %#v", fakeClient)
}
}
func TestServicesFromZeroError(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("test")}
services := make(chan ServiceUpdate)
source := SourceAPI{client: fakeClient, services: services}
resourceVersion := uint64(0)
ch := make(chan struct{})
go func() {
source.runServices(&resourceVersion)
close(ch)
}()
// should have listed only
<-ch
if resourceVersion != 0 {
t.Errorf("unexpected resource version, got %#v", resourceVersion)
}
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"list-services", nil}}) {
t.Errorf("unexpected actions, got %#v", fakeClient)
}
}
func TestEndpoints(t *testing.T) {
endpoint := api.Endpoints{JSONBase: api.JSONBase{ID: "bar", ResourceVersion: uint64(2)}, Endpoints: []string{"127.0.0.1:9000"}}
@@ -77,7 +158,7 @@ func TestEndpoints(t *testing.T) {
fakeClient := &client.Fake{Watch: fakeWatch}
endpoints := make(chan EndpointsUpdate)
source := SourceAPI{client: fakeClient, endpoints: endpoints}
resourceVersion := uint64(0)
resourceVersion := uint64(1)
go func() {
// called twice
source.runEndpoints(&resourceVersion)
@@ -86,12 +167,12 @@ func TestEndpoints(t *testing.T) {
// test adding an endpoint to the watch
fakeWatch.Add(&endpoint)
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-endpoints", uint64(0)}}) {
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-endpoints", uint64(1)}}) {
t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
}
actual := <-endpoints
expected := EndpointsUpdate{Op: SET, Endpoints: []api.Endpoints{endpoint}}
expected := EndpointsUpdate{Op: ADD, Endpoints: []api.Endpoints{endpoint}}
if !reflect.DeepEqual(expected, actual) {
t.Errorf("expected %#v, got %#v", expected, actual)
}
@@ -99,7 +180,7 @@ func TestEndpoints(t *testing.T) {
// verify that a delete results in a config change
fakeWatch.Delete(&endpoint)
actual = <-endpoints
expected = EndpointsUpdate{Op: SET}
expected = EndpointsUpdate{Op: REMOVE, Endpoints: []api.Endpoints{endpoint}}
if !reflect.DeepEqual(expected, actual) {
t.Errorf("expected %#v, got %#v", expected, actual)
}
@@ -110,7 +191,87 @@ func TestEndpoints(t *testing.T) {
fakeWatch.Stop()
newFakeWatch.Add(&endpoint)
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-endpoints", uint64(0)}, {"watch-endpoints", uint64(3)}}) {
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-endpoints", uint64(1)}, {"watch-endpoints", uint64(3)}}) {
t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
}
}
func TestEndpointsFromZero(t *testing.T) {
endpoint := api.Endpoints{JSONBase: api.JSONBase{ID: "bar", ResourceVersion: uint64(2)}, Endpoints: []string{"127.0.0.1:9000"}}
fakeWatch := watch.NewFake()
fakeWatch.Stop()
fakeClient := &client.Fake{Watch: fakeWatch}
fakeClient.EndpointsList = api.EndpointsList{
JSONBase: api.JSONBase{ResourceVersion: 2},
Items: []api.Endpoints{
endpoint,
},
}
endpoints := make(chan EndpointsUpdate)
source := SourceAPI{client: fakeClient, endpoints: endpoints}
resourceVersion := uint64(0)
ch := make(chan struct{})
go func() {
source.runEndpoints(&resourceVersion)
close(ch)
}()
// should get endpoints SET
actual := <-endpoints
expected := EndpointsUpdate{Op: SET, Endpoints: []api.Endpoints{endpoint}}
if !reflect.DeepEqual(expected, actual) {
t.Errorf("expected %#v, got %#v", expected, actual)
}
// should have listed, then watched
<-ch
if resourceVersion != 2 {
t.Errorf("unexpected resource version, got %#v", resourceVersion)
}
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"list-endpoints", nil}, {"watch-endpoints", uint64(2)}}) {
t.Errorf("unexpected actions, got %#v", fakeClient)
}
}
func TestEndpointsError(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("test")}
endpoints := make(chan EndpointsUpdate)
source := SourceAPI{client: fakeClient, endpoints: endpoints}
resourceVersion := uint64(1)
ch := make(chan struct{})
go func() {
source.runEndpoints(&resourceVersion)
close(ch)
}()
// should have listed only
<-ch
if resourceVersion != 1 {
t.Errorf("unexpected resource version, got %#v", resourceVersion)
}
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-endpoints", uint64(1)}}) {
t.Errorf("unexpected actions, got %#v", fakeClient)
}
}
func TestEndpointsFromZeroError(t *testing.T) {
fakeClient := &client.Fake{Err: errors.New("test")}
endpoints := make(chan EndpointsUpdate)
source := SourceAPI{client: fakeClient, endpoints: endpoints}
resourceVersion := uint64(0)
ch := make(chan struct{})
go func() {
source.runEndpoints(&resourceVersion)
close(ch)
}()
// should have listed only
<-ch
if resourceVersion != 0 {
t.Errorf("unexpected resource version, got %#v", resourceVersion)
}
if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"list-endpoints", nil}}) {
t.Errorf("unexpected actions, got %#v", fakeClient)
}
}