Merge pull request kubernetes#94443 from aojea/slicesLabels

endpointslice controller should mirror parent service labels

Author: k8s-ci-robot

pkg/controller/endpointslice/BUILD

@@ -14,6 +14,7 @@ go_library(
     deps = [
         "//pkg/api/v1/pod:go_default_library",
         "//pkg/apis/core:go_default_library",
+        "//pkg/apis/core/v1/helper:go_default_library",
         "//pkg/apis/discovery/validation:go_default_library",
         "//pkg/controller:go_default_library",
         "//pkg/controller/endpointslice/metrics:go_default_library",

pkg/controller/endpointslice/reconciler.go

@@ -248,9 +248,11 @@ func (r *reconciler) finalize(
 
 // reconcileByPortMapping compares the endpoints found in existing slices with
 // the list of desired endpoints and returns lists of slices to create, update,
-// and delete. The logic is split up into several main steps:
+// and delete. It also checks that the slices mirror the parent services labels.
+// The logic is split up into several main steps:
 // 1. Iterate through existing slices, delete endpoints that are no longer
-//    desired and update matching endpoints that have changed.
+//    desired and update matching endpoints that have changed. It also checks
+//    if the slices have the labels of the parent services, and updates them if not.
 // 2. Iterate through slices that have been modified in 1 and fill them up with
 //    any remaining desired endpoints.
 // 3. If there still desired endpoints left, try to fit them into a previously
@@ -289,6 +291,9 @@ func (r *reconciler) reconcileByPortMapping(
 			}
 		}
 
+		// generate the slice labels and check if parent labels have changed
+		labels, labelsChanged := setEndpointSliceLabels(existingSlice, service)
+
 		// If an endpoint was updated or removed, mark for update or delete
 		if endpointUpdated || len(existingSlice.Endpoints) != len(newEndpoints) {
 			if len(existingSlice.Endpoints) > len(newEndpoints) {
@@ -301,9 +306,16 @@ func (r *reconciler) reconcileByPortMapping(
 				// otherwise, copy and mark for update
 				epSlice := existingSlice.DeepCopy()
 				epSlice.Endpoints = newEndpoints
+				epSlice.Labels = labels
 				slicesByName[existingSlice.Name] = epSlice
 				sliceNamesToUpdate.Insert(epSlice.Name)
 			}
+		} else if labelsChanged {
+			// if labels have changed, copy and mark for update
+			epSlice := existingSlice.DeepCopy()
+			epSlice.Labels = labels
+			slicesByName[existingSlice.Name] = epSlice
+			sliceNamesToUpdate.Insert(epSlice.Name)
 		} else {
 			// slices with no changes will be useful if there are leftover endpoints
 			sliceNamesUnchanged.Insert(existingSlice.Name)

pkg/controller/endpointslice/reconciler_test.go

@@ -72,6 +72,13 @@ func TestReconcile1Pod(t *testing.T) {
 	namespace := "test"
 	ipv6Family := corev1.IPv6Protocol
 	svcv4, _ := newServiceAndEndpointMeta("foo", namespace)
+	svcv4ClusterIP, _ := newServiceAndEndpointMeta("foo", namespace)
+	svcv4ClusterIP.Spec.ClusterIP = "1.1.1.1"
+	svcv4Labels, _ := newServiceAndEndpointMeta("foo", namespace)
+	svcv4Labels.Labels = map[string]string{"foo": "bar"}
+	svcv4BadLabels, _ := newServiceAndEndpointMeta("foo", namespace)
+	svcv4BadLabels.Labels = map[string]string{discovery.LabelServiceName: "bad",
+		discovery.LabelManagedBy: "actor", corev1.IsHeadlessService: "invalid"}
 	svcv6, _ := newServiceAndEndpointMeta("foo", namespace)
 	svcv6.Spec.IPFamily = &ipv6Family
 	svcv6ClusterIP, _ := newServiceAndEndpointMeta("foo", namespace)
@@ -94,6 +101,7 @@ func TestReconcile1Pod(t *testing.T) {
 		service             corev1.Service
 		expectedAddressType discovery.AddressType
 		expectedEndpoint    discovery.Endpoint
+		expectedLabels      map[string]string
 	}{
 		"ipv4": {
 			service:             svcv4,
@@ -112,6 +120,80 @@ func TestReconcile1Pod(t *testing.T) {
 					Name:      "pod1",
 				},
 			},
+			expectedLabels: map[string]string{
+				discovery.LabelManagedBy:   controllerName,
+				discovery.LabelServiceName: "foo",
+				corev1.IsHeadlessService:   "",
+			},
+		},
+		"ipv4-clusterip": {
+			service:             svcv4ClusterIP,
+			expectedAddressType: discovery.AddressTypeIPv4,
+			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",
+				},
+				TargetRef: &corev1.ObjectReference{
+					Kind:      "Pod",
+					Namespace: namespace,
+					Name:      "pod1",
+				},
+			},
+			expectedLabels: map[string]string{
+				discovery.LabelManagedBy:   controllerName,
+				discovery.LabelServiceName: "foo",
+			},
+		},
+		"ipv4-labels": {
+			service:             svcv4Labels,
+			expectedAddressType: discovery.AddressTypeIPv4,
+			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",
+				},
+				TargetRef: &corev1.ObjectReference{
+					Kind:      "Pod",
+					Namespace: namespace,
+					Name:      "pod1",
+				},
+			},
+			expectedLabels: map[string]string{
+				discovery.LabelManagedBy:   controllerName,
+				discovery.LabelServiceName: "foo",
+				"foo":                      "bar",
+				corev1.IsHeadlessService:   "",
+			},
+		},
+		"ipv4-bad-labels": {
+			service:             svcv4BadLabels,
+			expectedAddressType: discovery.AddressTypeIPv4,
+			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",
+				},
+				TargetRef: &corev1.ObjectReference{
+					Kind:      "Pod",
+					Namespace: namespace,
+					Name:      "pod1",
+				},
+			},
+			expectedLabels: map[string]string{
+				discovery.LabelManagedBy:   controllerName,
+				discovery.LabelServiceName: "foo",
+				corev1.IsHeadlessService:   "",
+			},
 		},
 		"ipv6": {
 			service:             svcv6,
@@ -130,6 +212,11 @@ func TestReconcile1Pod(t *testing.T) {
 					Name:      "pod1",
 				},
 			},
