Merge pull request #831 from cybozu-go/fix-eps-ops

Confirm EndpointSlice in mtest

Author: masa213f

mtest/kubernetes_test.go

@@ -317,6 +317,8 @@ func testKubernetes() {
 		Expect(err).NotTo(HaveOccurred(), "stderr=%s", stderr)
 		_, stderr, err = kubectl("-n", "kube-system", "get", "endpoints/cke-etcd")
 		Expect(err).NotTo(HaveOccurred(), "stderr=%s", stderr)
+		_, stderr, err = kubectl("-n", "kube-system", "get", "endpointslices/cke-etcd")
+		Expect(err).NotTo(HaveOccurred(), "stderr=%s", stderr)
 	})
 
 	It("can output audit log to journal log", func() {

mtest/operators_test.go

@@ -11,8 +11,10 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	kubeletv1beta1 "k8s.io/kubelet/config/v1beta1"
+	"k8s.io/utils/ptr"
 )
 
 func testOperators(isDegraded bool) {
@@ -76,6 +78,44 @@ func testOperators(isDegraded bool) {
 			Expect(ep.Subsets[0].NotReadyAddresses).Should(BeEmpty())
 		}
 
+		By("Testing default/kubernetes EndpointSlices")
+		out, _, err = kubectl("get", "-o=json", "endpointslices/kubernetes")
+		Expect(err).NotTo(HaveOccurred())
+		var eps discoveryv1.EndpointSlice
+		err = json.Unmarshal(out, &eps)
+		Expect(err).NotTo(HaveOccurred())
+		if isDegraded {
+			Expect(eps.Endpoints).To(ConsistOf(
+				discoveryv1.Endpoint{
+					Addresses:  []string{node1},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)},
+				},
+				discoveryv1.Endpoint{
+					Addresses:  []string{node2},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(false)},
+				},
+				discoveryv1.Endpoint{
+					Addresses:  []string{node3},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(false)},
+				},
+			))
+		} else {
+			Expect(eps.Endpoints).To(ConsistOf(
+				discoveryv1.Endpoint{
+					Addresses:  []string{node1},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)},
+				},
+				discoveryv1.Endpoint{
+					Addresses:  []string{node2},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)},
+				},
+				discoveryv1.Endpoint{
+					Addresses:  []string{node3},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)},
+				},
+			))
+		}
+
 		By("Stopping etcd servers")
 		// this will run:
 		// - EtcdStartOp
@@ -129,6 +169,36 @@ func testOperators(isDegraded bool) {
 			Expect(ep.Subsets[0].NotReadyAddresses).Should(BeEmpty())
 		}
 
+		By("Testing default/kubernetes EndpointSlices")
+		out, _, err = kubectl("get", "-o=json", "endpointslices/kubernetes")
+		Expect(err).NotTo(HaveOccurred())
+		eps = discoveryv1.EndpointSlice{}
+		err = json.Unmarshal(out, &eps)
+		Expect(err).NotTo(HaveOccurred())
+		if isDegraded {
+			Expect(eps.Endpoints).To(ConsistOf(
+				discoveryv1.Endpoint{
+					Addresses:  []string{node1},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)},
+				},
+				discoveryv1.Endpoint{
+					Addresses:  []string{node3},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(false)},
+				},
+			))
+		} else {
+			Expect(eps.Endpoints).To(ConsistOf(
+				discoveryv1.Endpoint{
+					Addresses:  []string{node1},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)},
+				},
+				discoveryv1.Endpoint{
+					Addresses:  []string{node3},
+					Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)},
+				},
+			))
+		}
+
 		By("Adding a new node to the cluster as a control plane")
 		// this will run AddMemberOp as well as other boot/restart ops.
 

mtest/reboot_test.go

@@ -12,6 +12,7 @@ import (
 	. "github.com/onsi/gomega"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 )
@@ -581,7 +582,6 @@ func testRebootOperations() {
 			Expect(apiServerInProgress).Should(BeNumerically("<=", 1), "API servers should be processed one by one")
 
 			// Both default/kubernetes and kube-system/cke-etcd Endpoints should have no less than two endpoints during API server reboots.
-			// (Check for EndpointsSlices is omitted since their members are same as Endpoints)
 			epNames := []struct {
 				name      string
 				namespace string
@@ -605,6 +605,23 @@ func testRebootOperations() {
 				Expect(ep.Subsets).Should(HaveLen(1))
 				Expect(len(ep.Subsets[0].Addresses)).Should(BeNumerically(">=", 2))
 			}
+			for _, epName := range epNames {
+				out, _, err := kubectl("get", "endpointslices", "-n", epName.namespace, epName.name, "-o=json")
+				Expect(err).NotTo(HaveOccurred())
+
+				var eps discoveryv1.EndpointSlice
+				err = json.Unmarshal(out, &eps)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(eps.Endpoints).To(HaveLen(3))
+
+				readyCount := 0
+				for _, e := range eps.Endpoints {
+					if *e.Conditions.Ready {
+						readyCount++
+					}
+				}
+				Expect(readyCount).To(BeNumerically(">=", 2))
+			}
 
 			if apiServerRemain == 0 {
 				break

server/strategy.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"slices"
 	"time"
 
 	"github.com/cybozu-go/cke"
@@ -14,6 +15,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
 )
 
 // DecideOps returns the next operations to do and the operation phase.
@@ -441,6 +443,17 @@ func decideEpEpsOps(expect *endpointParams, actualEP *corev1.Endpoints, actualEP
 		ops = append(ops, epOp)
 	}
 
+	ready := make(map[string]bool)
+	for _, ip := range expect.readyIPs {
+		ready[ip] = true
+	}
+	for _, ip := range expect.notReadyIPs {
+		ready[ip] = false
+	}
+	addrs := slices.Concat(expect.readyIPs, expect.notReadyIPs)
+	slices.Sort(addrs)
+	addrs = slices.Compact(addrs)
+
 	eps := &discoveryv1.EndpointSlice{}
 	eps.Namespace = expect.namespace
 	eps.Name = expect.name
@@ -449,23 +462,11 @@ func decideEpEpsOps(expect *endpointParams, actualEP *corev1.Endpoints, actualEP
 		"kubernetes.io/service-name":             expect.serviceName,
 	}
 	eps.AddressType = discoveryv1.AddressTypeIPv4
-	eps.Endpoints = make([]discoveryv1.Endpoint, len(expect.readyIPs)+len(expect.notReadyIPs))
-	readyTrue := true
-	for i := range expect.readyIPs {
+	eps.Endpoints = make([]discoveryv1.Endpoint, len(addrs))
+	for i, ip := range addrs {
 		eps.Endpoints[i] = discoveryv1.Endpoint{
-			Addresses: expect.readyIPs[i : i+1],
-			Conditions: discoveryv1.EndpointConditions{
-				Ready: &readyTrue,
-			},
-		}
-	}
-	readyFalse := false
-	for i := range expect.notReadyIPs {
-		eps.Endpoints[len(expect.readyIPs)+i] = discoveryv1.Endpoint{
-			Addresses: expect.notReadyIPs[i : i+1],
-			Conditions: discoveryv1.EndpointConditions{
-				Ready: &readyFalse,
-			},
+			Addresses:  []string{ip},
+			Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(ready[ip])},
 		}
 	}
 	eps.Ports = []discoveryv1.EndpointPort{