k3s/pkg/etcd/apiaddresses_controller.go
Brad Davidson f2ceeb01d9
Fix issue with long-running apiserver endpoints watch (#5478)
Use ListWatch helpers to retry when the watch channel is closed.

Signed-off-by: Brad Davidson <brad.davidson@rancher.com>
2022-04-21 09:24:34 -07:00

85 lines
2.2 KiB
Go

package etcd
import (
"bytes"
"context"
"encoding/json"
"github.com/k3s-io/k3s/pkg/util"
"github.com/sirupsen/logrus"
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) {
if etcd.config.DisableAPIServer {
return
}
endpoints := etcd.config.Runtime.Core.Core().V1().Endpoints()
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)
}
type handler struct {
etcd *ETCD
watch watch.Interface
}
// This controller will update the version.program/apiaddresses etcd key with a list of
// api addresses endpoints found in the kubernetes service in the default namespace
func (h *handler) watchEndpoints(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case ev, ok := <-h.watch.ResultChan():
endpoint, ok := ev.Object.(*v1.Endpoints)
if !ok {
logrus.Fatalf("Failed to watch apiserver addresses: could not convert event object to endpoint: %v", ev)
}
w := &bytes.Buffer{}
if err := json.NewEncoder(w).Encode(util.GetAddresses(endpoint)); err != nil {
logrus.Warnf("Failed to encode apiserver addresses: %v", err)
continue
}
_, err := h.etcd.client.Put(ctx, AddressKey, w.String())
if err != nil {
logrus.Warnf("Failed to store apiserver addresses in etcd: %v", err)
}
}
}
}