Merge pull request #1861 from derekwaynecarr/fix_watch_endpoints

Error when watching service endpoints across all namespaces

pkg/registry/etcd/etcd.go

@@ -575,11 +575,7 @@ func (r *Registry) WatchEndpoints(ctx api.Context, label, field labels.Selector,
 		return r.Watch(key, version), nil
 	}
 	if field.Empty() {
-		key, err := makeServiceEndpointsKey(ctx, "")
+		return r.WatchList(makeServiceEndpointsListKey(ctx), version, tools.Everything)
-		if err != nil {
-			return nil, err
-		}
-		return r.WatchList(key, version, tools.Everything)
 	}
 	return nil, fmt.Errorf("only the 'ID' and default (everything) field selectors are supported")
 }

pkg/registry/etcd/etcd_test.go

@@ -1327,6 +1327,35 @@ func TestEtcdWatchEndpoints(t *testing.T) {
 	watching.Stop()
 }
 
+func TestEtcdWatchEndpointsAcrossNamespaces(t *testing.T) {
+	ctx := api.NewContext()
+	fakeClient := tools.NewFakeEtcdClient(t)
+	registry := NewTestEtcdRegistry(fakeClient)
+	watching, err := registry.WatchEndpoints(
+		ctx,
+		labels.Everything(),
+		labels.Everything(),
+		"1",
+	)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	fakeClient.WaitForWatchCompletion()
+
+	select {
+	case _, ok := <-watching.ResultChan():
+		if !ok {
+			t.Errorf("watching channel should be open")
+		}
+	default:
+	}
+	fakeClient.WatchInjectError <- nil
+	if _, ok := <-watching.ResultChan(); ok {
+		t.Errorf("watching channel should be closed")
+	}
+	watching.Stop()
+}
+
 func TestEtcdWatchEndpointsBadSelector(t *testing.T) {
 	ctx := api.NewContext()
 	fakeClient := tools.NewFakeEtcdClient(t)