mirror of
https://github.com/k3s-io/k3s.git
synced 2024-06-07 19:41:36 +00:00
f37e7565b8
This controller only needs to run when using managed etcd, so move it in with the rest of the etcd stuff. This change also modifies the controller to only watch the Kubernetes service endpoint, instead of watching all endpoints in the entire cluster. Fixes an error message revealed by use of a newer grpc client in Kubernetes 1.24, which logs an error when the Put to etcd failed because kine doesn't support the etcd Put operation. The controller shouldn't have been running without etcd in the first place. Signed-off-by: Brad Davidson <brad.davidson@rancher.com>
73 lines
1.7 KiB
Go
73 lines
1.7 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/watch"
|
|
)
|
|
|
|
func registerEndpointsHandlers(ctx context.Context, etcd *ETCD) error {
|
|
if etcd.config.DisableAPIServer {
|
|
return nil
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
h := &handler{
|
|
etcd: etcd,
|
|
watch: watch,
|
|
}
|
|
|
|
logrus.Infof("Starting managed etcd apiserver addresses controller")
|
|
go h.watchEndpoints(ctx)
|
|
|
|
return nil
|
|
}
|
|
|
|
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.Errorf("Failed to watch apiserver addresses: could not convert event object to endpoint: %v", ev)
|
|
continue
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
}
|
|
}
|