github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-08-08 15:25:26 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #129837 from danwinship/aggregated-apiserver-endpointslices

Browse Source 
Port aggregated apiserver discovery to EndpointSlices
This commit is contained in:
Kubernetes Prow Robot
2025-07-23 10:30:28 -07:00
committed by GitHub
parent 7383971ab3 33b45c8383
commit c41cc0a144
10 changed files with 911 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
cmd/kube-apiserver/app/server.go
View File

@@ -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"
@@ -216,7 +218,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
@@ -275,16 +280,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(
@@ -296,5 +306,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
}

114
staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice.go Normal file
View File

@@ -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}))
}

200
staging/src/k8s.io/apiserver/pkg/util/proxy/endpointslice_test.go Normal file
View File

@@ -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)
}

66
staging/src/k8s.io/apiserver/pkg/util/proxy/proxy.go
View File

@@ -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
}
}
}
}

429
staging/src/k8s.io/apiserver/pkg/util/proxy/proxy_test.go
View File

@@ -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))
}
})
}
}

2
staging/src/k8s.io/kube-aggregator/pkg/apiserver/apiserver.go
View File

@@ -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),

12
staging/src/k8s.io/kube-aggregator/pkg/apiserver/resolvers.go
View File

@@ -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

123
staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go
View File

@@ -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)
}
}

134
staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go
View File

@@ -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)

24
test/e2e/network/proxy.go
View File

@@ -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)
}