+			expectedLabels: map[string]string{
+				discovery.LabelManagedBy:   controllerName,
+				discovery.LabelServiceName: "foo",
+				corev1.IsHeadlessService:   "",
+			},
 		},
 		"ipv6-clusterip": {
 			service:             svcv6ClusterIP,
@@ -148,6 +235,10 @@ func TestReconcile1Pod(t *testing.T) {
 					Name:      "pod1",
 				},
 			},
+			expectedLabels: map[string]string{
+				discovery.LabelManagedBy:   controllerName,
+				discovery.LabelServiceName: "foo",
+			},
 		},
 	}
 
@@ -174,8 +265,8 @@ func TestReconcile1Pod(t *testing.T) {
 				t.Errorf("Expected EndpointSlice name to start with %s, got %s", testCase.service.Name, slice.Name)
 			}
 
-			if slice.Labels[discovery.LabelServiceName] != testCase.service.Name {
-				t.Errorf("Expected EndpointSlice to have label set with %s value, got %s", testCase.service.Name, slice.Labels[discovery.LabelServiceName])
+			if !reflect.DeepEqual(testCase.expectedLabels, slice.Labels) {
+				t.Errorf("Expected EndpointSlice to have labels: %v , got %v", testCase.expectedLabels, slice.Labels)
 			}
 
 			if slice.Annotations[corev1.EndpointsLastChangeTriggerTime] != triggerTime.Format(time.RFC3339Nano) {
@@ -431,6 +522,81 @@ func TestReconcileEndpointSlicesUpdating(t *testing.T) {
 	expectUnorderedSlicesWithLengths(t, fetchEndpointSlices(t, client, namespace), []int{100, 100, 50})
 }
 
+// In some cases, such as service labels updates, all slices for that service will require a change
+// This test ensures that we are updating those slices and not calling create + delete for each
+func TestReconcileEndpointSlicesServicesLabelsUpdating(t *testing.T) {
+	client := newClientset()
+	namespace := "test"
+	svc, _ := newServiceAndEndpointMeta("foo", namespace)
+
+	// start with 250 pods
+	pods := []*corev1.Pod{}
+	for i := 0; i < 250; i++ {
+		ready := !(i%3 == 0)
+		pods = append(pods, newPod(i, namespace, ready, 1))
+	}
+
+	r := newReconciler(client, []*corev1.Node{{ObjectMeta: metav1.ObjectMeta{Name: "node-1"}}}, defaultMaxEndpointsPerSlice)
+	reconcileHelper(t, r, &svc, pods, []*discovery.EndpointSlice{}, time.Now())
+	numActionsExpected := 3
+	assert.Len(t, client.Actions(), numActionsExpected, "Expected 3 additional clientset actions")
+
+	slices := fetchEndpointSlices(t, client, namespace)
+	numActionsExpected++
+	expectUnorderedSlicesWithLengths(t, slices, []int{100, 100, 50})
+
+	// update service with new labels
+	svc.Labels = map[string]string{"foo": "bar"}
+	reconcileHelper(t, r, &svc, pods, []*discovery.EndpointSlice{&slices[0], &slices[1], &slices[2]}, time.Now())
+
+	numActionsExpected += 3
+	assert.Len(t, client.Actions(), numActionsExpected, "Expected 3 additional clientset actions")
+	expectActions(t, client.Actions(), 3, "update", "endpointslices")
+
+	newSlices := fetchEndpointSlices(t, client, namespace)
+	expectUnorderedSlicesWithLengths(t, newSlices, []int{100, 100, 50})
+	// check that the labels were updated
+	for _, slice := range newSlices {
+		w, ok := slice.Labels["foo"]
+		if !ok {
+			t.Errorf("Expected label \"foo\" from parent service not found")
+		} else if "bar" != w {
+			t.Errorf("Expected EndpointSlice to have parent service labels: have %s value, expected bar", w)
+		}
+	}
+}
+
+// In some cases, such as service labels updates, all slices for that service will require a change
+// However, this should not happen for reserved labels
+func TestReconcileEndpointSlicesServicesReservedLabels(t *testing.T) {
+	client := newClientset()
+	namespace := "test"
+	svc, _ := newServiceAndEndpointMeta("foo", namespace)
+
+	// start with 250 pods
+	pods := []*corev1.Pod{}
+	for i := 0; i < 250; i++ {
+		ready := !(i%3 == 0)
+		pods = append(pods, newPod(i, namespace, ready, 1))
+	}
+
+	r := newReconciler(client, []*corev1.Node{{ObjectMeta: metav1.ObjectMeta{Name: "node-1"}}}, defaultMaxEndpointsPerSlice)
+	reconcileHelper(t, r, &svc, pods, []*discovery.EndpointSlice{}, time.Now())
+	numActionsExpected := 3
+	assert.Len(t, client.Actions(), numActionsExpected, "Expected 3 additional clientset actions")
+	slices := fetchEndpointSlices(t, client, namespace)
+	numActionsExpected++
+	expectUnorderedSlicesWithLengths(t, slices, []int{100, 100, 50})
+
+	// update service with new labels
+	svc.Labels = map[string]string{discovery.LabelServiceName: "bad", discovery.LabelManagedBy: "actor", corev1.IsHeadlessService: "invalid"}
+	reconcileHelper(t, r, &svc, pods, []*discovery.EndpointSlice{&slices[0], &slices[1], &slices[2]}, time.Now())
+	assert.Len(t, client.Actions(), numActionsExpected, "Expected no additional clientset actions")
+
+	newSlices := fetchEndpointSlices(t, client, namespace)
+	expectUnorderedSlicesWithLengths(t, newSlices, []int{100, 100, 50})
+}
+
 // In this test, we start with 10 slices that only have 30 endpoints each
 // An initial reconcile makes no changes (as desired to limit writes)
 // When we change a service port, all slices will need to be updated in some way

pkg/controller/endpointslice/utils.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1beta1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -30,6 +31,7 @@ import (
 	"k8s.io/klog/v2"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	api "k8s.io/kubernetes/pkg/apis/core"
+	helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	"k8s.io/kubernetes/pkg/apis/discovery/validation"
 	endpointutil "k8s.io/kubernetes/pkg/controller/util/endpoint"
 	utilnet "k8s.io/utils/net"
@@ -152,12 +154,9 @@ func endpointsEqualBeyondHash(ep1, ep2 *discovery.Endpoint) bool {
 func newEndpointSlice(service *corev1.Service, endpointMeta *endpointMeta) *discovery.EndpointSlice {
 	gvk := schema.GroupVersionKind{Version: "v1", Kind: "Service"}
 	ownerRef := metav1.NewControllerRef(service, gvk)
-	return &discovery.EndpointSlice{
+	epSlice := &discovery.EndpointSlice{
 		ObjectMeta: metav1.ObjectMeta{
-			Labels: map[string]string{
-				discovery.LabelServiceName: service.Name,
-				discovery.LabelManagedBy:   controllerName,
-			},
+			Labels:          map[string]string{},
 			GenerateName:    getEndpointSlicePrefix(service.Name),
 			OwnerReferences: []metav1.OwnerReference{*ownerRef},
 			Namespace:       service.Namespace,
@@ -166,6 +165,10 @@ func newEndpointSlice(service *corev1.Service, endpointMeta *endpointMeta) *disc
 		AddressType: endpointMeta.AddressType,
 		Endpoints:   []discovery.Endpoint{},
 	}
+	// add parent service labels
+	epSlice.Labels, _ = setEndpointSliceLabels(epSlice, service)
+
+	return epSlice
 }
 
 // getEndpointSlicePrefix returns a suitable prefix for an EndpointSlice name.
@@ -278,6 +281,62 @@ func serviceControllerKey(endpointSlice *discovery.EndpointSlice) (string, error
 	return fmt.Sprintf("%s/%s", endpointSlice.Namespace, serviceName), nil
 }
 
+// setEndpointSliceLabels returns a map with the new endpoint slices labels and true if there was an update.
+// Slices labels must be equivalent to the Service labels except for the reserved IsHeadlessService, LabelServiceName and LabelManagedBy labels
+// Changes to IsHeadlessService, LabelServiceName and LabelManagedBy labels on the Service do not result in updates to EndpointSlice labels.
+func setEndpointSliceLabels(epSlice *discovery.EndpointSlice, service *corev1.Service) (map[string]string, bool) {
+	updated := false
+	epLabels := make(map[string]string)
+	svcLabels := make(map[string]string)
+
+	// check if the endpoint slice and the service have the same labels
+	// clone current slice labels except the reserved labels
+	for key, value := range epSlice.Labels {
+		if IsReservedLabelKey(key) {
+			continue
+		}
+		// copy endpoint slice labels
+		epLabels[key] = value
+	}
+
+	for key, value := range service.Labels {
+		if IsReservedLabelKey(key) {
+			klog.Warningf("Service %s/%s using reserved endpoint slices label, skipping label %s: %s", service.Namespace, service.Name, key, value)
+			continue
+		}
+		// copy service labels
+		svcLabels[key] = value
+	}
+
+	// if the labels are not identical update the slice with the corresponding service labels
+	if !apiequality.Semantic.DeepEqual(epLabels, svcLabels) {
+		updated = true
+	}
+
+	// add or remove headless label depending on the service Type
+	if !helper.IsServiceIPSet(service) {
+		svcLabels[v1.IsHeadlessService] = ""
+	} else {
+		delete(svcLabels, v1.IsHeadlessService)
+	}
+
+	// override endpoint slices reserved labels
+	svcLabels[discovery.LabelServiceName] = service.Name
+	svcLabels[discovery.LabelManagedBy] = controllerName
+
+	return svcLabels, updated
+}
+
+// IsReservedLabelKey return true if the label is one of the reserved label for slices
+func IsReservedLabelKey(label string) bool {
+	if label == discovery.LabelServiceName ||
+		label == discovery.LabelManagedBy ||
+		label == v1.IsHeadlessService {
+		return true
+	}
+	return false
+}
+
 // endpointSliceEndpointLen helps sort endpoint slices by the number of
 // endpoints they contain.
 type endpointSliceEndpointLen []*discovery.EndpointSlice