diff --git a/manifests/rolebindings.yaml b/manifests/rolebindings.yaml index 29c65465cd..d04f24d63a 100644 --- a/manifests/rolebindings.yaml +++ b/manifests/rolebindings.yaml @@ -55,6 +55,14 @@ rules: - list - get - watch +- apiGroups: + - "discovery.k8s.io" + resources: + - endpointslices + verbs: + - list + - get + - watch --- apiVersion: rbac.authorization.k8s.io/v1 diff --git a/pkg/agent/tunnel/tunnel.go b/pkg/agent/tunnel/tunnel.go index 8c5a51e410..e5e4658641 100644 --- a/pkg/agent/tunnel/tunnel.go +++ b/pkg/agent/tunnel/tunnel.go @@ -24,8 +24,10 @@ import ( "github.com/yl2chen/cidranger" authorizationv1 "k8s.io/api/authorization/v1" v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/fields" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/wait" @@ -70,7 +72,7 @@ func (p *podEntry) Network() net.IPNet { } // Setup sets up the agent tunnel, which is reponsible for connecting websocket tunnels to -// control-plane nodes, syncing endpoints for the tunnel authorizer, and updating proxy endpoints. +// control-plane nodes, syncing endpointslices for the tunnel authorizer, and updating proxy endpoints. func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) error { client, err := util.GetClientSet(config.AgentConfig.KubeConfigK3sController) if err != nil { @@ -103,7 +105,7 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er return nil } -// startWatches starts watching for changes to endpoints, both for the tunnel authorizer, +// startWatches starts watching for changes to endpointslices, both for the tunnel authorizer, // and to sync supervisor addresses into the proxy. This will block until the context is cancelled. func (a *agentTunnel) startWatches(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) { rbacReady := make(chan struct{}) @@ -112,7 +114,8 @@ func (a *agentTunnel) startWatches(ctx context.Context, config *daemonconfig.Nod if err := util.WaitForRBACReady(ctx, config.AgentConfig.KubeConfigK3sController, util.DefaultAPIServerReadyTimeout, authorizationv1.ResourceAttributes{ Namespace: metav1.NamespaceDefault, Verb: "list", - Resource: "endpoints", + Group: "discovery.k8s.io", + Resource: "endpointslices", }, ""); err != nil { logrus.Fatalf("Tunnel watches failed to wait for RBAC: %v", err) } @@ -150,7 +153,7 @@ func (a *agentTunnel) startWatches(ctx context.Context, config *daemonconfig.Nod if proxy.IsSupervisorLBEnabled() && proxy.SupervisorURL() != "" { logrus.Info("Getting list of apiserver endpoints from server") // If not running an apiserver locally, try to get a list of apiservers from the server we're - // connecting to. If that fails, fall back to querying the endpoints list from Kubernetes. This + // connecting to. If that fails, fall back to querying the endpointslice list from Kubernetes. This // fallback requires that the server we're joining be running an apiserver, but is the only safe // thing to do if its supervisor is down-level and can't provide us with an endpoint list. addresses := agentconfig.WaitForAPIServers(ctx, config, proxy) @@ -161,11 +164,12 @@ func (a *agentTunnel) startWatches(ctx context.Context, config *daemonconfig.Nod } proxy.Update(addresses) } else { - if endpoint, err := a.client.CoreV1().Endpoints(metav1.NamespaceDefault).Get(ctx, "kubernetes", metav1.GetOptions{}); err != nil { - logrus.Errorf("Failed to get apiserver addresses from kubernetes endpoints: %v", err) + labelSelector := labels.Set{discoveryv1.LabelServiceName: "kubernetes"}.String() + if endpointSlices, err := a.client.DiscoveryV1().EndpointSlices(metav1.NamespaceDefault).List(ctx, metav1.ListOptions{LabelSelector: labelSelector}); err != nil { + logrus.Errorf("Failed to get apiserver addresses from kubernetes endpointslices: %v", err) } else { - addresses := util.GetAddresses(endpoint) - logrus.Infof("Got apiserver addresses from kubernetes endpoints: %v", addresses) + addresses := util.GetAddressesFromSlices(endpointSlices.Items...) + logrus.Infof("Got apiserver addresses from kubernetes endpointslice: %v", addresses) if len(addresses) > 0 { proxy.Update(addresses) } @@ -173,7 +177,7 @@ func (a *agentTunnel) startWatches(ctx context.Context, config *daemonconfig.Nod } } - a.watchEndpoints(ctx, rbacReady, config, proxy) + a.watchEndpointSlices(ctx, rbacReady, config, proxy) } // setKubeletPort retrieves the configured kubelet port from our node object @@ -298,17 +302,17 @@ func (a *agentTunnel) watchPods(ctx context.Context, rbacReady <-chan struct{}, } } -// WatchEndpoints attempts to create tunnels to all supervisor addresses. Once the +// WatchEndpointSlices attempts to create tunnels to all supervisor addresses. Once the // apiserver is up, go into a watch loop, adding and removing tunnels as endpoints come // and go from the cluster. -func (a *agentTunnel) watchEndpoints(ctx context.Context, rbacReady <-chan struct{}, node *daemonconfig.Node, proxy proxy.Proxy) { +func (a *agentTunnel) watchEndpointSlices(ctx context.Context, rbacReady <-chan struct{}, node *daemonconfig.Node, proxy proxy.Proxy) { syncProxyAddresses := a.getProxySyncer(ctx, proxy) refreshFromSupervisor := getAPIServersRequester(node, proxy, syncProxyAddresses) <-rbacReady - endpoints := a.client.CoreV1().Endpoints(metav1.NamespaceDefault) - fieldSelector := fields.Set{metav1.ObjectNameField: "kubernetes"}.String() + endpointslices := a.client.DiscoveryV1().EndpointSlices(metav1.NamespaceDefault) + labelSelector := labels.Set{discoveryv1.LabelServiceName: "kubernetes"}.String() lw := &cache.ListWatch{ ListFunc: func(options metav1.ListOptions) (object runtime.Object, e error) { // if we're being called to re-list, then likely there was an @@ -316,16 +320,16 @@ func (a *agentTunnel) watchEndpoints(ctx context.Context, rbacReady <-chan struc // its connection. This is a good suggestion that it might be necessary // to refresh the apiserver address from the supervisor. go refreshFromSupervisor(ctx) - options.FieldSelector = fieldSelector - return endpoints.List(ctx, options) + options.LabelSelector = labelSelector + return endpointslices.List(ctx, options) }, WatchFunc: func(options metav1.ListOptions) (i watch.Interface, e error) { - options.FieldSelector = fieldSelector - return endpoints.Watch(ctx, options) + options.LabelSelector = labelSelector + return endpointslices.Watch(ctx, options) }, } - _, _, watch, done := toolswatch.NewIndexerInformerWatcher(lw, &v1.Endpoints{}) + _, _, watch, done := toolswatch.NewIndexerInformerWatcher(lw, &discoveryv1.EndpointSlice{}) defer func() { watch.Stop() @@ -337,9 +341,9 @@ func (a *agentTunnel) watchEndpoints(ctx context.Context, rbacReady <-chan struc case <-ctx.Done(): return case ev, ok := <-watch.ResultChan(): - endpoint, ok := ev.Object.(*v1.Endpoints) + endpointslice, ok := ev.Object.(*discoveryv1.EndpointSlice) if !ok { - logrus.Errorf("Tunnel watch failed: event object not of type v1.Endpoints") + logrus.Errorf("Tunnel watch failed: event object not of type discoveryv1.EndpointSlice") continue } @@ -349,7 +353,7 @@ func (a *agentTunnel) watchEndpoints(ctx context.Context, rbacReady <-chan struc // goroutine that sleeps for a short period before checking for changes and updating // the proxy addresses. If another update occurs, the previous update operation // will be cancelled and a new one queued. - addresses := util.GetAddresses(endpoint) + addresses := util.GetAddressesFromSlices(*endpointslice) logrus.Debugf("Syncing apiserver addresses from tunnel watch: %v", addresses) syncProxyAddresses(addresses) } diff --git a/pkg/daemons/config/types.go b/pkg/daemons/config/types.go index af1361e4e3..6b8758cd73 100644 --- a/pkg/daemons/config/types.go +++ b/pkg/daemons/config/types.go @@ -12,6 +12,7 @@ import ( "github.com/k3s-io/kine/pkg/endpoint" "github.com/rancher/wharfie/pkg/registries" "github.com/rancher/wrangler/v3/pkg/generated/controllers/core" + "github.com/rancher/wrangler/v3/pkg/generated/controllers/discovery" "github.com/rancher/wrangler/v3/pkg/leader" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" utilnet "k8s.io/apimachinery/pkg/util/net" @@ -379,8 +380,9 @@ type ControlRuntime struct { ClientETCDKey string K8s kubernetes.Interface - K3s *k3s.Factory + K3s K3sFactory Core CoreFactory + Discovery DiscoveryFactory Event record.EventRecorder EtcdConfig endpoint.ETCDConfig } @@ -391,12 +393,24 @@ type Cluster interface { Start(ctx context.Context) error } +type K3sFactory interface { + K3s() k3s.Interface + Sync(ctx context.Context) error + Start(ctx context.Context, defaultThreadiness int) error +} + type CoreFactory interface { Core() core.Interface Sync(ctx context.Context) error Start(ctx context.Context, defaultThreadiness int) error } +type DiscoveryFactory interface { + Discovery() discovery.Interface + Sync(ctx context.Context) error + Start(ctx context.Context, defaultThreadiness int) error +} + func NewRuntime() *ControlRuntime { return &ControlRuntime{ ClusterControllerStarts: map[string]leader.Callback{}, diff --git a/pkg/deploy/zz_generated_bindata.go b/pkg/deploy/zz_generated_bindata.go index 459f87ef3d..f98c67ccc5 100644 --- a/pkg/deploy/zz_generated_bindata.go +++ b/pkg/deploy/zz_generated_bindata.go @@ -292,7 +292,7 @@ func metricsServerResourceReaderYaml() (*asset, error) { return a, nil } -var _rolebindingsYaml = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xb4\x94\x41\x8f\xda\x30\x10\x85\xef\xfe\x15\x16\x77\x83\xaa\x5e\xaa\x1c\xdb\x43\xef\x48\xed\xdd\xb1\xa7\x30\x8d\x63\x5b\x33\x63\x50\xfb\xeb\xab\x90\x40\x17\x92\xb0\x64\x97\x3d\x25\xb1\xec\xf7\x8d\x67\xde\x8b\xcd\xf8\x13\x88\x31\xc5\x4a\x53\x6d\xdd\xda\x16\xd9\x27\xc2\xbf\x56\x30\xc5\x75\xf3\x85\xd7\x98\x36\x87\x4f\xaa\xc1\xe8\x2b\xfd\x2d\x14\x16\xa0\x6d\x0a\xf0\x15\xa3\xc7\xb8\x53\x2d\x88\xf5\x56\x6c\xa5\xb4\x8e\xb6\x85\x4a\x37\xa5\x06\x63\x33\x32\xd0\x01\xc8\x74\x9f\x01\xc4\x58\xdf\x62\x54\x94\x02\x6c\xe1\x57\xb7\xdb\x66\xfc\x4e\xa9\xe4\x3b\x64\xa5\xf5\x08\x7c\xe1\xf0\x1f\x16\x68\xab\x8b\x7e\xc6\x81\xc1\xa5\xfe\x0d\x4e\xb8\x52\x66\x11\xe4\x07\x03\xcd\xdc\x42\x29\x63\x8c\x7a\x7b\xb7\x26\xda\x74\x2e\xff\x33\x1b\x97\xa2\x50\x0a\x01\x48\x51\x09\x70\x55\x38\x77\x27\x8c\x5e\xad\x94\xd6\x04\x9c\x0a\x39\x18\xd6\x62\xf2\xc0\x4a\xeb\x03\x50\x3d\x2c\xed\x40\x4e\xcf\x80\xdc\xbf\x1c\xad\xb8\xfd\x02\xb9\x0d\x8b\x95\x72\xa3\x9a\x17\x88\xd8\x16\x38\x5b\x77\x5b\xd8\xab\x05\x45\x90\x63\xa2\x06\xe3\x6e\xe8\xe3\x94\x78\xbf\x27\xa7\x80\x0e\x4f\x04\xa3\x5d\xdf\x64\x87\x9e\x96\x22\x27\x08\x10\x7d\x4e\x18\xa5\xd7\xce\xc9\xcf\x69\x9e\x1b\xdd\x6b\xbf\xd3\x1d\xf3\x59\x9a\x31\xc9\xf3\x43\x74\x0d\xf8\x9f\xa0\xee\x8e\x8f\x31\x6e\x52\x74\x1f\xf0\xfc\x38\xbd\xf4\x81\xe9\xac\x3c\x1b\xa5\x91\xd3\xc6\x36\x78\xd8\x54\x1f\x36\xf8\x89\xeb\x3c\x6f\xe8\x63\xf1\xeb\x81\xf7\x27\x4f\x88\xf1\x24\xcf\x7f\x9d\xc7\xca\xf8\x17\x00\x00\xff\xff\x40\xa6\x57\x0f\x61\x06\x00\x00") +var _rolebindingsYaml = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xb4\x94\x4f\x8f\xd3\x30\x10\xc5\xef\xfe\x14\xd6\xde\xdd\x15\xe2\x82\x72\x84\x03\xf7\x95\xe0\xee\xd8\x43\x77\x88\xe3\xb1\x66\xec\xac\x96\x4f\x8f\xd2\xa4\x85\x34\xf1\xd2\x40\x39\xb5\xb5\xec\xf7\x9b\x3f\xef\xd5\x26\xfc\x0a\x2c\x48\xb1\xd1\xdc\x5a\x77\xb0\x25\x3f\x13\xe3\x0f\x9b\x91\xe2\xa1\xfb\x20\x07\xa4\xc7\xe1\x9d\xea\x30\xfa\x46\x7f\x0a\x45\x32\xf0\x13\x05\xf8\x88\xd1\x63\x3c\xaa\x1e\xb2\xf5\x36\xdb\x46\x69\x1d\x6d\x0f\x8d\xee\x4a\x0b\xc6\x26\x14\xe0\x01\xd8\x8c\x3f\x03\x64\x63\x7d\x8f\x51\x31\x05\x78\x82\x6f\xe3\x6d\x9b\xf0\x33\x53\x49\x6f\x90\x95\xd6\x2b\xf0\x85\x23\xaf\x92\xa1\x6f\x2e\xfa\x09\x67\x86\x94\xf6\x3b\xb8\x2c\x8d\x32\xbb\x20\x5f\x04\xb8\xd2\x85\x52\xc6\x18\xf5\xf7\xd3\xda\x18\xd3\xb9\xfc\xf7\x62\x1c\xc5\xcc\x14\x02\xb0\xe2\x12\x60\x51\xb8\x8c\x2f\x8c\x7e\x78\x50\x5a\x33\x08\x15\x76\x30\x9f\x45\xf2\x20\x4a\xeb\x01\xb8\x9d\x8f\x8e\x90\x4f\x9f\x01\x65\xfa\xf2\x62\xb3\x7b\xde\x21\xf7\x28\xd9\xe6\x72\xa5\x9a\x76\x88\xd8\x1e\x24\x59\x77\x5d\xd8\x1f\x0b\x8a\x90\x5f\x88\x3b\x8c\xc7\x79\x8e\x5b\xe2\xd3\x9d\x44\x01\x1d\x9e\x08\x46\xbb\x69\xc8\x0e\x3d\xef\x45\x6e\x10\x20\xfa\x44\x18\xf3\xa4\x9d\xc8\xd7\x34\xcf\x83\xae\x69\x7b\x14\x47\x03\xf0\x6b\xbd\x9b\x0b\x2b\x60\x7d\x5c\x4b\xce\x3f\xba\xb0\x9e\xd9\x8a\x19\xef\x1f\xd6\x25\xe0\x57\x52\xc7\x1e\x6f\x63\x5c\xa5\xf5\x6d\xc0\xfd\x63\xfb\xbb\xdf\xcc\x18\x99\x6a\x64\x57\x8e\x5e\x5b\xe0\x66\xf3\xfe\xb7\xc5\x6f\xb4\x73\xbf\xa5\xaf\xc5\x97\x0b\x9f\x5e\x9e\x10\xeb\x4d\x9e\xff\xdd\x6e\x2b\xe3\x67\x00\x00\x00\xff\xff\x99\x4e\xc0\xe4\xc9\x06\x00\x00") func rolebindingsYamlBytes() ([]byte, error) { return bindataRead( diff --git a/pkg/etcd/apiaddresses_controller.go b/pkg/etcd/apiaddresses_controller.go index 916f2b2148..444faaa3c2 100644 --- a/pkg/etcd/apiaddresses_controller.go +++ b/pkg/etcd/apiaddresses_controller.go @@ -7,9 +7,9 @@ import ( "github.com/k3s-io/k3s/pkg/util" "github.com/sirupsen/logrus" - v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/fields" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" @@ -17,20 +17,20 @@ import ( ) func registerEndpointsHandlers(ctx context.Context, etcd *ETCD) { - endpoints := etcd.config.Runtime.Core.Core().V1().Endpoints() - fieldSelector := fields.Set{metav1.ObjectNameField: "kubernetes"}.String() + endpointslice := etcd.config.Runtime.Discovery.Discovery().V1().EndpointSlice() + labelSelector := labels.Set{discoveryv1.LabelServiceName: "kubernetes"}.String() lw := &cache.ListWatch{ ListFunc: func(options metav1.ListOptions) (object runtime.Object, e error) { - options.FieldSelector = fieldSelector - return endpoints.List(metav1.NamespaceDefault, options) + options.LabelSelector = labelSelector + return endpointslice.List(metav1.NamespaceDefault, options) }, WatchFunc: func(options metav1.ListOptions) (i watch.Interface, e error) { - options.FieldSelector = fieldSelector - return endpoints.Watch(metav1.NamespaceDefault, options) + options.LabelSelector = labelSelector + return endpointslice.Watch(metav1.NamespaceDefault, options) }, } - _, _, watch, done := toolswatch.NewIndexerInformerWatcher(lw, &v1.Endpoints{}) + _, _, watch, done := toolswatch.NewIndexerInformerWatcher(lw, &discoveryv1.EndpointSlice{}) go func() { <-ctx.Done() @@ -44,7 +44,7 @@ func registerEndpointsHandlers(ctx context.Context, etcd *ETCD) { } logrus.Infof("Starting managed etcd apiserver addresses controller") - go h.watchEndpoints(ctx) + go h.watchEndpointSlice(ctx) } type handler struct { @@ -53,20 +53,20 @@ type handler struct { } // 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) { +// api addresses endpoint slices found in the kubernetes service in the default namespace +func (h *handler) watchEndpointSlice(ctx context.Context) { for { select { case <-ctx.Done(): return case ev, ok := <-h.watch.ResultChan(): - endpoint, ok := ev.Object.(*v1.Endpoints) + slice, ok := ev.Object.(*discoveryv1.EndpointSlice) if !ok { - logrus.Fatalf("Failed to watch apiserver addresses: could not convert event object to endpoint: %v", ev) + logrus.Fatalf("Failed to watch apiserver addresses: could not convert event object to endpointslice: %v", ev) } w := &bytes.Buffer{} - if err := json.NewEncoder(w).Encode(util.GetAddresses(endpoint)); err != nil { + if err := json.NewEncoder(w).Encode(util.GetAddressesFromSlices(*slice)); err != nil { logrus.Warnf("Failed to encode apiserver addresses: %v", err) continue } diff --git a/pkg/server/context.go b/pkg/server/context.go index 9c34b6d5d6..9483fbafdf 100644 --- a/pkg/server/context.go +++ b/pkg/server/context.go @@ -3,10 +3,10 @@ package server import ( "context" + "github.com/k3s-io/api/pkg/generated/controllers/k3s.cattle.io" helmcrd "github.com/k3s-io/helm-controller/pkg/crd" "github.com/k3s-io/helm-controller/pkg/generated/controllers/helm.cattle.io" addoncrd "github.com/k3s-io/k3s/pkg/crd" - "github.com/k3s-io/api/pkg/generated/controllers/k3s.cattle.io" "github.com/k3s-io/k3s/pkg/util" "github.com/k3s-io/k3s/pkg/version" pkgerrors "github.com/pkg/errors" @@ -14,6 +14,7 @@ import ( "github.com/rancher/wrangler/v3/pkg/generated/controllers/apps" "github.com/rancher/wrangler/v3/pkg/generated/controllers/batch" "github.com/rancher/wrangler/v3/pkg/generated/controllers/core" + "github.com/rancher/wrangler/v3/pkg/generated/controllers/discovery" "github.com/rancher/wrangler/v3/pkg/generated/controllers/rbac" "github.com/rancher/wrangler/v3/pkg/start" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -23,14 +24,15 @@ import ( ) type Context struct { - K3s *k3s.Factory - Helm *helm.Factory - Batch *batch.Factory - Apps *apps.Factory - Auth *rbac.Factory - Core *core.Factory - K8s kubernetes.Interface - Event record.EventRecorder + K3s *k3s.Factory + Helm *helm.Factory + Batch *batch.Factory + Apps *apps.Factory + Auth *rbac.Factory + Core *core.Factory + Discovery *discovery.Factory + K8s kubernetes.Interface + Event record.EventRecorder } func (c *Context) Start(ctx context.Context) error { @@ -62,14 +64,15 @@ func NewContext(ctx context.Context, config *Config, forServer bool) (*Context, } return &Context{ - K3s: k3s.NewFactoryFromConfigOrDie(restConfig), - Helm: helm.NewFactoryFromConfigOrDie(restConfig), - K8s: k8s, - Auth: rbac.NewFactoryFromConfigOrDie(restConfig), - Apps: apps.NewFactoryFromConfigOrDie(restConfig), - Batch: batch.NewFactoryFromConfigOrDie(restConfig), - Core: core.NewFactoryFromConfigOrDie(restConfig), - Event: recorder, + K3s: k3s.NewFactoryFromConfigOrDie(restConfig), + Helm: helm.NewFactoryFromConfigOrDie(restConfig), + K8s: k8s, + Auth: rbac.NewFactoryFromConfigOrDie(restConfig), + Apps: apps.NewFactoryFromConfigOrDie(restConfig), + Batch: batch.NewFactoryFromConfigOrDie(restConfig), + Core: core.NewFactoryFromConfigOrDie(restConfig), + Discovery: discovery.NewFactoryFromConfigOrDie(restConfig), + Event: recorder, }, nil } diff --git a/pkg/server/handlers/handlers.go b/pkg/server/handlers/handlers.go index 162e2750a4..4c2650699a 100644 --- a/pkg/server/handlers/handlers.go +++ b/pkg/server/handlers/handlers.go @@ -23,11 +23,13 @@ import ( pkgerrors "github.com/pkg/errors" certutil "github.com/rancher/dynamiclistener/cert" "github.com/sirupsen/logrus" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/util/json" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apiserver/pkg/authentication/user" - typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" + typeddiscoveryv1 "k8s.io/client-go/kubernetes/typed/discovery/v1" ) func CACerts(config *config.Control) http.Handler { @@ -317,20 +319,21 @@ type addressGetter func(ctx context.Context) <-chan []string // kubernetesGetter returns a function that returns a channel that can be read to get apiserver addresses from kubernetes endpoints func kubernetesGetter(control *config.Control) addressGetter { - var endpointsClient typedcorev1.EndpointsInterface + var endpointSliceClient typeddiscoveryv1.EndpointSliceInterface + labelSelector := labels.Set{discoveryv1.LabelServiceName: "kubernetes"}.String() return func(ctx context.Context) <-chan []string { ch := make(chan []string, 1) go func() { - if endpointsClient == nil { + if endpointSliceClient == nil { if control.Runtime.K8s != nil { - endpointsClient = control.Runtime.K8s.CoreV1().Endpoints(metav1.NamespaceDefault) + endpointSliceClient = control.Runtime.K8s.DiscoveryV1().EndpointSlices(metav1.NamespaceDefault) } } - if endpointsClient != nil { - if endpoint, err := endpointsClient.Get(ctx, "kubernetes", metav1.GetOptions{}); err != nil { + if endpointSliceClient != nil { + if endpointSlices, err := endpointSliceClient.List(ctx, metav1.ListOptions{LabelSelector: labelSelector}); err != nil { logrus.Debugf("Failed to get apiserver addresses from kubernetes: %v", err) } else { - ch <- util.GetAddresses(endpoint) + ch <- util.GetAddressesFromSlices(endpointSlices.Items...) } } close(ch) diff --git a/pkg/server/server.go b/pkg/server/server.go index 911572218a..3d02bfd8d1 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -128,6 +128,7 @@ func runControllers(ctx context.Context, config *Config) error { controlConfig.Runtime.K3s = sc.K3s controlConfig.Runtime.Event = sc.Event controlConfig.Runtime.Core = sc.Core + controlConfig.Runtime.Discovery = sc.Discovery for name, cb := range controlConfig.Runtime.ClusterControllerStarts { go runOrDie(ctx, name, cb) diff --git a/pkg/util/api.go b/pkg/util/api.go index bf13c2412a..a2bf6fadb4 100644 --- a/pkg/util/api.go +++ b/pkg/util/api.go @@ -14,6 +14,7 @@ import ( "github.com/sirupsen/logrus" authorizationv1 "k8s.io/api/authorization/v1" v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/util/wait" @@ -51,6 +52,27 @@ func GetAddresses(endpoint *v1.Endpoints) []string { return serverAddresses } +func GetAddressesFromSlices(slices ...discoveryv1.EndpointSlice) []string { + serverAddresses := []string{} + for _, slice := range slices { + var port string + if len(slice.Ports) > 0 && slice.Ports[0].Port != nil { + port = strconv.Itoa(int(*slice.Ports[0].Port)) + } + if port == "" { + port = "443" + } + for _, endpoint := range slice.Endpoints { + if endpoint.Conditions.Ready == nil || *endpoint.Conditions.Ready == true { + for _, address := range endpoint.Addresses { + serverAddresses = append(serverAddresses, net.JoinHostPort(address, port)) + } + } + } + } + return serverAddresses +} + // WaitForAPIServerReady waits for the API server's /readyz endpoint to report "ok" with timeout. // This is modified from WaitForAPIServer from the Kubernetes controller-manager app, but checks the // readyz endpoint instead of the deprecated healthz endpoint, and supports context.