From 41dc2d3b0240c925110152c28fb41339d57d14e7 Mon Sep 17 00:00:00 2001 From: Dan Winship Date: Thu, 22 May 2025 13:30:11 -0400 Subject: [PATCH 1/3] Add utilities for getting EndpointSlices for a Service Co-Authored-by: Jordan Liggitt --- .../apiserver/pkg/util/proxy/endpointslice.go | 114 ++++++++++ .../pkg/util/proxy/endpointslice_test.go | 200 ++++++++++++++++++ 2 files changed, 314 insertions(+) create mode 100644 staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice.go create mode 100644 staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice_test.go diff --git a/staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice.go b/staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice.go new file mode 100644 index 00000000000..865bc794b53 --- /dev/null +++ b/staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice.go @@ -0,0 +1,114 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "fmt" + + discoveryv1 "k8s.io/api/discovery/v1" + "k8s.io/apimachinery/pkg/labels" + discoveryv1informer "k8s.io/client-go/informers/discovery/v1" + discoveryv1lister "k8s.io/client-go/listers/discovery/v1" + "k8s.io/client-go/tools/cache" +) + +// EndpointSliceGetter is an interface for a helper that lets you easily get all +// EndpointSlices for a Service. +type EndpointSliceGetter interface { + // GetEndpointSlices returns all of the known slices associated with the given + // service. If there are no slices associated with the service, it will return an + // empty list, not an error. + GetEndpointSlices(namespaceName, serviceName string) ([]*discoveryv1.EndpointSlice, error) +} + +const indexKey = "namespaceName_serviceName" + +// ensureServiceNameIndexer ensures that indexer has a namespace/serviceName indexer +func ensureServiceNameIndexer(indexer cache.Indexer) error { + if _, exists := indexer.GetIndexers()[indexKey]; exists { + return nil + } + err := indexer.AddIndexers(map[string]cache.IndexFunc{indexKey: func(obj any) ([]string, error) { + ep, ok := obj.(*discoveryv1.EndpointSlice) + if !ok { + return nil, fmt.Errorf("expected *discoveryv1.EndpointSlice, got %T", obj) + } + serviceName, labelExists := ep.Labels[discoveryv1.LabelServiceName] + if !labelExists { + // Not associated with a service; don't add to this index. + return nil, nil + } + return []string{ep.Namespace + "/" + serviceName}, nil + }}) + if err != nil { + // Check if the indexer exists now; if so, that means we were racing with + // another thread, and they successfully installed the indexer, so we can + // ignore the error. + if _, exists := indexer.GetIndexers()[indexKey]; exists { + err = nil + } + } + return err +} + +// NewEndpointSliceIndexerGetter returns an EndpointSliceGetter that wraps an informer and +// updates its indexes so that you can efficiently find the EndpointSlices associated with +// a Service later. (Note that sliceInformer will continue the additional indexing for as +// long as it runs, even if if the EndpointSliceGetter is destroyed. Use +// NewEndpointSliceListerGetter if you want want to fetch EndpointSlices without changing +// the underlying cache.) +func NewEndpointSliceIndexerGetter(sliceInformer discoveryv1informer.EndpointSliceInformer) (EndpointSliceGetter, error) { + indexer := sliceInformer.Informer().GetIndexer() + if err := ensureServiceNameIndexer(indexer); err != nil { + return nil, err + } + return &endpointSliceIndexerGetter{indexer: indexer}, nil +} + +type endpointSliceIndexerGetter struct { + indexer cache.Indexer +} + +func (e *endpointSliceIndexerGetter) GetEndpointSlices(namespaceName, serviceName string) ([]*discoveryv1.EndpointSlice, error) { + objs, err := e.indexer.ByIndex(indexKey, namespaceName+"/"+serviceName) + if err != nil { + return nil, err + } + eps := make([]*discoveryv1.EndpointSlice, 0, len(objs)) + for _, obj := range objs { + ep, ok := obj.(*discoveryv1.EndpointSlice) + if !ok { + return nil, fmt.Errorf("expected *discoveryv1.EndpointSlice, got %T", obj) + } + eps = append(eps, ep) + } + return eps, nil +} + +// NewEndpointSliceListerGetter returns an EndpointSliceGetter that uses a lister to do a +// full selection on every lookup. +func NewEndpointSliceListerGetter(sliceLister discoveryv1lister.EndpointSliceLister) (EndpointSliceGetter, error) { + return &endpointSliceListerGetter{lister: sliceLister}, nil +} + +type endpointSliceListerGetter struct { + lister discoveryv1lister.EndpointSliceLister +} + +func (e *endpointSliceListerGetter) GetEndpointSlices(namespaceName, serviceName string) ([]*discoveryv1.EndpointSlice, error) { + return e.lister.EndpointSlices(namespaceName).List(labels.SelectorFromSet(labels.Set{discoveryv1.LabelServiceName: serviceName})) +} diff --git a/staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice_test.go b/staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice_test.go new file mode 100644 index 00000000000..8d98293a8e7 --- /dev/null +++ b/staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice_test.go @@ -0,0 +1,200 @@ +/* +Copyright 2025 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + + discoveryv1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/informers" + clientset "k8s.io/client-go/kubernetes" + clientsetfake "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/tools/cache" +) + +func makeEndpointSlice(namespace, service, slice int, ip string) *discoveryv1.EndpointSlice { + namespaceName := fmt.Sprintf("namespace%d", namespace) + serviceName := fmt.Sprintf("service%d", service) + sliceName := fmt.Sprintf("service%d-%d%d%d", service, slice, slice, slice) + return &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: sliceName, + Namespace: namespaceName, + Labels: map[string]string{ + discoveryv1.LabelServiceName: serviceName, + }, + }, + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{ip}, + }, + }, + } +} + +// Note that we resuse the same service names in the two namespaces to test proper +// indexing/namespacing. +var ( + // namespace1 service1: + // - initial: 1 slice + // - phase1: add 1 slice + // - phase2: no change + n1s1FirstSlice = makeEndpointSlice(1, 1, 1, "10.1.1.1") + n1s1SecondSlice = makeEndpointSlice(1, 1, 2, "10.1.1.2") + n1s1InitialSlices = []*discoveryv1.EndpointSlice{n1s1FirstSlice} + n1s1Phase1Slices = []*discoveryv1.EndpointSlice{n1s1FirstSlice, n1s1SecondSlice} + n1s1Phase2Slices = []*discoveryv1.EndpointSlice{n1s1FirstSlice, n1s1SecondSlice} + + // namespace1 service2: + // - initial: 1 slice + // - phase1: update slice + // - phase2: delete slice + n1s2FirstSlice = makeEndpointSlice(1, 2, 1, "10.1.2.1") + n1s2UpdatedSlice = makeEndpointSlice(1, 2, 1, "10.1.2.99") + n1s2InitialSlices = []*discoveryv1.EndpointSlice{n1s2FirstSlice} + n1s2Phase1Slices = []*discoveryv1.EndpointSlice{n1s2UpdatedSlice} + n1s2Phase2Slices = []*discoveryv1.EndpointSlice{} + + // namespace2 service 1: + // - initial: 2 slices + // - phase1: delete first slice + // - phase2: delete second slice + n2s1FirstSlice = makeEndpointSlice(2, 1, 1, "10.2.1.1") + n2s1SecondSlice = makeEndpointSlice(2, 1, 2, "10.2.1.2") + n2s1InitialSlices = []*discoveryv1.EndpointSlice{n2s1FirstSlice, n2s1SecondSlice} + n2s1Phase1Slices = []*discoveryv1.EndpointSlice{n2s1SecondSlice} + n2s1Phase2Slices = []*discoveryv1.EndpointSlice{} + + // namespace2 service 2: + // - initial: no slices + // - phase1: no change + // - phase2: create slice + n2s2FirstSlice = makeEndpointSlice(2, 2, 1, "10.2.2.1") + n2s2InitialSlices = []*discoveryv1.EndpointSlice{} + n2s2Phase1Slices = []*discoveryv1.EndpointSlice{} + n2s2Phase2Slices = []*discoveryv1.EndpointSlice{n2s2FirstSlice} + + initialSlices = []runtime.Object{ + n1s1FirstSlice, n1s2FirstSlice, n2s1FirstSlice, n2s1SecondSlice, + } +) + +func assertSlices(t *testing.T, getter EndpointSliceGetter, namespace, service string, expected []*discoveryv1.EndpointSlice) { + t.Helper() + + // Poll because the informers may not sync immediately + var lastErr error + err := wait.PollUntilContextTimeout(context.Background(), 10*time.Millisecond, time.Second, true, func(ctx context.Context) (bool, error) { + slices, err := getter.GetEndpointSlices(namespace, service) + if err != nil { + lastErr = fmt.Errorf("unexpected error getting %s/%s slices: %w", namespace, service, err) + return false, nil + } + // cmp.Diff doesn't deal with nil vs [] + if len(expected) == 0 && len(slices) == 0 { + return true, nil + } + if diff := cmp.Diff(expected, slices); diff != "" { + lastErr = fmt.Errorf("slices for %s/%s did not match expectation:\n%s", namespace, service, diff) + return false, nil + } + return true, nil + }) + if err != nil { + t.Errorf("%s", lastErr) + } +} + +func testGetter(t *testing.T, client clientset.Interface, getter EndpointSliceGetter) { + ctx := context.Background() + + // Check initial state + assertSlices(t, getter, "namespace1", "service1", n1s1InitialSlices) + assertSlices(t, getter, "namespace1", "service2", n1s2InitialSlices) + assertSlices(t, getter, "namespace2", "service1", n2s1InitialSlices) + assertSlices(t, getter, "namespace2", "service2", n2s2InitialSlices) + + _, err := client.DiscoveryV1().EndpointSlices("namespace1").Create(ctx, n1s1SecondSlice, metav1.CreateOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + _, err = client.DiscoveryV1().EndpointSlices("namespace1").Update(ctx, n1s2UpdatedSlice, metav1.UpdateOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + err = client.DiscoveryV1().EndpointSlices("namespace2").Delete(ctx, n2s1FirstSlice.Name, metav1.DeleteOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + assertSlices(t, getter, "namespace1", "service1", n1s1Phase1Slices) + assertSlices(t, getter, "namespace1", "service2", n1s2Phase1Slices) + assertSlices(t, getter, "namespace2", "service1", n2s1Phase1Slices) + assertSlices(t, getter, "namespace2", "service2", n2s2Phase1Slices) + + err = client.DiscoveryV1().EndpointSlices("namespace1").Delete(ctx, n1s2FirstSlice.Name, metav1.DeleteOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + err = client.DiscoveryV1().EndpointSlices("namespace2").Delete(ctx, n2s1SecondSlice.Name, metav1.DeleteOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + _, err = client.DiscoveryV1().EndpointSlices("namespace2").Create(ctx, n2s2FirstSlice, metav1.CreateOptions{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + assertSlices(t, getter, "namespace1", "service1", n1s1Phase2Slices) + assertSlices(t, getter, "namespace1", "service2", n1s2Phase2Slices) + assertSlices(t, getter, "namespace2", "service1", n2s1Phase2Slices) + assertSlices(t, getter, "namespace2", "service2", n2s2Phase2Slices) +} + +func TestNewEndpointSliceIndexerGetter(t *testing.T) { + client := clientsetfake.NewSimpleClientset(initialSlices...) + informerFactory := informers.NewSharedInformerFactory(client, 30*time.Second) + getter, err := NewEndpointSliceIndexerGetter(informerFactory.Discovery().V1().EndpointSlices()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + informerFactory.Start(wait.NeverStop) + cache.WaitForCacheSync(nil, informerFactory.Discovery().V1().EndpointSlices().Informer().HasSynced) + + testGetter(t, client, getter) +} + +func TestNewEndpointSliceListerGetter(t *testing.T) { + client := clientsetfake.NewSimpleClientset(initialSlices...) + informerFactory := informers.NewSharedInformerFactory(client, 30*time.Second) + getter, err := NewEndpointSliceListerGetter(informerFactory.Discovery().V1().EndpointSlices().Lister()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + informerFactory.Start(wait.NeverStop) + cache.WaitForCacheSync(nil, informerFactory.Discovery().V1().EndpointSlices().Informer().HasSynced) + + testGetter(t, client, getter) +} From d70e7e223cf5a02d0bdf557a57da11a276ec1e8c Mon Sep 17 00:00:00 2001 From: Dan Winship Date: Sun, 26 Jan 2025 08:55:01 -0500 Subject: [PATCH 2/3] Port aggregated apiserver discovery to EndpointSlices Co-Authored-by: Jordan Liggitt --- cmd/kube-apiserver/app/server.go | 20 +- .../k8s.io/apiserver/pkg/util/proxy/proxy.go | 66 ++- .../apiserver/pkg/util/proxy/proxy_test.go | 429 +++++++++++++++--- .../pkg/apiserver/apiserver.go | 2 +- .../pkg/apiserver/resolvers.go | 12 +- .../remote/remote_available_controller.go | 123 +++-- .../remote_available_controller_test.go | 134 ++++-- 7 files changed, 595 insertions(+), 191 deletions(-) diff --git a/cmd/kube-apiserver/app/server.go b/cmd/kube-apiserver/app/server.go index 042ddc9714a..c835092047a 100644 --- a/cmd/kube-apiserver/app/server.go +++ b/cmd/kube-apiserver/app/server.go @@ -26,6 +26,7 @@ import ( "os" "github.com/spf13/cobra" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" utilerrors "k8s.io/apimachinery/pkg/util/errors" utilruntime "k8s.io/apimachinery/pkg/util/runtime" @@ -36,6 +37,7 @@ import ( serverstorage "k8s.io/apiserver/pkg/server/storage" utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/apiserver/pkg/util/notfoundhandler" + "k8s.io/apiserver/pkg/util/proxy" "k8s.io/apiserver/pkg/util/webhook" clientgoinformers "k8s.io/client-go/informers" "k8s.io/client-go/rest" @@ -215,7 +217,10 @@ func CreateKubeAPIServerConfig( return nil, nil, nil, fmt.Errorf("failed to create admission plugin initializer: %w", err) } - serviceResolver := buildServiceResolver(opts.EnableAggregatorRouting, genericConfig.LoopbackClientConfig.Host, versionedInformers) + serviceResolver, err := buildServiceResolver(opts.EnableAggregatorRouting, genericConfig.LoopbackClientConfig.Host, versionedInformers) + if err != nil { + return nil, nil, nil, fmt.Errorf("error building service resolver: %w", err) + } controlplaneConfig, admissionInitializers, err := controlplaneapiserver.CreateConfig(opts.CompletedOptions, genericConfig, versionedInformers, storageFactory, serviceResolver, kubeInitializers) if err != nil { return nil, nil, nil, err @@ -274,16 +279,21 @@ func SetServiceResolverForTests(resolver webhook.ServiceResolver) func() { } } -func buildServiceResolver(enabledAggregatorRouting bool, hostname string, informer clientgoinformers.SharedInformerFactory) webhook.ServiceResolver { +func buildServiceResolver(enabledAggregatorRouting bool, hostname string, informer clientgoinformers.SharedInformerFactory) (webhook.ServiceResolver, error) { if testServiceResolver != nil { - return testServiceResolver + return testServiceResolver, nil + } + + endpointSliceGetter, err := proxy.NewEndpointSliceIndexerGetter(informer.Discovery().V1().EndpointSlices()) + if err != nil { + return nil, err } var serviceResolver webhook.ServiceResolver if enabledAggregatorRouting { serviceResolver = aggregatorapiserver.NewEndpointServiceResolver( informer.Core().V1().Services().Lister(), - informer.Core().V1().Endpoints().Lister(), + endpointSliceGetter, ) } else { serviceResolver = aggregatorapiserver.NewClusterIPServiceResolver( @@ -295,5 +305,5 @@ func buildServiceResolver(enabledAggregatorRouting bool, hostname string, inform if localHost, err := url.Parse(hostname); err == nil { serviceResolver = aggregatorapiserver.NewLoopbackServiceResolver(serviceResolver, localHost) } - return serviceResolver + return serviceResolver, nil } diff --git a/staging/src/k8s.io/apiserver/pkg/util/proxy/proxy.go b/staging/src/k8s.io/apiserver/pkg/util/proxy/proxy.go index 9e0bdfaf03c..d64120163d6 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/proxy/proxy.go +++ b/staging/src/k8s.io/apiserver/pkg/util/proxy/proxy.go @@ -23,11 +23,13 @@ import ( "net" "net/http" "net/url" + "sort" "strconv" "strings" "time" - "k8s.io/api/core/v1" + v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" "k8s.io/apimachinery/pkg/api/errors" utilnet "k8s.io/apimachinery/pkg/util/net" auditinternal "k8s.io/apiserver/pkg/apis/audit" @@ -52,7 +54,9 @@ func findServicePort(svc *v1.Service, port int32) (*v1.ServicePort, error) { } // ResolveEndpoint returns a URL to which one can send traffic for the specified service. -func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.EndpointsLister, namespace, id string, port int32) (*url.URL, error) { +// If the service is dual-stack, the URL will preferentially point to an endpoint of the +// service's primary IP family. +func ResolveEndpoint(services listersv1.ServiceLister, endpointSlices EndpointSliceGetter, namespace, id string, port int32) (*url.URL, error) { svc, err := services.Services(namespace).Get(id) if err != nil { return nil, err @@ -70,32 +74,52 @@ func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.Endpo return nil, err } - eps, err := endpoints.Endpoints(namespace).Get(svc.Name) + slices, err := endpointSlices.GetEndpointSlices(namespace, svc.Name) if err != nil { return nil, err } - if len(eps.Subsets) == 0 { + if len(slices) == 0 { return nil, errors.NewServiceUnavailable(fmt.Sprintf("no endpoints available for service %q", svc.Name)) + } else if len(slices) > 1 && len(svc.Spec.IPFamilies) > 0 { + // If there are multiple slices, we want to look at them in a random + // order, but we need to look at all of the slices of the primary IP + // family first. + preferredAddressType := discoveryv1.AddressType(svc.Spec.IPFamilies[0]) + randomOrder := rand.Perm(len(slices)) + sort.Slice(slices, func(i, j int) bool { + if slices[i].AddressType != slices[j].AddressType { + return slices[i].AddressType == preferredAddressType + } + return randomOrder[i] < randomOrder[j] + }) } - // Pick a random Subset to start searching from. - ssSeed := rand.Intn(len(eps.Subsets)) + // Find a slice that has the port. + for _, slice := range slices { + for i := range slice.Ports { + if slice.Ports[i].Name == nil || *slice.Ports[i].Name != svcPort.Name { + continue + } + if slice.Ports[i].Port == nil { + continue + } + if len(slice.Endpoints) == 0 { + continue + } - // Find a Subset that has the port. - for ssi := 0; ssi < len(eps.Subsets); ssi++ { - ss := &eps.Subsets[(ssSeed+ssi)%len(eps.Subsets)] - if len(ss.Addresses) == 0 { - continue - } - for i := range ss.Ports { - if ss.Ports[i].Name == svcPort.Name { - // Pick a random address. - ip := ss.Addresses[rand.Intn(len(ss.Addresses))].IP - port := int(ss.Ports[i].Port) - return &url.URL{ - Scheme: "https", - Host: net.JoinHostPort(ip, strconv.Itoa(port)), - }, nil + // Starting from a random index, find a Ready endpoint + offset := rand.Intn(len(slice.Endpoints)) + for epi := range slice.Endpoints { + ep := &slice.Endpoints[(epi+offset)%len(slice.Endpoints)] + if ep.Conditions.Ready == nil || *ep.Conditions.Ready { + // (Addresses is an array but only Addresses[0] is used.) + ip := ep.Addresses[0] + port := int(*slice.Ports[i].Port) + return &url.URL{ + Scheme: "https", + Host: net.JoinHostPort(ip, strconv.Itoa(port)), + }, nil + } } } } diff --git a/staging/src/k8s.io/apiserver/pkg/util/proxy/proxy_test.go b/staging/src/k8s.io/apiserver/pkg/util/proxy/proxy_test.go index 760c0d16399..c4eca906914 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/proxy/proxy_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/proxy/proxy_test.go @@ -21,28 +21,39 @@ import ( "testing" v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/apimachinery/pkg/util/sets" v1listers "k8s.io/client-go/listers/core/v1" + discoveryv1listers "k8s.io/client-go/listers/discovery/v1" "k8s.io/client-go/tools/cache" + "k8s.io/utils/ptr" ) func TestResolve(t *testing.T) { - matchingEndpoints := func(svc *v1.Service) []*v1.Endpoints { - ports := []v1.EndpointPort{} + matchingEndpointSlices := func(svc *v1.Service) []*discoveryv1.EndpointSlice { + ports := []discoveryv1.EndpointPort{} for _, p := range svc.Spec.Ports { if p.TargetPort.Type != intstr.Int { continue } - ports = append(ports, v1.EndpointPort{Name: p.Name, Port: p.TargetPort.IntVal}) + ports = append(ports, discoveryv1.EndpointPort{Name: &p.Name, Port: &p.TargetPort.IntVal}) } - return []*v1.Endpoints{{ - ObjectMeta: metav1.ObjectMeta{Namespace: svc.Namespace, Name: svc.Name}, - Subsets: []v1.EndpointSubset{{ - Addresses: []v1.EndpointAddress{{Hostname: "dummy-host", IP: "127.0.0.1"}}, - Ports: ports, + return []*discoveryv1.EndpointSlice{{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: svc.Namespace, + Name: svc.Name + "-xxx", + Labels: map[string]string{ + discoveryv1.LabelServiceName: svc.Name, + }, + }, + Endpoints: []discoveryv1.Endpoint{{ + Hostname: ptr.To("dummy-host"), + Addresses: []string{"127.0.0.1"}, }}, + Ports: ports, }} } @@ -52,9 +63,9 @@ func TestResolve(t *testing.T) { } tests := []struct { - name string - services []*v1.Service - endpoints func(svc *v1.Service) []*v1.Endpoints + name string + services []*v1.Service + endpointSlices func(svc *v1.Service) []*discoveryv1.EndpointSlice clusterMode expectation endpointMode expectation @@ -73,7 +84,7 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: matchingEndpoints, + endpointSlices: matchingEndpointSlices, clusterMode: expectation{error: true}, endpointMode: expectation{error: true}, @@ -93,13 +104,13 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: matchingEndpoints, + endpointSlices: matchingEndpointSlices, clusterMode: expectation{url: "https://hit:443"}, endpointMode: expectation{url: "https://127.0.0.1:1443"}, }, { - name: "cluster ip without endpoints", + name: "cluster ip without endpointslices", services: []*v1.Service{ { ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"}, @@ -113,13 +124,13 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: nil, + endpointSlices: nil, clusterMode: expectation{url: "https://hit:443"}, endpointMode: expectation{error: true}, }, { - name: "endpoint without subset", + name: "endpointslice without addresses", services: []*v1.Service{ { ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"}, @@ -133,38 +144,15 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: func(svc *v1.Service) []*v1.Endpoints { - return []*v1.Endpoints{{ - ObjectMeta: metav1.ObjectMeta{Namespace: svc.Namespace, Name: svc.Name}, - Subsets: []v1.EndpointSubset{}, - }} - }, - - clusterMode: expectation{url: "https://hit:443"}, - endpointMode: expectation{error: true}, - }, - { - name: "endpoint subset without addresses", - services: []*v1.Service{ - { - ObjectMeta: metav1.ObjectMeta{Namespace: "one", Name: "alfa"}, - Spec: v1.ServiceSpec{ - Type: v1.ServiceTypeClusterIP, - ClusterIP: "hit", - Ports: []v1.ServicePort{ - {Name: "https", Port: 443, TargetPort: intstr.FromInt32(1443)}, - {Port: 1234, TargetPort: intstr.FromInt32(1234)}, + endpointSlices: func(svc *v1.Service) []*discoveryv1.EndpointSlice { + return []*discoveryv1.EndpointSlice{{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: svc.Namespace, + Name: svc.Name + "-xxx", + Labels: map[string]string{ + discoveryv1.LabelServiceName: svc.Name, }, }, - }, - }, - endpoints: func(svc *v1.Service) []*v1.Endpoints { - return []*v1.Endpoints{{ - ObjectMeta: metav1.ObjectMeta{Namespace: svc.Namespace, Name: svc.Name}, - Subsets: []v1.EndpointSubset{{ - Addresses: []v1.EndpointAddress{}, - Ports: []v1.EndpointPort{{Name: "https", Port: 443}}, - }}, }} }, @@ -182,7 +170,7 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: nil, + endpointSlices: nil, clusterMode: expectation{error: true}, endpointMode: expectation{error: true}, @@ -202,7 +190,7 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: matchingEndpoints, + endpointSlices: matchingEndpointSlices, clusterMode: expectation{url: "https://lb:443"}, endpointMode: expectation{url: "https://127.0.0.1:1443"}, @@ -222,7 +210,7 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: matchingEndpoints, + endpointSlices: matchingEndpointSlices, clusterMode: expectation{url: "https://np:443"}, endpointMode: expectation{url: "https://127.0.0.1:1443"}, @@ -238,7 +226,7 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: nil, + endpointSlices: nil, clusterMode: expectation{url: "https://foo.bar.com:443"}, endpointMode: expectation{error: true}, @@ -253,15 +241,15 @@ func TestResolve(t *testing.T) { }, }, }, - endpoints: nil, + endpointSlices: nil, clusterMode: expectation{error: true}, endpointMode: expectation{error: true}, }, { - name: "missing service", - services: nil, - endpoints: nil, + name: "missing service", + services: nil, + endpointSlices: nil, clusterMode: expectation{error: true}, endpointMode: expectation{error: true}, @@ -277,13 +265,13 @@ func TestResolve(t *testing.T) { } } - endpointCache := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) - endpointLister := v1listers.NewEndpointsLister(endpointCache) - if test.endpoints != nil { + endpointSliceCache := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + endpointSliceLister := discoveryv1listers.NewEndpointSliceLister(endpointSliceCache) + if test.endpointSlices != nil { for _, svc := range test.services { - for _, ep := range test.endpoints(svc) { - if err := endpointCache.Add(ep); err != nil { - t.Fatalf("%s unexpected endpoint add error: %v", test.name, err) + for _, ep := range test.endpointSlices(svc) { + if err := endpointSliceCache.Add(ep); err != nil { + t.Fatalf("%s unexpected endpointslice add error: %v", test.name, err) } } } @@ -305,7 +293,326 @@ func TestResolve(t *testing.T) { clusterURL, err := ResolveCluster(serviceLister, "one", "alfa", 443) check("cluster", test.clusterMode, clusterURL, err) - endpointURL, err := ResolveEndpoint(serviceLister, endpointLister, "one", "alfa", 443) + endpointSliceGetter, err := NewEndpointSliceListerGetter(endpointSliceLister) + if err != nil { + t.Fatal(err) + } + + endpointURL, err := ResolveEndpoint(serviceLister, endpointSliceGetter, "one", "alfa", 443) check("endpoint", test.endpointMode, endpointURL, err) } } + +// Tests that ResolveEndpoint picks randomly among endpoints in the expected way +func TestResolveEndpointDistribution(t *testing.T) { + singleStackService := &v1.Service{ + ObjectMeta: metav1.ObjectMeta{Namespace: "test", Name: "single-stack"}, + Spec: v1.ServiceSpec{ + Type: v1.ServiceTypeClusterIP, + IPFamilies: []v1.IPFamily{v1.IPv4Protocol}, + Ports: []v1.ServicePort{ + {Name: "https", Port: 443, TargetPort: intstr.FromInt32(443)}, + }, + }, + } + dualStackService := &v1.Service{ + ObjectMeta: metav1.ObjectMeta{Namespace: "test", Name: "dual-stack"}, + Spec: v1.ServiceSpec{ + Type: v1.ServiceTypeClusterIP, + IPFamilies: []v1.IPFamily{v1.IPv4Protocol, v1.IPv6Protocol}, + Ports: []v1.ServicePort{ + {Name: "https", Port: 443, TargetPort: intstr.FromInt32(443)}, + }, + }, + } + svcPort := singleStackService.Spec.Ports[0] + wrongPort := v1.ServicePort{Name: "http", Port: 80, TargetPort: intstr.FromInt32(443)} + + makeEndpointSlice := func(svc *v1.Service, suffix string, addressType discoveryv1.AddressType, port v1.ServicePort, endpoints ...discoveryv1.Endpoint) *discoveryv1.EndpointSlice { + return &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: svc.Namespace, + Name: svc.Name + "-" + suffix, + Labels: map[string]string{ + discoveryv1.LabelServiceName: svc.Name, + }, + }, + AddressType: addressType, + Endpoints: endpoints, + Ports: []discoveryv1.EndpointPort{{ + Name: &port.Name, + Port: &port.TargetPort.IntVal, + }}, + } + } + + testCases := []struct { + name string + service *v1.Service + endpointSlices []*discoveryv1.EndpointSlice + + expectedURLs []string + }{ + { + name: "simple", + service: singleStackService, + endpointSlices: []*discoveryv1.EndpointSlice{ + makeEndpointSlice(singleStackService, "1", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.1"}, + }, + ), + makeEndpointSlice(singleStackService, "2", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.2"}, + }, + ), + }, + expectedURLs: []string{ + "https://10.0.0.1:443", + "https://10.0.0.2:443", + }, + }, + { + name: "multiple endpoints, some non-ready", + service: singleStackService, + endpointSlices: []*discoveryv1.EndpointSlice{ + makeEndpointSlice(singleStackService, "1", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + // implied Ready + }, + }, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.2"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(false), + }, + }, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.3"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + ), + makeEndpointSlice(singleStackService, "2", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.4"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + ), + makeEndpointSlice(singleStackService, "3", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.5"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(false), + }, + }, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.6"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(false), + }, + }, + ), + }, + expectedURLs: []string{ + "https://10.0.0.1:443", + "https://10.0.0.3:443", + "https://10.0.0.4:443", + }, + }, + { + name: "dual-stack, primary-family endpoints ready", + service: dualStackService, + endpointSlices: []*discoveryv1.EndpointSlice{ + makeEndpointSlice(dualStackService, "v6", + discoveryv1.AddressTypeIPv6, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"fd00::1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + discoveryv1.Endpoint{ + Addresses: []string{"fd00::2"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + ), + makeEndpointSlice(dualStackService, "v4", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.2"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + ), + }, + expectedURLs: []string{ + "https://10.0.0.1:443", + "https://10.0.0.2:443", + }, + }, + { + name: "dual-stack, primary-family endpoints non-ready", + service: dualStackService, + endpointSlices: []*discoveryv1.EndpointSlice{ + makeEndpointSlice(dualStackService, "v4", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(false), + }, + }, + discoveryv1.Endpoint{ + Addresses: []string{"10.0.0.2"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(false), + }, + }, + ), + makeEndpointSlice(dualStackService, "v6", + discoveryv1.AddressTypeIPv6, svcPort, + discoveryv1.Endpoint{ + Addresses: []string{"fd00::1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + discoveryv1.Endpoint{ + Addresses: []string{"fd00::2"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + ), + }, + expectedURLs: []string{ + "https://[fd00::1]:443", + "https://[fd00::2]:443", + }, + }, + { + name: "many slices, many endpoints, most unusable", + service: dualStackService, + endpointSlices: []*discoveryv1.EndpointSlice{ + makeEndpointSlice(dualStackService, "v4-1", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + // Not ready + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(false), + }, + }, + ), + makeEndpointSlice(dualStackService, "v6-1", + discoveryv1.AddressTypeIPv6, svcPort, + discoveryv1.Endpoint{ + // wrong IP family + Addresses: []string{"fd00::1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + discoveryv1.Endpoint{ + // wrong IP family + Addresses: []string{"fd00::2"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + ), + makeEndpointSlice(dualStackService, "v4-2", + discoveryv1.AddressTypeIPv4, svcPort, + // (no endpoints) + ), + makeEndpointSlice(dualStackService, "v4-3", + discoveryv1.AddressTypeIPv4, svcPort, + discoveryv1.Endpoint{ + // This is the good one + Addresses: []string{"10.0.0.2"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + ), + makeEndpointSlice(dualStackService, "v4-4", + discoveryv1.AddressTypeIPv4, wrongPort, + discoveryv1.Endpoint{ + // Uses wrongPort above, so it won't have + // the right port name. + Addresses: []string{"10.0.0.3"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + }, + ), + }, + expectedURLs: []string{ + "https://10.0.0.2:443", + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + serviceCache := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + serviceLister := v1listers.NewServiceLister(serviceCache) + if err := serviceCache.Add(tc.service); err != nil { + t.Fatalf("unexpected service add error: %v", err) + } + + endpointSliceCache := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + endpointSliceLister := discoveryv1listers.NewEndpointSliceLister(endpointSliceCache) + for _, ep := range tc.endpointSlices { + if err := endpointSliceCache.Add(ep); err != nil { + t.Fatalf("unexpected endpointslice add error: %v", err) + } + } + + endpointSliceGetter, err := NewEndpointSliceListerGetter(endpointSliceLister) + if err != nil { + t.Fatal(err) + } + + expectedURLs := sets.New(tc.expectedURLs...) + gotURLs := sets.New[string]() + for i := 0; i < 100; i++ { + endpointURL, err := ResolveEndpoint(serviceLister, endpointSliceGetter, tc.service.Namespace, tc.service.Name, tc.service.Spec.Ports[0].Port) + if err != nil { + t.Fatalf("unexpected error from ResolveEndpoint: %v", err) + } + gotURLs.Insert(endpointURL.String()) + } + + extraURLs := gotURLs.Difference(expectedURLs) + if len(extraURLs) > 0 { + t.Errorf("ResolveEndpoint picked invalid endpoints: %v", sets.List(extraURLs)) + } + missingURLs := expectedURLs.Difference(gotURLs) + if len(missingURLs) > 0 { + t.Errorf("ResolveEndpoint failed to pick some valid endpoints: %v", sets.List(missingURLs)) + } + }) + } +} diff --git a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/apiserver.go b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/apiserver.go index a1fc88d7b7b..e69d45a77ca 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/apiserver.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/apiserver.go @@ -334,7 +334,7 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg remote, err := remoteavailability.New( informerFactory.Apiregistration().V1().APIServices(), c.GenericConfig.SharedInformerFactory.Core().V1().Services(), - c.GenericConfig.SharedInformerFactory.Core().V1().Endpoints(), + c.GenericConfig.SharedInformerFactory.Discovery().V1().EndpointSlices(), apiregistrationClient.ApiregistrationV1(), proxyTransportDial, (func() ([]byte, []byte))(s.proxyCurrentCertKeyContent), diff --git a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/resolvers.go b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/resolvers.go index 74bcb24d987..8c91f9c2c8d 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/resolvers.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/resolvers.go @@ -30,20 +30,20 @@ type ServiceResolver interface { // NewEndpointServiceResolver returns a ServiceResolver that chooses one of the // service's endpoints. -func NewEndpointServiceResolver(services listersv1.ServiceLister, endpoints listersv1.EndpointsLister) ServiceResolver { +func NewEndpointServiceResolver(services listersv1.ServiceLister, endpointSliceGetter proxy.EndpointSliceGetter) ServiceResolver { return &aggregatorEndpointRouting{ - services: services, - endpoints: endpoints, + services: services, + endpointSliceGetter: endpointSliceGetter, } } type aggregatorEndpointRouting struct { - services listersv1.ServiceLister - endpoints listersv1.EndpointsLister + services listersv1.ServiceLister + endpointSliceGetter proxy.EndpointSliceGetter } func (r *aggregatorEndpointRouting) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) { - return proxy.ResolveEndpoint(r.services, r.endpoints, namespace, name, port) + return proxy.ResolveEndpoint(r.services, r.endpointSliceGetter, namespace, name, port) } // NewClusterIPServiceResolver returns a ServiceResolver that directly calls the diff --git a/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go b/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go index a94e254cd8f..b3e9f3ceff7 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go @@ -26,15 +26,16 @@ import ( "time" v1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" - "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/apiserver/pkg/util/proxy" v1informers "k8s.io/client-go/informers/core/v1" + discoveryv1informers "k8s.io/client-go/informers/discovery/v1" v1listers "k8s.io/client-go/listers/core/v1" "k8s.io/client-go/tools/cache" "k8s.io/client-go/transport" @@ -67,8 +68,8 @@ type AvailableConditionController struct { serviceLister v1listers.ServiceLister servicesSynced cache.InformerSynced - endpointsLister v1listers.EndpointsLister - endpointsSynced cache.InformerSynced + endpointSliceGetter proxy.EndpointSliceGetter + endpointSlicesSynced cache.InformerSynced // proxyTransportDial specifies the dial function for creating unencrypted TCP connections. proxyTransportDial *transport.DialHolder @@ -92,19 +93,25 @@ type AvailableConditionController struct { func New( apiServiceInformer informers.APIServiceInformer, serviceInformer v1informers.ServiceInformer, - endpointsInformer v1informers.EndpointsInformer, + endpointSliceInformer discoveryv1informers.EndpointSliceInformer, apiServiceClient apiregistrationclient.APIServicesGetter, proxyTransportDial *transport.DialHolder, proxyCurrentCertKeyContent certKeyFunc, serviceResolver ServiceResolver, metrics *availabilitymetrics.Metrics, ) (*AvailableConditionController, error) { + + endpointSliceGetter, err := proxy.NewEndpointSliceIndexerGetter(endpointSliceInformer) + if err != nil { + return nil, err + } + c := &AvailableConditionController{ - apiServiceClient: apiServiceClient, - apiServiceLister: apiServiceInformer.Lister(), - serviceLister: serviceInformer.Lister(), - endpointsLister: endpointsInformer.Lister(), - serviceResolver: serviceResolver, + apiServiceClient: apiServiceClient, + apiServiceLister: apiServiceInformer.Lister(), + serviceLister: serviceInformer.Lister(), + endpointSliceGetter: endpointSliceGetter, + serviceResolver: serviceResolver, queue: workqueue.NewTypedRateLimitingQueueWithConfig( // We want a fairly tight requeue time. The controller listens to the API, but because it relies on the routability of the // service network, it is possible for an external, non-watchable factor to affect availability. This keeps @@ -137,12 +144,12 @@ func New( }) c.servicesSynced = serviceHandler.HasSynced - endpointsHandler, _ := endpointsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ - AddFunc: c.addEndpoints, - UpdateFunc: c.updateEndpoints, - DeleteFunc: c.deleteEndpoints, + endpointSliceHandler, _ := endpointSliceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: c.addEndpointSlice, + UpdateFunc: c.updateEndpointSlice, + DeleteFunc: c.deleteEndpointSlice, }) - c.endpointsSynced = endpointsHandler.HasSynced + c.endpointSlicesSynced = endpointSliceHandler.HasSynced c.syncFn = c.sync @@ -239,30 +246,37 @@ func (c *AvailableConditionController) sync(key string) error { return err } - endpoints, err := c.endpointsLister.Endpoints(apiService.Spec.Service.Namespace).Get(apiService.Spec.Service.Name) - if apierrors.IsNotFound(err) { - availableCondition.Status = apiregistrationv1.ConditionFalse - availableCondition.Reason = "EndpointsNotFound" - availableCondition.Message = fmt.Sprintf("cannot find endpoints for service/%s in %q", apiService.Spec.Service.Name, apiService.Spec.Service.Namespace) - apiregistrationv1apihelper.SetAPIServiceCondition(apiService, availableCondition) - _, err := c.updateAPIServiceStatus(originalAPIService, apiService) - return err - } else if err != nil { + endpointSlices, err := c.endpointSliceGetter.GetEndpointSlices(apiService.Spec.Service.Namespace, apiService.Spec.Service.Name) + if err != nil { availableCondition.Status = apiregistrationv1.ConditionUnknown availableCondition.Reason = "EndpointsAccessError" availableCondition.Message = fmt.Sprintf("service/%s in %q cannot be checked due to: %v", apiService.Spec.Service.Name, apiService.Spec.Service.Namespace, err) apiregistrationv1apihelper.SetAPIServiceCondition(apiService, availableCondition) _, err := c.updateAPIServiceStatus(originalAPIService, apiService) return err + } else if len(endpointSlices) == 0 { + availableCondition.Status = apiregistrationv1.ConditionFalse + availableCondition.Reason = "EndpointsNotFound" + availableCondition.Message = fmt.Sprintf("cannot find endpointslices for service/%s in %q", apiService.Spec.Service.Name, apiService.Spec.Service.Namespace) + apiregistrationv1apihelper.SetAPIServiceCondition(apiService, availableCondition) + _, err := c.updateAPIServiceStatus(originalAPIService, apiService) + return err } hasActiveEndpoints := false outer: - for _, subset := range endpoints.Subsets { - if len(subset.Addresses) == 0 { + for _, slice := range endpointSlices { + ready := false + for _, endpoint := range slice.Endpoints { + if endpoint.Conditions.Ready == nil || *endpoint.Conditions.Ready { + ready = true + break + } + } + if !ready { continue } - for _, endpointPort := range subset.Ports { - if endpointPort.Name == portName { + for _, endpointPort := range slice.Ports { + if endpointPort.Name != nil && *endpointPort.Name == portName && endpointPort.Port != nil { hasActiveEndpoints = true break outer } @@ -271,7 +285,7 @@ func (c *AvailableConditionController) sync(key string) error { if !hasActiveEndpoints { availableCondition.Status = apiregistrationv1.ConditionFalse availableCondition.Reason = "MissingEndpoints" - availableCondition.Message = fmt.Sprintf("endpoints for service/%s in %q have no addresses with port name %q", apiService.Spec.Service.Name, apiService.Spec.Service.Namespace, portName) + availableCondition.Message = fmt.Sprintf("endpointslices for service/%s in %q have no addresses with port name %q", apiService.Spec.Service.Name, apiService.Spec.Service.Namespace, portName) apiregistrationv1apihelper.SetAPIServiceCondition(apiService, availableCondition) _, err := c.updateAPIServiceStatus(originalAPIService, apiService) return err @@ -415,7 +429,7 @@ func (c *AvailableConditionController) Run(workers int, stopCh <-chan struct{}) // to be called; since the handlers are three different ways of // enqueueing the same thing, waiting for this permits the queue to // maximally de-duplicate the entries. - if !controllers.WaitForCacheSync("RemoteAvailability", stopCh, c.apiServiceSynced, c.servicesSynced, c.endpointsSynced) { + if !controllers.WaitForCacheSync("RemoteAvailability", stopCh, c.apiServiceSynced, c.servicesSynced, c.endpointSlicesSynced) { return } @@ -491,18 +505,13 @@ func (c *AvailableConditionController) deleteAPIService(obj interface{}) { c.queue.Add(castObj.Name) } -func (c *AvailableConditionController) getAPIServicesFor(obj runtime.Object) []string { - metadata, err := meta.Accessor(obj) - if err != nil { - utilruntime.HandleError(err) - return nil - } +func (c *AvailableConditionController) getAPIServicesFor(serviceNamespace, serviceName string) []string { c.cacheLock.RLock() defer c.cacheLock.RUnlock() - return c.cache[metadata.GetNamespace()][metadata.GetName()] + return c.cache[serviceNamespace][serviceName] } -// if the service/endpoint handler wins the race against the cache rebuilding, it may queue a no-longer-relevant apiservice +// if the service/endpointslice handler wins the race against the cache rebuilding, it may queue a no-longer-relevant apiservice // (which will get processed an extra time - this doesn't matter), // and miss a newly relevant apiservice (which will get queued by the apiservice handler) func (c *AvailableConditionController) rebuildAPIServiceCache() { @@ -526,13 +535,15 @@ func (c *AvailableConditionController) rebuildAPIServiceCache() { // TODO, think of a way to avoid checking on every service manipulation func (c *AvailableConditionController) addService(obj interface{}) { - for _, apiService := range c.getAPIServicesFor(obj.(*v1.Service)) { + service := obj.(*v1.Service) + for _, apiService := range c.getAPIServicesFor(service.Namespace, service.Name) { c.queue.Add(apiService) } } func (c *AvailableConditionController) updateService(obj, _ interface{}) { - for _, apiService := range c.getAPIServicesFor(obj.(*v1.Service)) { + service := obj.(*v1.Service) + for _, apiService := range c.getAPIServicesFor(service.Namespace, service.Name) { c.queue.Add(apiService) } } @@ -551,38 +562,52 @@ func (c *AvailableConditionController) deleteService(obj interface{}) { return } } - for _, apiService := range c.getAPIServicesFor(castObj) { + for _, apiService := range c.getAPIServicesFor(castObj.Namespace, castObj.Name) { c.queue.Add(apiService) } } -func (c *AvailableConditionController) addEndpoints(obj interface{}) { - for _, apiService := range c.getAPIServicesFor(obj.(*v1.Endpoints)) { +func (c *AvailableConditionController) addEndpointSlice(obj interface{}) { + slice := obj.(*discoveryv1.EndpointSlice) + serviceName := slice.Labels[discoveryv1.LabelServiceName] + if serviceName == "" { + return + } + for _, apiService := range c.getAPIServicesFor(slice.Namespace, serviceName) { c.queue.Add(apiService) } } -func (c *AvailableConditionController) updateEndpoints(obj, _ interface{}) { - for _, apiService := range c.getAPIServicesFor(obj.(*v1.Endpoints)) { +func (c *AvailableConditionController) updateEndpointSlice(obj, _ interface{}) { + slice := obj.(*discoveryv1.EndpointSlice) + serviceName := slice.Labels[discoveryv1.LabelServiceName] + if serviceName == "" { + return + } + for _, apiService := range c.getAPIServicesFor(slice.Namespace, serviceName) { c.queue.Add(apiService) } } -func (c *AvailableConditionController) deleteEndpoints(obj interface{}) { - castObj, ok := obj.(*v1.Endpoints) +func (c *AvailableConditionController) deleteEndpointSlice(obj interface{}) { + castObj, ok := obj.(*discoveryv1.EndpointSlice) if !ok { tombstone, ok := obj.(cache.DeletedFinalStateUnknown) if !ok { klog.Errorf("Couldn't get object from tombstone %#v", obj) return } - castObj, ok = tombstone.Obj.(*v1.Endpoints) + castObj, ok = tombstone.Obj.(*discoveryv1.EndpointSlice) if !ok { klog.Errorf("Tombstone contained object that is not expected %#v", obj) return } } - for _, apiService := range c.getAPIServicesFor(castObj) { + serviceName := castObj.Labels[discoveryv1.LabelServiceName] + if serviceName == "" { + return + } + for _, apiService := range c.getAPIServicesFor(castObj.Namespace, serviceName) { c.queue.Add(apiService) } } diff --git a/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go b/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go index acfe9ba3952..5de6af66f12 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go @@ -26,10 +26,13 @@ import ( "time" 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" "k8s.io/apimachinery/pkg/util/dump" + "k8s.io/apiserver/pkg/util/proxy" v1listers "k8s.io/client-go/listers/core/v1" + discoveryv1listers "k8s.io/client-go/listers/discovery/v1" clienttesting "k8s.io/client-go/testing" "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/workqueue" @@ -46,33 +49,45 @@ const ( testServicePortName = "testPort" ) -func newEndpoints(namespace, name string) *v1.Endpoints { - return &v1.Endpoints{ - ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, - } -} - -func newEndpointsWithAddress(namespace, name string, port int32, portName string) *v1.Endpoints { - return &v1.Endpoints{ - ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, - Subsets: []v1.EndpointSubset{ - { - Addresses: []v1.EndpointAddress{ - { - IP: "val", - }, - }, - Ports: []v1.EndpointPort{ - { - Name: portName, - Port: port, - }, - }, +func newEndpointSlice(namespace, serviceName string) *discoveryv1.EndpointSlice { + return &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: serviceName + "-xxx", + Labels: map[string]string{ + discoveryv1.LabelServiceName: serviceName, }, }, } } +func newEndpointSliceWithAddress(namespace, serviceName string, port int32, portName string) *discoveryv1.EndpointSlice { + slice := newEndpointSlice(namespace, serviceName) + slice.Endpoints = []discoveryv1.Endpoint{{ + Addresses: []string{"val"}, + }} + slice.Ports = []discoveryv1.EndpointPort{{ + Name: &portName, + Port: &port, + }} + return slice +} + +func newUnreadyEndpointSliceWithAddress(namespace, serviceName string, port int32, portName string) *discoveryv1.EndpointSlice { + slice := newEndpointSlice(namespace, serviceName) + slice.Endpoints = []discoveryv1.Endpoint{{ + Addresses: []string{"val"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(false), + }, + }} + slice.Ports = []discoveryv1.EndpointPort{{ + Name: &portName, + Port: &port, + }} + return slice +} + func newService(namespace, name string, port int32, portName string) *v1.Service { return &v1.Service{ ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, @@ -115,7 +130,12 @@ func setupAPIServices(t T, apiServices []runtime.Object) (*AvailableConditionCon fakeClient := fake.NewSimpleClientset() apiServiceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) serviceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) - endpointsIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + endpointSliceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + + endpointSliceGetter, err := proxy.NewEndpointSliceListerGetter(discoveryv1listers.NewEndpointSliceLister(endpointSliceIndexer)) + if err != nil { + t.Fatalf("error creating endpointSliceGetter: %v", err) + } testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) @@ -129,11 +149,11 @@ func setupAPIServices(t T, apiServices []runtime.Object) (*AvailableConditionCon } c := AvailableConditionController{ - apiServiceClient: fakeClient.ApiregistrationV1(), - apiServiceLister: listers.NewAPIServiceLister(apiServiceIndexer), - serviceLister: v1listers.NewServiceLister(serviceIndexer), - endpointsLister: v1listers.NewEndpointsLister(endpointsIndexer), - serviceResolver: &fakeServiceResolver{url: testServer.URL}, + apiServiceClient: fakeClient.ApiregistrationV1(), + apiServiceLister: listers.NewAPIServiceLister(apiServiceIndexer), + serviceLister: v1listers.NewServiceLister(serviceIndexer), + endpointSliceGetter: endpointSliceGetter, + serviceResolver: &fakeServiceResolver{url: testServer.URL}, queue: workqueue.NewTypedRateLimitingQueueWithConfig( // We want a fairly tight requeue time. The controller listens to the API, but because it relies on the routability of the // service network, it is possible for an external, non-watchable factor to affect availability. This keeps @@ -184,7 +204,6 @@ func TestBuildCache(t *testing.T) { apiServiceName string apiServices []runtime.Object services []*v1.Service - endpoints []*v1.Endpoints expectedAvailability apiregistration.APIServiceCondition }{ @@ -219,7 +238,7 @@ func TestSync(t *testing.T) { apiServiceName string apiServices []runtime.Object services []*v1.Service - endpoints []*v1.Endpoints + endpointSlices []*discoveryv1.EndpointSlice backendStatus int backendLocation string @@ -266,8 +285,8 @@ func TestSync(t *testing.T) { }, }, }}, - endpoints: []*v1.Endpoints{newEndpointsWithAddress("foo", "bar", testServicePort, testServicePortName)}, - backendStatus: http.StatusOK, + endpointSlices: []*discoveryv1.EndpointSlice{newEndpointSliceWithAddress("foo", "bar", testServicePort, testServicePortName)}, + backendStatus: http.StatusOK, expectedAvailability: apiregistration.APIServiceCondition{ Type: apiregistration.Available, Status: apiregistration.ConditionFalse, @@ -285,7 +304,7 @@ func TestSync(t *testing.T) { Type: apiregistration.Available, Status: apiregistration.ConditionFalse, Reason: "EndpointsNotFound", - Message: `cannot find endpoints for service/bar in "foo"`, + Message: `cannot find endpointslices for service/bar in "foo"`, }, }, { @@ -293,13 +312,13 @@ func TestSync(t *testing.T) { apiServiceName: "remote.group", apiServices: []runtime.Object{newRemoteAPIService("remote.group")}, services: []*v1.Service{newService("foo", "bar", testServicePort, testServicePortName)}, - endpoints: []*v1.Endpoints{newEndpoints("foo", "bar")}, + endpointSlices: []*discoveryv1.EndpointSlice{newEndpointSlice("foo", "bar")}, backendStatus: http.StatusOK, expectedAvailability: apiregistration.APIServiceCondition{ Type: apiregistration.Available, Status: apiregistration.ConditionFalse, Reason: "MissingEndpoints", - Message: `endpoints for service/bar in "foo" have no addresses with port name "testPort"`, + Message: `endpointslices for service/bar in "foo" have no addresses with port name "testPort"`, }, }, { @@ -307,13 +326,27 @@ func TestSync(t *testing.T) { apiServiceName: "remote.group", apiServices: []runtime.Object{newRemoteAPIService("remote.group")}, services: []*v1.Service{newService("foo", "bar", testServicePort, testServicePortName)}, - endpoints: []*v1.Endpoints{newEndpointsWithAddress("foo", "bar", testServicePort, "wrongName")}, + endpointSlices: []*discoveryv1.EndpointSlice{newEndpointSliceWithAddress("foo", "bar", testServicePort, "wrongName")}, backendStatus: http.StatusOK, expectedAvailability: apiregistration.APIServiceCondition{ Type: apiregistration.Available, Status: apiregistration.ConditionFalse, Reason: "MissingEndpoints", - Message: fmt.Sprintf(`endpoints for service/bar in "foo" have no addresses with port name "%s"`, testServicePortName), + Message: fmt.Sprintf(`endpointslices for service/bar in "foo" have no addresses with port name "%s"`, testServicePortName), + }, + }, + { + name: "endpoints not ready", + apiServiceName: "remote.group", + apiServices: []runtime.Object{newRemoteAPIService("remote.group")}, + services: []*v1.Service{newService("foo", "bar", testServicePort, testServicePortName)}, + endpointSlices: []*discoveryv1.EndpointSlice{newUnreadyEndpointSliceWithAddress("foo", "bar", testServicePort, testServicePortName)}, + backendStatus: http.StatusOK, + expectedAvailability: apiregistration.APIServiceCondition{ + Type: apiregistration.Available, + Status: apiregistration.ConditionFalse, + Reason: "MissingEndpoints", + Message: fmt.Sprintf(`endpointslices for service/bar in "foo" have no addresses with port name "%s"`, testServicePortName), }, }, { @@ -321,7 +354,7 @@ func TestSync(t *testing.T) { apiServiceName: "remote.group", apiServices: []runtime.Object{newRemoteAPIService("remote.group")}, services: []*v1.Service{newService("foo", "bar", testServicePort, testServicePortName)}, - endpoints: []*v1.Endpoints{newEndpointsWithAddress("foo", "bar", testServicePort, testServicePortName)}, + endpointSlices: []*discoveryv1.EndpointSlice{newEndpointSliceWithAddress("foo", "bar", testServicePort, testServicePortName)}, backendStatus: http.StatusOK, expectedAvailability: apiregistration.APIServiceCondition{ Type: apiregistration.Available, @@ -335,7 +368,7 @@ func TestSync(t *testing.T) { apiServiceName: "remote.group", apiServices: []runtime.Object{newRemoteAPIService("remote.group")}, services: []*v1.Service{newService("foo", "bar", testServicePort, testServicePortName)}, - endpoints: []*v1.Endpoints{newEndpointsWithAddress("foo", "bar", testServicePort, testServicePortName)}, + endpointSlices: []*discoveryv1.EndpointSlice{newEndpointSliceWithAddress("foo", "bar", testServicePort, testServicePortName)}, backendStatus: http.StatusForbidden, expectedAvailability: apiregistration.APIServiceCondition{ Type: apiregistration.Available, @@ -350,7 +383,7 @@ func TestSync(t *testing.T) { apiServiceName: "remote.group", apiServices: []runtime.Object{newRemoteAPIService("remote.group")}, services: []*v1.Service{newService("foo", "bar", testServicePort, testServicePortName)}, - endpoints: []*v1.Endpoints{newEndpointsWithAddress("foo", "bar", testServicePort, testServicePortName)}, + endpointSlices: []*discoveryv1.EndpointSlice{newEndpointSliceWithAddress("foo", "bar", testServicePort, testServicePortName)}, backendStatus: http.StatusFound, backendLocation: "/test", expectedAvailability: apiregistration.APIServiceCondition{ @@ -366,7 +399,7 @@ func TestSync(t *testing.T) { apiServiceName: "remote.group", apiServices: []runtime.Object{newRemoteAPIService("remote.group")}, services: []*v1.Service{newService("foo", "bar", testServicePort, testServicePortName)}, - endpoints: []*v1.Endpoints{newEndpointsWithAddress("foo", "bar", testServicePort, testServicePortName)}, + endpointSlices: []*discoveryv1.EndpointSlice{newEndpointSliceWithAddress("foo", "bar", testServicePort, testServicePortName)}, backendStatus: http.StatusNotModified, expectedAvailability: apiregistration.APIServiceCondition{ Type: apiregistration.Available, @@ -383,15 +416,20 @@ func TestSync(t *testing.T) { fakeClient := fake.NewSimpleClientset(tc.apiServices...) apiServiceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) serviceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) - endpointsIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + endpointSliceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) for _, obj := range tc.apiServices { - apiServiceIndexer.Add(obj) + apiServiceIndexer.Add(obj) //nolint:errcheck } for _, obj := range tc.services { - serviceIndexer.Add(obj) + serviceIndexer.Add(obj) //nolint:errcheck } - for _, obj := range tc.endpoints { - endpointsIndexer.Add(obj) + for _, obj := range tc.endpointSlices { + endpointSliceIndexer.Add(obj) //nolint:errcheck + } + + endpointSliceGetter, err := proxy.NewEndpointSliceListerGetter(discoveryv1listers.NewEndpointSliceLister(endpointSliceIndexer)) + if err != nil { + t.Fatalf("error creating endpointSliceGetter: %v", err) } testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -406,12 +444,12 @@ func TestSync(t *testing.T) { apiServiceClient: fakeClient.ApiregistrationV1(), apiServiceLister: listers.NewAPIServiceLister(apiServiceIndexer), serviceLister: v1listers.NewServiceLister(serviceIndexer), - endpointsLister: v1listers.NewEndpointsLister(endpointsIndexer), + endpointSliceGetter: endpointSliceGetter, serviceResolver: &fakeServiceResolver{url: testServer.URL}, proxyCurrentCertKeyContent: func() ([]byte, []byte) { return emptyCert(), emptyCert() }, metrics: availabilitymetrics.New(), } - err := c.sync(tc.apiServiceName) + err = c.sync(tc.apiServiceName) if tc.expectedSyncError != "" { if err == nil { t.Fatalf("%v expected error with %q, got none", tc.name, tc.expectedSyncError) From 33b45c8383ee8273612e0ace371f85e58dab6b9d Mon Sep 17 00:00:00 2001 From: Dan Winship Date: Mon, 12 May 2025 16:22:56 -0400 Subject: [PATCH 3/3] Update "should proxy through a service and a pod" to look at EndpointSlices --- test/e2e/network/proxy.go | 24 ++---------------------- 1 file changed, 2 insertions(+), 22 deletions(-) diff --git a/test/e2e/network/proxy.go b/test/e2e/network/proxy.go index 41813427664..518679ec039 100644 --- a/test/e2e/network/proxy.go +++ b/test/e2e/network/proxy.go @@ -36,12 +36,12 @@ import ( "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/apimachinery/pkg/util/net" "k8s.io/apimachinery/pkg/util/wait" - clientset "k8s.io/client-go/kubernetes" "k8s.io/client-go/transport" "k8s.io/kubernetes/test/e2e/framework" e2edeployment "k8s.io/kubernetes/test/e2e/framework/deployment" e2enode "k8s.io/kubernetes/test/e2e/framework/node" e2epod "k8s.io/kubernetes/test/e2e/framework/pod" + e2eservice "k8s.io/kubernetes/test/e2e/framework/service" "k8s.io/kubernetes/test/e2e/network/common" imageutils "k8s.io/kubernetes/test/utils/image" admissionapi "k8s.io/pod-security-admission/api" @@ -223,7 +223,7 @@ var _ = common.SIGDescribe("Proxy", func() { framework.ExpectNoError(err) pods := podList.Items - err = waitForEndpoint(ctx, f.ClientSet, f.Namespace.Name, service.Name) + err = framework.WaitForServiceEndpointsNum(ctx, f.ClientSet, f.Namespace.Name, service.Name, 1, time.Second, e2eservice.ServiceEndpointsTimeout) framework.ExpectNoError(err) // table constructors @@ -640,23 +640,3 @@ func nodeProxyTest(ctx context.Context, f *framework.Framework, prefix, nodeDest maxFailures := int(math.Floor(0.1 * float64(proxyAttempts))) gomega.Expect(serviceUnavailableErrors).To(gomega.BeNumerically("<", maxFailures)) } - -// waitForEndpoint waits for the specified endpoint to be ready. -func waitForEndpoint(ctx context.Context, c clientset.Interface, ns, name string) error { - // registerTimeout is how long to wait for an endpoint to be registered. - registerTimeout := time.Minute - for t := time.Now(); time.Since(t) < registerTimeout; time.Sleep(framework.Poll) { - endpoint, err := c.CoreV1().Endpoints(ns).Get(ctx, name, metav1.GetOptions{}) - if apierrors.IsNotFound(err) { - framework.Logf("Endpoint %s/%s is not ready yet", ns, name) - continue - } - framework.ExpectNoError(err, "Failed to get endpoints for %s/%s", ns, name) - if len(endpoint.Subsets) == 0 || len(endpoint.Subsets[0].Addresses) == 0 { - framework.Logf("Endpoint %s/%s is not ready yet", ns, name) - continue - } - return nil - } - return fmt.Errorf("failed to get endpoints for %s/%s", ns, name) -}