Skip to content

Commit

Permalink
Fix issue with long-running apiserver endpoints watch
Browse files Browse the repository at this point in the history
Use ListWatch helpers to retry when the watch channel is closed.

Signed-off-by: Brad Davidson <brad.davidson@rancher.com>
  • Loading branch information
brandond committed Apr 21, 2022
1 parent 6a8de31 commit 33340cc
Showing 1 changed file with 24 additions and 12 deletions.
36 changes: 24 additions & 12 deletions pkg/etcd/apiaddresses_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,32 +10,45 @@ import (
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/tools/cache"
toolswatch "k8s.io/client-go/tools/watch"
)

func registerEndpointsHandlers(ctx context.Context, etcd *ETCD) error {
func registerEndpointsHandlers(ctx context.Context, etcd *ETCD) {
if etcd.config.DisableAPIServer {
return nil
return
}

endpoints := etcd.config.Runtime.Core.Core().V1().Endpoints()
watch, err := endpoints.Watch(metav1.NamespaceDefault, metav1.ListOptions{
FieldSelector: fields.Set{"metadata.name": "kubernetes"}.String(),
ResourceVersion: "0",
})
if err != nil {
return err
fieldSelector := fields.Set{metav1.ObjectNameField: "kubernetes"}.String()
lw := &cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (object runtime.Object, e error) {
options.FieldSelector = fieldSelector
return endpoints.List(metav1.NamespaceDefault, options)
},
WatchFunc: func(options metav1.ListOptions) (i watch.Interface, e error) {
options.FieldSelector = fieldSelector
return endpoints.Watch(metav1.NamespaceDefault, options)
},
}

_, _, watch, done := toolswatch.NewIndexerInformerWatcher(lw, &v1.Endpoints{})

go func() {
<-ctx.Done()
watch.Stop()
<-done
}()

h := &handler{
etcd: etcd,
watch: watch,
}

logrus.Infof("Starting managed etcd apiserver addresses controller")
go h.watchEndpoints(ctx)

return nil
}

type handler struct {
Expand All @@ -53,8 +66,7 @@ func (h *handler) watchEndpoints(ctx context.Context) {
case ev, ok := <-h.watch.ResultChan():
endpoint, ok := ev.Object.(*v1.Endpoints)
if !ok {
logrus.Errorf("Failed to watch apiserver addresses: could not convert event object to endpoint: %v", ev)
continue
logrus.Fatalf("Failed to watch apiserver addresses: could not convert event object to endpoint: %v", ev)
}

w := &bytes.Buffer{}
Expand Down

0 comments on commit 33340cc

Please sign in to comment.