github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Graduate EndpointSlice Controllers to GA

Browse Source 
- EndpointSlice controller will stop writing to Topology field
 - EndpointSlice controller will only provide NodeName and Zone on
 EndpointSlices
This commit is contained in:
Swetha Repakula 2021-03-04 20:53:35 -08:00
parent f8151b121f
commit 1925d94b18
26 changed files with 170 additions and 298 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
cmd/kube-controller-manager/app/discovery.go
View File

@ -23,7 +23,7 @@ package app
import (
"net/http"
discoveryv1beta1 "k8s.io/api/discovery/v1beta1"
discoveryv1 "k8s.io/api/discovery/v1"
utilfeature "k8s.io/apiserver/pkg/util/feature"
"k8s.io/klog/v2"
endpointslicecontroller "k8s.io/kubernetes/pkg/controller/endpointslice"
@ -37,8 +37,8 @@ func startEndpointSliceController(ctx ControllerContext) (http.Handler, bool, er
return nil, false, nil
}
if !ctx.AvailableResources[discoveryv1beta1.SchemeGroupVersion.WithResource("endpointslices")] {
klog.Warningf("Not starting endpointslice-controller since discovery.k8s.io/v1beta1 resources are not available")
if !ctx.AvailableResources[discoveryv1.SchemeGroupVersion.WithResource("endpointslices")] {
klog.Warningf("Not starting endpointslice-controller since discovery.k8s.io/v1 resources are not available")
return nil, false, nil
}
@ -46,7 +46,7 @@ func startEndpointSliceController(ctx ControllerContext) (http.Handler, bool, er
ctx.InformerFactory.Core().V1().Pods(),
ctx.InformerFactory.Core().V1().Services(),
ctx.InformerFactory.Core().V1().Nodes(),
ctx.InformerFactory.Discovery().V1beta1().EndpointSlices(),
ctx.InformerFactory.Discovery().V1().EndpointSlices(),
ctx.ComponentConfig.EndpointSliceController.MaxEndpointsPerSlice,
ctx.ClientBuilder.ClientOrDie("endpointslice-controller"),
ctx.ComponentConfig.EndpointSliceController.EndpointUpdatesBatchPeriod.Duration,
@ -60,14 +60,14 @@ func startEndpointSliceMirroringController(ctx ControllerContext) (http.Handler,
return nil, false, nil
}
if !ctx.AvailableResources[discoveryv1beta1.SchemeGroupVersion.WithResource("endpointslices")] {
klog.Warningf("Not starting endpointslicemirroring-controller since discovery.k8s.io/v1beta1 resources are not available")
if !ctx.AvailableResources[discoveryv1.SchemeGroupVersion.WithResource("endpointslices")] {
klog.Warningf("Not starting endpointslicemirroring-controller since discovery.k8s.io/v1 resources are not available")
return nil, false, nil
}
go endpointslicemirroringcontroller.NewController(
ctx.InformerFactory.Core().V1().Endpoints(),
ctx.InformerFactory.Discovery().V1beta1().EndpointSlices(),
ctx.InformerFactory.Discovery().V1().EndpointSlices(),
ctx.InformerFactory.Core().V1().Services(),
ctx.ComponentConfig.EndpointSliceMirroringController.MirroringMaxEndpointsPerSubset,
ctx.ClientBuilder.ClientOrDie("endpointslicemirroring-controller"),

2
pkg/controller/endpointslice/endpointset.go
View File

@ -19,7 +19,7 @@ package endpointslice
import (
"sort"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
endpointutil "k8s.io/kubernetes/pkg/controller/util/endpoint"
)

8
pkg/controller/endpointslice/endpointslice_controller.go
View File

@ -23,18 +23,18 @@ import (
"golang.org/x/time/rate"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/labels"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
coreinformers "k8s.io/client-go/informers/core/v1"
discoveryinformers "k8s.io/client-go/informers/discovery/v1beta1"
discoveryinformers "k8s.io/client-go/informers/discovery/v1"
clientset "k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/scheme"
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
corelisters "k8s.io/client-go/listers/core/v1"
discoverylisters "k8s.io/client-go/listers/discovery/v1beta1"
discoverylisters "k8s.io/client-go/listers/discovery/v1"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/record"
"k8s.io/client-go/util/workqueue"
@ -87,7 +87,7 @@ func NewController(podInformer coreinformers.PodInformer,
recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "endpoint-slice-controller"})
if client != nil && client.CoreV1().RESTClient().GetRateLimiter() != nil {
ratelimiter.RegisterMetricAndTrackRateLimiterUsage("endpoint_slice_controller", client.DiscoveryV1beta1().RESTClient().GetRateLimiter())
ratelimiter.RegisterMetricAndTrackRateLimiterUsage("endpoint_slice_controller", client.DiscoveryV1().RESTClient().GetRateLimiter())
}
endpointslicemetrics.RegisterMetrics()

25
pkg/controller/endpointslice/endpointslice_controller_test.go
View File

@ -26,7 +26,7 @@ import (
"github.com/stretchr/testify/assert"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
@ -71,7 +71,7 @@ func newController(nodeNames []string, batchPeriod time.Duration) (*fake.Clients
informerFactory.Core().V1().Pods(),
informerFactory.Core().V1().Services(),
nodeInformer,
informerFactory.Discovery().V1beta1().EndpointSlices(),
informerFactory.Discovery().V1().EndpointSlices(),
int32(100),
client,
batchPeriod)
@ -83,7 +83,7 @@ func newController(nodeNames []string, batchPeriod time.Duration) (*fake.Clients
return client, &endpointSliceController{
esController,
informerFactory.Discovery().V1beta1().EndpointSlices().Informer().GetStore(),
informerFactory.Discovery().V1().EndpointSlices().Informer().GetStore(),
informerFactory.Core().V1().Nodes().Informer().GetStore(),
informerFactory.Core().V1().Pods().Informer().GetStore(),
informerFactory.Core().V1().Services().Informer().GetStore(),
@ -134,7 +134,7 @@ func TestSyncServiceWithSelector(t *testing.T) {
standardSyncService(t, esController, ns, serviceName)
expectActions(t, client.Actions(), 1, "create", "endpointslices")
sliceList, err := client.DiscoveryV1beta1().EndpointSlices(ns).List(context.TODO(), metav1.ListOptions{})
sliceList, err := client.DiscoveryV1().EndpointSlices(ns).List(context.TODO(), metav1.ListOptions{})
assert.Nil(t, err, "Expected no error fetching endpoint slices")
assert.Len(t, sliceList.Items, 1, "Expected 1 endpoint slices")
slice := sliceList.Items[0]
@ -201,7 +201,7 @@ func TestSyncServicePodSelection(t *testing.T) {
expectActions(t, client.Actions(), 1, "create", "endpointslices")
// an endpoint slice should be created, it should only reference pod1 (not pod2)
slices, err := client.DiscoveryV1beta1().EndpointSlices(ns).List(context.TODO(), metav1.ListOptions{})
slices, err := client.DiscoveryV1().EndpointSlices(ns).List(context.TODO(), metav1.ListOptions{})
assert.Nil(t, err, "Expected no error fetching endpoint slices")
assert.Len(t, slices.Items, 1, "Expected 1 endpoint slices")
slice := slices.Items[0]
@ -283,7 +283,7 @@ func TestSyncServiceEndpointSliceLabelSelection(t *testing.T) {
if err != nil {
t.Fatalf("Expected no error adding EndpointSlice: %v", err)
}
_, err = client.DiscoveryV1beta1().EndpointSlices(ns).Create(context.TODO(), endpointSlice, metav1.CreateOptions{})
_, err = client.DiscoveryV1().EndpointSlices(ns).Create(context.TODO(), endpointSlice, metav1.CreateOptions{})
if err != nil {
t.Fatalf("Expected no error creating EndpointSlice: %v", err)
}
@ -451,7 +451,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"10.0.0.1"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod0"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
{
@ -460,7 +459,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"10.0.0.2"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod1"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
},
@ -566,7 +564,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"fd08::5678:0000:0000:9abc:def0"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod1"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
},
@ -672,7 +669,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"10.0.0.1"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod0"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
{
@ -683,7 +679,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"10.0.0.2"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod1"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
},
@ -788,7 +783,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"10.0.0.1"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod0"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
},
@ -895,7 +889,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"10.0.0.1"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod0"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
{
@ -906,7 +899,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"10.0.0.2"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod1"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
},
@ -1011,7 +1003,6 @@ func TestSyncService(t *testing.T) {
},
Addresses: []string{"10.0.0.1"},
TargetRef: &v1.ObjectReference{Kind: "Pod", Namespace: "default", Name: "pod0"},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
},
},
@ -1038,14 +1029,14 @@ func TestSyncService(t *testing.T) {
// last action should be to create endpoint slice
expectActions(t, client.Actions(), 1, "create", "endpointslices")
sliceList, err := client.DiscoveryV1beta1().EndpointSlices(testcase.service.Namespace).List(context.TODO(), metav1.ListOptions{})
sliceList, err := client.DiscoveryV1().EndpointSlices(testcase.service.Namespace).List(context.TODO(), metav1.ListOptions{})
assert.Nil(t, err, "Expected no error fetching endpoint slices")
assert.Len(t, sliceList.Items, 1, "Expected 1 endpoint slices")
// ensure all attributes of endpoint slice match expected state
slice := sliceList.Items[0]
assert.Equal(t, slice.Annotations["endpoints.kubernetes.io/last-change-trigger-time"], creationTimestamp.Format(time.RFC3339Nano))
assert.EqualValues(t, testcase.expectedEndpointPorts, slice.Ports)
assert.ElementsMatch(t, testcase.expectedEndpointPorts, slice.Ports)
assert.ElementsMatch(t, testcase.expectedEndpoints, slice.Endpoints)
})
}

4
pkg/controller/endpointslice/endpointslice_tracker.go
View File

@ -19,8 +19,8 @@ package endpointslice
import (
"sync"
"k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1"
"k8s.io/apimachinery/pkg/types"
)

4
pkg/controller/endpointslice/endpointslice_tracker_test.go
View File

@ -19,8 +19,8 @@ package endpointslice
import (
"testing"
"k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
)

2
pkg/controller/endpointslice/metrics/cache_test.go
View File

@ -19,7 +19,7 @@ package metrics
import (
"testing"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
"k8s.io/apimachinery/pkg/types"
endpointutil "k8s.io/kubernetes/pkg/controller/util/endpoint"
)

12
pkg/controller/endpointslice/reconciler.go
View File

@ -23,7 +23,7 @@ import (
"time"
corev1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
@ -97,7 +97,7 @@ func (r *reconciler) reconcile(service *corev1.Service, pods []*corev1.Pod, exis
// delete those which are of addressType that is no longer supported
// by the service
for _, sliceToDelete := range slicesToDelete {
err := r.client.DiscoveryV1beta1().EndpointSlices(service.Namespace).Delete(context.TODO(), sliceToDelete.Name, metav1.DeleteOptions{})
err := r.client.DiscoveryV1().EndpointSlices(service.Namespace).Delete(context.TODO(), sliceToDelete.Name, metav1.DeleteOptions{})
if err != nil {
errs = append(errs, fmt.Errorf("Error deleting %s EndpointSlice for Service %s/%s: %v", sliceToDelete.Name, service.Namespace, service.Name, err))
} else {
@ -265,7 +265,7 @@ func (r *reconciler) finalize(
if service.DeletionTimestamp == nil {
for _, endpointSlice := range slicesToCreate {
addTriggerTimeAnnotation(endpointSlice, triggerTime)
createdSlice, err := r.client.DiscoveryV1beta1().EndpointSlices(service.Namespace).Create(context.TODO(), endpointSlice, metav1.CreateOptions{})
createdSlice, err := r.client.DiscoveryV1().EndpointSlices(service.Namespace).Create(context.TODO(), endpointSlice, metav1.CreateOptions{})
if err != nil {
// If the namespace is terminating, creates will continue to fail. Simply drop the item.
if errors.HasStatusCause(err, corev1.NamespaceTerminatingCause) {
@ -280,7 +280,7 @@ func (r *reconciler) finalize(
for _, endpointSlice := range slicesToUpdate {
addTriggerTimeAnnotation(endpointSlice, triggerTime)
updatedSlice, err := r.client.DiscoveryV1beta1().EndpointSlices(service.Namespace).Update(context.TODO(), endpointSlice, metav1.UpdateOptions{})
updatedSlice, err := r.client.DiscoveryV1().EndpointSlices(service.Namespace).Update(context.TODO(), endpointSlice, metav1.UpdateOptions{})
if err != nil {
return fmt.Errorf("failed to update %s EndpointSlice for Service %s/%s: %v", endpointSlice.Name, service.Namespace, service.Name, err)
}
@ -289,7 +289,7 @@ func (r *reconciler) finalize(
}
for _, endpointSlice := range slicesToDelete {
err := r.client.DiscoveryV1beta1().EndpointSlices(service.Namespace).Delete(context.TODO(), endpointSlice.Name, metav1.DeleteOptions{})
err := r.client.DiscoveryV1().EndpointSlices(service.Namespace).Delete(context.TODO(), endpointSlice.Name, metav1.DeleteOptions{})
if err != nil {
return fmt.Errorf("failed to delete %s EndpointSlice for Service %s/%s: %v", endpointSlice.Name, service.Namespace, service.Name, err)
}
@ -336,7 +336,7 @@ func (r *reconciler) reconcileByPortMapping(
newEndpoints = append(newEndpoints, *got)
// If existing version of endpoint doesn't match desired version
// set endpointUpdated to ensure endpoint changes are persisted.
if !endpointsEqualBeyondHash(got, &endpoint) {
if !endpointutil.EndpointsEqualBeyondHash(got, &endpoint) {
endpointUpdated = true
}
// once an endpoint has been placed/found in a slice, it no

114
pkg/controller/endpointslice/reconciler_test.go
View File

@ -27,7 +27,7 @@ import (
"github.com/stretchr/testify/assert"
corev1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
apiequality "k8s.io/apimachinery/pkg/api/equality"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
@ -126,12 +126,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -152,12 +148,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -183,11 +175,7 @@ func TestReconcile1Pod(t *testing.T) {
Serving: utilpointer.BoolPtr(true),
Terminating: utilpointer.BoolPtr(false),
},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
@ -211,12 +199,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -229,12 +213,8 @@ func TestReconcile1Pod(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -253,12 +233,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -271,12 +247,8 @@ func TestReconcile1Pod(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -297,12 +269,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -315,12 +283,8 @@ func TestReconcile1Pod(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -341,12 +305,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1234::5678:0000:0000:9abc:def0"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -369,12 +329,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1234::5678:0000:0000:9abc:def0"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -396,12 +352,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1234::5678:0000:0000:9abc:def0"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -413,12 +365,8 @@ func TestReconcile1Pod(t *testing.T) {
{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &corev1.ObjectReference{
Kind: "Pod",
Namespace: namespace,
@ -512,7 +460,7 @@ func TestReconcile1EndpointSlice(t *testing.T) {
svc, endpointMeta := newServiceAndEndpointMeta("foo", namespace)
endpointSlice1 := newEmptyEndpointSlice(1, namespace, endpointMeta, svc)
_, createErr := client.DiscoveryV1beta1().EndpointSlices(namespace).Create(context.TODO(), endpointSlice1, metav1.CreateOptions{})
_, createErr := client.DiscoveryV1().EndpointSlices(namespace).Create(context.TODO(), endpointSlice1, metav1.CreateOptions{})
assert.Nil(t, createErr, "Expected no error creating endpoint slice")
numActionsBefore := len(client.Actions())
@ -1281,7 +1229,7 @@ func TestReconcilerFinalizeSvcDeletionTimestamp(t *testing.T) {
}
// Add EndpointSlice that can be updated.
esToUpdate, err := client.DiscoveryV1beta1().EndpointSlices(namespace).Create(context.TODO(), &discovery.EndpointSlice{
esToUpdate, err := client.DiscoveryV1().EndpointSlices(namespace).Create(context.TODO(), &discovery.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{
Name: "to-update",
OwnerReferences: []metav1.OwnerReference{*ownerRef},
@ -1297,7 +1245,7 @@ func TestReconcilerFinalizeSvcDeletionTimestamp(t *testing.T) {
esToUpdate.Endpoints = []discovery.Endpoint{{Addresses: []string{"10.2.3.4"}}}
// Add EndpointSlice that can be deleted.
esToDelete, err := client.DiscoveryV1beta1().EndpointSlices(namespace).Create(context.TODO(), &discovery.EndpointSlice{
esToDelete, err := client.DiscoveryV1().EndpointSlices(namespace).Create(context.TODO(), &discovery.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{
Name: "to-delete",
OwnerReferences: []metav1.OwnerReference{*ownerRef},
@ -1470,7 +1418,7 @@ func portsAndAddressTypeEqual(slice1, slice2 discovery.EndpointSlice) bool {
func createEndpointSlices(t *testing.T, client *fake.Clientset, namespace string, endpointSlices []*discovery.EndpointSlice) {
t.Helper()
for _, endpointSlice := range endpointSlices {
_, err := client.DiscoveryV1beta1().EndpointSlices(namespace).Create(context.TODO(), endpointSlice, metav1.CreateOptions{})
_, err := client.DiscoveryV1().EndpointSlices(namespace).Create(context.TODO(), endpointSlice, metav1.CreateOptions{})
if err != nil {
t.Fatalf("Expected no error creating Endpoint Slice, got: %v", err)
}
@ -1479,7 +1427,7 @@ func createEndpointSlices(t *testing.T, client *fake.Clientset, namespace string
func fetchEndpointSlices(t *testing.T, client *fake.Clientset, namespace string) []discovery.EndpointSlice {
t.Helper()
fetchedSlices, err := client.DiscoveryV1beta1().EndpointSlices(namespace).List(context.TODO(), metav1.ListOptions{})
fetchedSlices, err := client.DiscoveryV1().EndpointSlices(namespace).List(context.TODO(), metav1.ListOptions{})
if err != nil {
t.Fatalf("Expected no error fetching Endpoint Slices, got: %v", err)
return []discovery.EndpointSlice{}

69
pkg/controller/endpointslice/utils.go
View File

@ -22,7 +22,7 @@ import (
corev1 "k8s.io/api/core/v1"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
apiequality "k8s.io/apimachinery/pkg/api/equality"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
@ -41,23 +41,6 @@ import (
// podToEndpoint returns an Endpoint object generated from a Pod, a Node, and a Service for a particular addressType.
func podToEndpoint(pod *corev1.Pod, node *corev1.Node, service *corev1.Service, addressType discovery.AddressType) discovery.Endpoint {
// Build out topology information. This is currently limited to hostname,
// zone, and region, but this will be expanded in the future.
topology := map[string]string{}
if node != nil {
topologyLabels := []string{
"topology.kubernetes.io/zone",
"topology.kubernetes.io/region",
}
for _, topologyLabel := range topologyLabels {
if node.Labels[topologyLabel] != "" {
topology[topologyLabel] = node.Labels[topologyLabel]
}
}
}
serving := podutil.IsPodReady(pod)
terminating := pod.DeletionTimestamp != nil
// For compatibility reasons, "ready" should never be "true" if a pod is terminatng, unless
@ -83,12 +66,14 @@ func podToEndpoint(pod *corev1.Pod, node *corev1.Node, service *corev1.Service,
}
if pod.Spec.NodeName != "" {
topology["kubernetes.io/hostname"] = pod.Spec.NodeName
ep.Topology = topology
ep.NodeName = &pod.Spec.NodeName
}
if node != nil && node.Labels[corev1.LabelTopologyZone] != "" {
zone := node.Labels[corev1.LabelTopologyZone]
ep.Zone = &zone
}
if endpointutil.ShouldSetHostname(pod, service) {
ep.Hostname = &pod.Spec.Hostname
}
@ -147,25 +132,6 @@ func getEndpointAddresses(podStatus corev1.PodStatus, service *corev1.Service, a
return addresses
}
// endpointsEqualBeyondHash returns true if endpoints have equal attributes
// but excludes equality checks that would have already been covered with
// endpoint hashing (see hashEndpoint func for more info).
func endpointsEqualBeyondHash(ep1, ep2 *discovery.Endpoint) bool {
if !apiequality.Semantic.DeepEqual(ep1.Topology, ep2.Topology) {
return false
}
if boolPtrChanged(ep1.Conditions.Ready, ep2.Conditions.Ready) {
return false
}
if objectRefPtrChanged(ep1.TargetRef, ep2.TargetRef) {
return false
}
return true
}
// newEndpointSlice returns an EndpointSlice generated from a service and
// endpointMeta.
func newEndpointSlice(service *corev1.Service, endpointMeta *endpointMeta) *discovery.EndpointSlice {
@ -198,29 +164,6 @@ func getEndpointSlicePrefix(serviceName string) string {
return prefix
}
// boolPtrChanged returns true if a set of bool pointers have different values.
func boolPtrChanged(ptr1, ptr2 *bool) bool {
if (ptr1 == nil) != (ptr2 == nil) {
return true
}
if ptr1 != nil && ptr2 != nil && *ptr1 != *ptr2 {
return true
}
return false
}
// objectRefPtrChanged returns true if a set of object ref pointers have
// different values.
func objectRefPtrChanged(ref1, ref2 *corev1.ObjectReference) bool {
if (ref1 == nil) != (ref2 == nil) {
return true
}
if ref1 != nil && ref2 != nil && !apiequality.Semantic.DeepEqual(*ref1, *ref2) {
return true
}
return false
}
// ownedBy returns true if the provided EndpointSlice is owned by the provided
// Service.
func ownedBy(endpointSlice *discovery.EndpointSlice, svc *corev1.Service) bool {

35
pkg/controller/endpointslice/utils_test.go
View File

@ -24,7 +24,7 @@ import (
"github.com/stretchr/testify/assert"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
@ -260,7 +260,6 @@ func TestPodToEndpoint(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.5"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@ -278,7 +277,6 @@ func TestPodToEndpoint(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.5"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@ -296,7 +294,6 @@ func TestPodToEndpoint(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.5"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(false)},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@ -314,7 +311,6 @@ func TestPodToEndpoint(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.5"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@ -333,12 +329,8 @@ func TestPodToEndpoint(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.5"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
Namespace: ns,
@ -356,12 +348,8 @@ func TestPodToEndpoint(t *testing.T) {
expectedEndpoint: discovery.Endpoint{
Addresses: []string{"1.2.3.4"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
Namespace: ns,
@ -380,12 +368,8 @@ func TestPodToEndpoint(t *testing.T) {
Addresses: []string{"1.2.3.5"},
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
Hostname: &readyPodHostname.Spec.Hostname,
Topology: map[string]string{
"kubernetes.io/hostname": "node-1",
"topology.kubernetes.io/zone": "us-central1-a",
"topology.kubernetes.io/region": "us-central1",
},
NodeName: utilpointer.StringPtr("node-1"),
Zone: utilpointer.StringPtr("us-central1-a"),
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
Namespace: ns,
@ -406,7 +390,6 @@ func TestPodToEndpoint(t *testing.T) {
Serving: utilpointer.BoolPtr(true),
Terminating: utilpointer.BoolPtr(false),
},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@ -427,7 +410,6 @@ func TestPodToEndpoint(t *testing.T) {
Conditions: discovery.EndpointConditions{
Ready: utilpointer.BoolPtr(false),
},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@ -450,7 +432,6 @@ func TestPodToEndpoint(t *testing.T) {
Serving: utilpointer.BoolPtr(true),
Terminating: utilpointer.BoolPtr(true),
},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@ -471,7 +452,6 @@ func TestPodToEndpoint(t *testing.T) {
Conditions: discovery.EndpointConditions{
Ready: utilpointer.BoolPtr(false),
},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",
@ -494,7 +474,6 @@ func TestPodToEndpoint(t *testing.T) {
Serving: utilpointer.BoolPtr(false),
Terminating: utilpointer.BoolPtr(true),
},
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
TargetRef: &v1.ObjectReference{
Kind: "Pod",

2
pkg/controller/endpointslicemirroring/endpointset.go
View File

@ -19,7 +19,7 @@ package endpointslicemirroring
import (
"sort"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
endpointutil "k8s.io/kubernetes/pkg/controller/util/endpoint"
)

4
pkg/controller/endpointslicemirroring/endpointslice_tracker.go
View File

@ -19,8 +19,8 @@ package endpointslicemirroring
import (
"sync"
"k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1"
"k8s.io/apimachinery/pkg/types"
)

4
pkg/controller/endpointslicemirroring/endpointslice_tracker_test.go
View File

@ -19,8 +19,8 @@ package endpointslicemirroring
import (
"testing"
"k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
)

8
pkg/controller/endpointslicemirroring/endpointslicemirroring_controller.go
View File

@ -23,18 +23,18 @@ import (
"golang.org/x/time/rate"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/labels"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
coreinformers "k8s.io/client-go/informers/core/v1"
discoveryinformers "k8s.io/client-go/informers/discovery/v1beta1"
discoveryinformers "k8s.io/client-go/informers/discovery/v1"
clientset "k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/scheme"
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
corelisters "k8s.io/client-go/listers/core/v1"
discoverylisters "k8s.io/client-go/listers/discovery/v1beta1"
discoverylisters "k8s.io/client-go/listers/discovery/v1"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/record"
"k8s.io/client-go/util/workqueue"
@ -79,7 +79,7 @@ func NewController(endpointsInformer coreinformers.EndpointsInformer,
recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "endpoint-slice-mirroring-controller"})
if client != nil && client.CoreV1().RESTClient().GetRateLimiter() != nil {
ratelimiter.RegisterMetricAndTrackRateLimiterUsage("endpoint_slice_mirroring_controller", client.DiscoveryV1beta1().RESTClient().GetRateLimiter())
ratelimiter.RegisterMetricAndTrackRateLimiterUsage("endpoint_slice_mirroring_controller", client.DiscoveryV1().RESTClient().GetRateLimiter())
}
metrics.RegisterMetrics()

8
pkg/controller/endpointslicemirroring/endpointslicemirroring_controller_test.go
View File

@ -23,7 +23,7 @@ import (
"time"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/informers"
@ -52,7 +52,7 @@ func newController(batchPeriod time.Duration) (*fake.Clientset, *endpointSliceMi
esController := NewController(
informerFactory.Core().V1().Endpoints(),
informerFactory.Discovery().V1beta1().EndpointSlices(),
informerFactory.Discovery().V1().EndpointSlices(),
informerFactory.Core().V1().Services(),
int32(1000),
client,
@ -65,7 +65,7 @@ func newController(batchPeriod time.Duration) (*fake.Clientset, *endpointSliceMi
return client, &endpointSliceMirroringController{
esController,
informerFactory.Core().V1().Endpoints().Informer().GetStore(),
informerFactory.Discovery().V1beta1().EndpointSlices().Informer().GetStore(),
informerFactory.Discovery().V1().EndpointSlices().Informer().GetStore(),
informerFactory.Core().V1().Services().Informer().GetStore(),
}
}
@ -229,7 +229,7 @@ func TestSyncEndpoints(t *testing.T) {
for _, epSlice := range tc.endpointSlices {
epSlice.Namespace = namespace
esController.endpointSliceStore.Add(epSlice)
_, err := client.DiscoveryV1beta1().EndpointSlices(namespace).Create(context.TODO(), epSlice, metav1.CreateOptions{})
_, err := client.DiscoveryV1().EndpointSlices(namespace).Create(context.TODO(), epSlice, metav1.CreateOptions{})
if err != nil {
t.Fatalf("Expected no error creating EndpointSlice, got %v", err)
}

2
pkg/controller/endpointslicemirroring/metrics/cache_test.go
View File

@ -19,7 +19,7 @@ package metrics
import (
"testing"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
"k8s.io/apimachinery/pkg/types"
endpointutil "k8s.io/kubernetes/pkg/controller/util/endpoint"
)

8
pkg/controller/endpointslicemirroring/reconciler.go
View File

@ -21,7 +21,7 @@ import (
"fmt"
corev1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
@ -230,7 +230,7 @@ func (r *reconciler) finalize(endpoints *corev1.Endpoints, slices slicesByAction
// be deleted.
recycleSlices(&slices)
epsClient := r.client.DiscoveryV1beta1().EndpointSlices(endpoints.Namespace)
epsClient := r.client.DiscoveryV1().EndpointSlices(endpoints.Namespace)
// Don't create more EndpointSlices if corresponding Endpoints resource is
// being deleted.
@ -276,7 +276,7 @@ func (r *reconciler) deleteEndpoints(namespace, name string, endpointSlices []*d
r.metricsCache.DeleteEndpoints(types.NamespacedName{Namespace: namespace, Name: name})
var errs []error
for _, endpointSlice := range endpointSlices {
err := r.client.DiscoveryV1beta1().EndpointSlices(namespace).Delete(context.TODO(), endpointSlice.Name, metav1.DeleteOptions{})
err := r.client.DiscoveryV1().EndpointSlices(namespace).Delete(context.TODO(), endpointSlice.Name, metav1.DeleteOptions{})
if err != nil {
errs = append(errs, err)
}
@ -314,7 +314,7 @@ func totalChanges(existingSlice *discovery.EndpointSlice, desiredSet endpointSet
// If existing version of endpoint doesn't match desired version
// increment number of endpoints to be updated.
if !endpointsEqualBeyondHash(got, &endpoint) {
if !endpointutil.EndpointsEqualBeyondHash(got, &endpoint) {
totals.updated++
}
}

4
pkg/controller/endpointslicemirroring/reconciler_helpers.go
View File

@ -17,8 +17,8 @@ limitations under the License.
package endpointslicemirroring
import (
"k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1"
)
// slicesByAction includes lists of slices to create, update, or delete.

2
pkg/controller/endpointslicemirroring/reconciler_helpers_test.go
View File

@ -20,7 +20,7 @@ import (
"sort"
"testing"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

20
pkg/controller/endpointslicemirroring/reconciler_test.go
View File

@ -22,7 +22,7 @@ import (
"testing"
corev1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/kubernetes/scheme"
@ -375,12 +375,12 @@ func TestReconcile(t *testing.T) {
Endpoints: []discovery.Endpoint{{
Addresses: []string{"10.0.0.1"},
Hostname: utilpointer.StringPtr("pod-1"),
Topology: map[string]string{"kubernetes.io/hostname": "node-1"},
NodeName: utilpointer.StringPtr("node-1"),
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
}, {
Addresses: []string{"10.0.0.2"},
Hostname: utilpointer.StringPtr("pod-2"),
Topology: map[string]string{"kubernetes.io/hostname": "node-2"},
NodeName: utilpointer.StringPtr("node-2"),
Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
}},
}},
@ -656,7 +656,7 @@ func TestReconcile(t *testing.T) {
discovery.LabelServiceName: endpoints.Name,
discovery.LabelManagedBy: controllerName,
}
_, err := client.DiscoveryV1beta1().EndpointSlices(namespace).Create(context.TODO(), epSlice, metav1.CreateOptions{})
_, err := client.DiscoveryV1().EndpointSlices(namespace).Create(context.TODO(), epSlice, metav1.CreateOptions{})
if err != nil {
t.Fatalf("Expected no error creating EndpointSlice, got %v", err)
}
@ -846,11 +846,11 @@ func expectMatchingAddresses(t *testing.T, epSubset corev1.EndpointSubset, esEnd
}
if expectedEndpoint.epAddress.NodeName != nil {
topologyHostname, ok := endpoint.Topology["kubernetes.io/hostname"]
if !ok {
t.Errorf("Expected topology[kubernetes.io/hostname] to be set")
} else if *expectedEndpoint.epAddress.NodeName != topologyHostname {
t.Errorf("Expected topology[kubernetes.io/hostname] to be %s, got %s", *expectedEndpoint.epAddress.NodeName, topologyHostname)
if endpoint.NodeName == nil {
t.Errorf("Expected nodeName to be set")
}
if *expectedEndpoint.epAddress.NodeName != *endpoint.NodeName {
t.Errorf("Expected nodeName to be %s, got %s", *expectedEndpoint.epAddress.NodeName, *endpoint.NodeName)
}
}
}
@ -858,7 +858,7 @@ func expectMatchingAddresses(t *testing.T, epSubset corev1.EndpointSubset, esEnd
func fetchEndpointSlices(t *testing.T, client *fake.Clientset, namespace string) []discovery.EndpointSlice {
t.Helper()
fetchedSlices, err := client.DiscoveryV1beta1().EndpointSlices(namespace).List(context.TODO(), metav1.ListOptions{
fetchedSlices, err := client.DiscoveryV1().EndpointSlices(namespace).List(context.TODO(), metav1.ListOptions{
LabelSelector: discovery.LabelManagedBy + "=" + controllerName,
})
if err != nil {

48
pkg/controller/endpointslicemirroring/utils.go
View File

@ -22,8 +22,7 @@ import (
"strings"
corev1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
apiequality "k8s.io/apimachinery/pkg/api/equality"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@ -62,25 +61,6 @@ func getAddressType(address string) *discovery.AddressType {
return &addressType
}
// endpointsEqualBeyondHash returns true if endpoints have equal attributes
// but excludes equality checks that would have already been covered with
// endpoint hashing (see hashEndpoint func for more info).
func endpointsEqualBeyondHash(ep1, ep2 *discovery.Endpoint) bool {
if !apiequality.Semantic.DeepEqual(ep1.Topology, ep2.Topology) {
return false
}
if !boolPtrEqual(ep1.Conditions.Ready, ep2.Conditions.Ready) {
return false
}
if !objectRefPtrEqual(ep1.TargetRef, ep2.TargetRef) {
return false
}
return true
}
// newEndpointSlice returns an EndpointSlice generated from an Endpoints
// resource, ports, and address type.
func newEndpointSlice(endpoints *corev1.Endpoints, ports []discovery.EndpointPort, addrType discovery.AddressType, sliceName string) *discovery.EndpointSlice {
@ -135,9 +115,6 @@ func addressToEndpoint(address corev1.EndpointAddress, ready bool) *discovery.En
}
if address.NodeName != nil {
endpoint.Topology = map[string]string{
"kubernetes.io/hostname": *address.NodeName,
}
endpoint.NodeName = address.NodeName
}
if address.Hostname != "" {
@ -163,29 +140,6 @@ func epPortsToEpsPorts(epPorts []corev1.EndpointPort) []discovery.EndpointPort {
return epsPorts
}
// boolPtrEqual returns true if a set of bool pointers have equivalent values.
func boolPtrEqual(ptr1, ptr2 *bool) bool {
if (ptr1 == nil) != (ptr2 == nil) {
return false
}
if ptr1 != nil && ptr2 != nil && *ptr1 != *ptr2 {
return false
}
return true
}
// objectRefPtrEqual returns true if a set of object ref pointers have
// equivalent values.
func objectRefPtrEqual(ref1, ref2 *corev1.ObjectReference) bool {
if (ref1 == nil) != (ref2 == nil) {
return false
}
if ref1 != nil && ref2 != nil && !apiequality.Semantic.DeepEqual(*ref1, *ref2) {
return false
}
return true
}
// getServiceFromDeleteAction parses a Service resource from a delete
// action.
func getServiceFromDeleteAction(obj interface{}) *corev1.Service {

5
pkg/controller/endpointslicemirroring/utils_test.go
View File

@ -22,7 +22,7 @@ import (
"github.com/stretchr/testify/assert"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
@ -97,9 +97,6 @@ func TestAddressToEndpoint(t *testing.T) {
Conditions: discovery.EndpointConditions{
Ready: utilpointer.BoolPtr(true),
},
Topology: map[string]string{
"kubernetes.io/hostname": "node-abc",
},
TargetRef: &v1.ObjectReference{
APIVersion: "v1",
Kind: "Pod",

60
pkg/controller/util/endpoint/controller_utils.go
View File

@ -25,7 +25,8 @@ import (
"sync"
v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1"
discovery "k8s.io/api/discovery/v1"
apiequality "k8s.io/apimachinery/pkg/api/equality"
"k8s.io/apimachinery/pkg/labels"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/sets"
@ -275,3 +276,60 @@ func (sl portsInOrder) Less(i, j int) bool {
h2 := DeepHashObjectToString(sl[j])
return h1 < h2
}
// endpointsEqualBeyondHash returns true if endpoints have equal attributes
// but excludes equality checks that would have already been covered with
// endpoint hashing (see hashEndpoint func for more info).
func EndpointsEqualBeyondHash(ep1, ep2 *discovery.Endpoint) bool {
if stringPtrChanged(ep1.NodeName, ep1.NodeName) {
return false
}
if stringPtrChanged(ep1.Zone, ep1.Zone) {
return false
}
if boolPtrChanged(ep1.Conditions.Ready, ep2.Conditions.Ready) {
return false
}
if objectRefPtrChanged(ep1.TargetRef, ep2.TargetRef) {
return false
}
return true
}
// boolPtrChanged returns true if a set of bool pointers have different values.
func boolPtrChanged(ptr1, ptr2 *bool) bool {
if (ptr1 == nil) != (ptr2 == nil) {
return true
}
if ptr1 != nil && ptr2 != nil && *ptr1 != *ptr2 {
return true
}
return false
}
// objectRefPtrChanged returns true if a set of object ref pointers have
// different values.
func objectRefPtrChanged(ref1, ref2 *v1.ObjectReference) bool {
if (ref1 == nil) != (ref2 == nil) {
return true
}
if ref1 != nil && ref2 != nil && !apiequality.Semantic.DeepEqual(*ref1, *ref2) {
return true
}
return false
}
// stringPtrChanged returns true if a set of string pointers have different values.
func stringPtrChanged(ptr1, ptr2 *string) bool {
if (ptr1 == nil) != (ptr2 == nil) {
return true
}
if ptr1 != nil && ptr2 != nil && *ptr1 != *ptr2 {
return true
}
return false
}

2
pkg/features/kube_features.go
View File

@ -466,6 +466,8 @@ const (
// owner: @robscott @freehan
// alpha: v1.16
// beta: v1.18
// ga: v1.21
//
// Enable Endpoint Slices for more scalable Service endpoints.
EndpointSlice featuregate.Feature = "EndpointSlice"

2
test/integration/dualstack/dualstack_endpoints_test.go
View File

@ -115,7 +115,7 @@ func TestDualStackEndpoints(t *testing.T) {
informers.Core().V1().Pods(),
informers.Core().V1().Services(),
informers.Core().V1().Nodes(),
informers.Discovery().V1beta1().EndpointSlices(),
informers.Discovery().V1().EndpointSlices(),
int32(100),
client,
1*time.Second)