From 8acda3631f798fbb04eff647d04827adac4aee17 Mon Sep 17 00:00:00 2001 From: Dan Winship Date: Mon, 21 Jul 2025 10:23:31 -0400 Subject: [PATCH] Belatedly add tests of proxy NodeConfig and ServiceCIDRConfig --- pkg/proxy/config/config_test.go | 497 +++++++++++++++++++++++++------- 1 file changed, 385 insertions(+), 112 deletions(-) diff --git a/pkg/proxy/config/config_test.go b/pkg/proxy/config/config_test.go index 88802c3667a..bd913b1732b 100644 --- a/pkg/proxy/config/config_test.go +++ b/pkg/proxy/config/config_test.go @@ -18,13 +18,15 @@ package config import ( "reflect" - "sort" + "slices" + "strings" "sync" "testing" "time" "k8s.io/api/core/v1" discoveryv1 "k8s.io/api/discovery/v1" + networkingv1 "k8s.io/api/networking/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/wait" @@ -36,18 +38,6 @@ import ( "k8s.io/utils/ptr" ) -type sortedServices []*v1.Service - -func (s sortedServices) Len() int { - return len(s) -} -func (s sortedServices) Swap(i, j int) { - s[i], s[j] = s[j], s[i] -} -func (s sortedServices) Less(i, j int) bool { - return s[i].Name < s[j].Name -} - type ServiceHandlerMock struct { lock sync.Mutex @@ -107,7 +97,7 @@ func (h *ServiceHandlerMock) sendServices() { for _, svc := range h.state { services = append(services, svc) } - sort.Sort(sortedServices(services)) + slices.SortFunc(services, func(a, b *v1.Service) int { return strings.Compare(a.Name, b.Name) }) h.process(services) } @@ -131,101 +121,6 @@ func (h *ServiceHandlerMock) ValidateServices(t *testing.T, expectedServices []* } } -type sortedEndpointSlices []*discoveryv1.EndpointSlice - -func (s sortedEndpointSlices) Len() int { - return len(s) -} -func (s sortedEndpointSlices) Swap(i, j int) { - s[i], s[j] = s[j], s[i] -} -func (s sortedEndpointSlices) Less(i, j int) bool { - return s[i].Name < s[j].Name -} - -type EndpointSliceHandlerMock struct { - lock sync.Mutex - - state map[types.NamespacedName]*discoveryv1.EndpointSlice - synced bool - updated chan []*discoveryv1.EndpointSlice - process func([]*discoveryv1.EndpointSlice) -} - -func NewEndpointSliceHandlerMock() *EndpointSliceHandlerMock { - ehm := &EndpointSliceHandlerMock{ - state: make(map[types.NamespacedName]*discoveryv1.EndpointSlice), - updated: make(chan []*discoveryv1.EndpointSlice, 5), - } - ehm.process = func(endpoints []*discoveryv1.EndpointSlice) { - ehm.updated <- endpoints - } - return ehm -} - -func (h *EndpointSliceHandlerMock) OnEndpointSliceAdd(slice *discoveryv1.EndpointSlice) { - h.lock.Lock() - defer h.lock.Unlock() - namespacedName := types.NamespacedName{Namespace: slice.Namespace, Name: slice.Name} - h.state[namespacedName] = slice - h.sendEndpointSlices() -} - -func (h *EndpointSliceHandlerMock) OnEndpointSliceUpdate(oldSlice, slice *discoveryv1.EndpointSlice) { - h.lock.Lock() - defer h.lock.Unlock() - namespacedName := types.NamespacedName{Namespace: slice.Namespace, Name: slice.Name} - h.state[namespacedName] = slice - h.sendEndpointSlices() -} - -func (h *EndpointSliceHandlerMock) OnEndpointSliceDelete(slice *discoveryv1.EndpointSlice) { - h.lock.Lock() - defer h.lock.Unlock() - namespacedName := types.NamespacedName{Namespace: slice.Namespace, Name: slice.Name} - delete(h.state, namespacedName) - h.sendEndpointSlices() -} - -func (h *EndpointSliceHandlerMock) OnEndpointSlicesSynced() { - h.lock.Lock() - defer h.lock.Unlock() - h.synced = true - h.sendEndpointSlices() -} - -func (h *EndpointSliceHandlerMock) sendEndpointSlices() { - if !h.synced { - return - } - slices := make([]*discoveryv1.EndpointSlice, 0, len(h.state)) - for _, eps := range h.state { - slices = append(slices, eps) - } - sort.Sort(sortedEndpointSlices(slices)) - h.process(slices) -} - -func (h *EndpointSliceHandlerMock) ValidateEndpointSlices(t *testing.T, expectedSlices []*discoveryv1.EndpointSlice) { - // We might get 1 or more updates for N endpointslice updates, because we - // over write older snapshots of endpointslices from the producer go-routine - // if the consumer falls behind. Unittests will hard timeout in 5m. - var slices []*discoveryv1.EndpointSlice - for { - select { - case slices = <-h.updated: - if reflect.DeepEqual(slices, expectedSlices) { - return - } - // Unittests will hard timeout in 5m with a stack trace, prevent that - // and surface a clearer reason for failure. - case <-time.After(wait.ForeverTestTimeout): - t.Errorf("Timed out. Expected %#v, Got %#v", expectedSlices, slices) - return - } - } -} - func TestNewServiceAddedAndNotified(t *testing.T) { _, ctx := klogtesting.NewTestContext(t) client := fake.NewSimpleClientset() @@ -323,6 +218,89 @@ func TestNewServicesMultipleHandlersAddedAndNotified(t *testing.T) { handler2.ValidateServices(t, services) } +type EndpointSliceHandlerMock struct { + lock sync.Mutex + + state map[types.NamespacedName]*discoveryv1.EndpointSlice + synced bool + updated chan []*discoveryv1.EndpointSlice + process func([]*discoveryv1.EndpointSlice) +} + +func NewEndpointSliceHandlerMock() *EndpointSliceHandlerMock { + ehm := &EndpointSliceHandlerMock{ + state: make(map[types.NamespacedName]*discoveryv1.EndpointSlice), + updated: make(chan []*discoveryv1.EndpointSlice, 5), + } + ehm.process = func(endpoints []*discoveryv1.EndpointSlice) { + ehm.updated <- endpoints + } + return ehm +} + +func (h *EndpointSliceHandlerMock) OnEndpointSliceAdd(slice *discoveryv1.EndpointSlice) { + h.lock.Lock() + defer h.lock.Unlock() + namespacedName := types.NamespacedName{Namespace: slice.Namespace, Name: slice.Name} + h.state[namespacedName] = slice + h.sendEndpointSlices() +} + +func (h *EndpointSliceHandlerMock) OnEndpointSliceUpdate(oldSlice, slice *discoveryv1.EndpointSlice) { + h.lock.Lock() + defer h.lock.Unlock() + namespacedName := types.NamespacedName{Namespace: slice.Namespace, Name: slice.Name} + h.state[namespacedName] = slice + h.sendEndpointSlices() +} + +func (h *EndpointSliceHandlerMock) OnEndpointSliceDelete(slice *discoveryv1.EndpointSlice) { + h.lock.Lock() + defer h.lock.Unlock() + namespacedName := types.NamespacedName{Namespace: slice.Namespace, Name: slice.Name} + delete(h.state, namespacedName) + h.sendEndpointSlices() +} + +func (h *EndpointSliceHandlerMock) OnEndpointSlicesSynced() { + h.lock.Lock() + defer h.lock.Unlock() + h.synced = true + h.sendEndpointSlices() +} + +func (h *EndpointSliceHandlerMock) sendEndpointSlices() { + if !h.synced { + return + } + endpointSlices := make([]*discoveryv1.EndpointSlice, 0, len(h.state)) + for _, eps := range h.state { + endpointSlices = append(endpointSlices, eps) + } + slices.SortFunc(endpointSlices, func(a, b *discoveryv1.EndpointSlice) int { return strings.Compare(a.Name, b.Name) }) + h.process(endpointSlices) +} + +func (h *EndpointSliceHandlerMock) ValidateEndpointSlices(t *testing.T, expectedSlices []*discoveryv1.EndpointSlice) { + // We might get 1 or more updates for N endpointslice updates, because we + // over write older snapshots of endpointslices from the producer go-routine + // if the consumer falls behind. Unittests will hard timeout in 5m. + var slices []*discoveryv1.EndpointSlice + for { + select { + case slices = <-h.updated: + if reflect.DeepEqual(slices, expectedSlices) { + return + } + // Unittests will hard timeout in 5m with a stack trace, prevent that + // and surface a clearer reason for failure. + case <-time.After(wait.ForeverTestTimeout): + t.Errorf("Timed out. Expected %#v, Got %#v", expectedSlices, slices) + return + } + } +} + func TestNewEndpointsMultipleHandlersAddedAndNotified(t *testing.T) { _, ctx := klogtesting.NewTestContext(t) client := fake.NewSimpleClientset() @@ -453,7 +431,302 @@ func TestNewEndpointsMultipleHandlersAddRemoveSetAndNotified(t *testing.T) { handler2.ValidateEndpointSlices(t, endpoints) } +type NodeHandlerMock struct { + lock sync.Mutex + + state map[string]*v1.Node + synced bool + updated chan []*v1.Node + process func([]*v1.Node) +} + +func NewNodeHandlerMock() *NodeHandlerMock { + h := &NodeHandlerMock{ + state: make(map[string]*v1.Node), + updated: make(chan []*v1.Node, 5), + } + h.process = func(nodes []*v1.Node) { + h.updated <- nodes + } + return h +} + +func (h *NodeHandlerMock) OnNodeAdd(node *v1.Node) { + h.lock.Lock() + defer h.lock.Unlock() + h.state[node.Name] = node + h.sendNodes() +} + +func (h *NodeHandlerMock) OnNodeUpdate(oldNode, node *v1.Node) { + h.lock.Lock() + defer h.lock.Unlock() + h.state[node.Name] = node + h.sendNodes() +} + +func (h *NodeHandlerMock) OnNodeDelete(node *v1.Node) { + h.lock.Lock() + defer h.lock.Unlock() + delete(h.state, node.Name) + h.sendNodes() +} + +func (h *NodeHandlerMock) OnNodeSynced() { + h.lock.Lock() + defer h.lock.Unlock() + h.synced = true + h.sendNodes() +} + +func (h *NodeHandlerMock) sendNodes() { + if !h.synced { + return + } + nodes := make([]*v1.Node, 0, len(h.state)) + for _, svc := range h.state { + nodes = append(nodes, svc) + } + slices.SortFunc(nodes, func(a, b *v1.Node) int { return strings.Compare(a.Name, b.Name) }) + h.process(nodes) +} + +func (h *NodeHandlerMock) ValidateNodes(t *testing.T, expectedNodes []*v1.Node) { + // We might get 1 or more updates for N node updates, because we + // over write older snapshots of nodes from the producer go-routine + // if the consumer falls behind. + var nodes []*v1.Node + for { + select { + case nodes = <-h.updated: + if reflect.DeepEqual(nodes, expectedNodes) { + return + } + // Unittests will hard timeout in 5m with a stack trace, prevent that + // and surface a clearer reason for failure. + case <-time.After(wait.ForeverTestTimeout): + t.Errorf("Timed out. Expected %#v, Got %#v", expectedNodes, nodes) + return + } + } +} + +func TestNewNodesMultipleHandlersAddRemoveSetAndNotified(t *testing.T) { + _, ctx := klogtesting.NewTestContext(t) + client := fake.NewSimpleClientset() + fakeWatch := watch.NewFake() + client.PrependWatchReactor("nodes", ktesting.DefaultWatchReactor(fakeWatch, nil)) + + stopCh := make(chan struct{}) + defer close(stopCh) + + sharedInformers := informers.NewSharedInformerFactory(client, time.Minute) + + config := NewNodeConfig(ctx, sharedInformers.Core().V1().Nodes(), time.Minute) + handler := NewNodeHandlerMock() + handler2 := NewNodeHandlerMock() + config.RegisterEventHandler(handler) + config.RegisterEventHandler(handler2) + sharedInformers.Start(stopCh) + go config.Run(stopCh) + + nodes1 := &v1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "foo"}, + Status: v1.NodeStatus{ + Addresses: []v1.NodeAddress{ + { + Type: v1.NodeInternalIP, + Address: "1.1.1.1", + }, + { + Type: v1.NodeExternalIP, + Address: "2.2.2.2", + }, + }, + }, + } + nodes2 := &v1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "bar"}, + Status: v1.NodeStatus{ + Addresses: []v1.NodeAddress{ + { + Type: v1.NodeInternalIP, + Address: "3.3.3.3", + }, + { + Type: v1.NodeInternalIP, + Address: "fc00::4", + }, + }, + }, + } + fakeWatch.Add(nodes1) + fakeWatch.Add(nodes2) + + nodes := []*v1.Node{nodes2, nodes1} + handler.ValidateNodes(t, nodes) + handler2.ValidateNodes(t, nodes) + + // Add one more + nodes3 := &v1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "foobar"}, + Status: v1.NodeStatus{ + Addresses: []v1.NodeAddress{ + { + Type: v1.NodeInternalIP, + Address: "5.5.5.5", + }, + }, + }, + } + fakeWatch.Add(nodes3) + nodes = []*v1.Node{nodes2, nodes1, nodes3} + handler.ValidateNodes(t, nodes) + handler2.ValidateNodes(t, nodes) + + // Update the "foo" node with a new address + nodes1v2 := &v1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "foo"}, + Status: v1.NodeStatus{ + Addresses: []v1.NodeAddress{ + { + Type: v1.NodeInternalIP, + Address: "6.6.6.6", + }, + }, + }, + } + fakeWatch.Modify(nodes1v2) + nodes = []*v1.Node{nodes2, nodes1v2, nodes3} + handler.ValidateNodes(t, nodes) + handler2.ValidateNodes(t, nodes) + + // Remove "bar" node + fakeWatch.Delete(nodes2) + nodes = []*v1.Node{nodes1v2, nodes3} + handler.ValidateNodes(t, nodes) + handler2.ValidateNodes(t, nodes) +} + +type ServiceCIDRHandlerMock struct { + lock sync.Mutex + + state []string + updated chan []string + process func([]string) +} + +func NewServiceCIDRHandlerMock() *ServiceCIDRHandlerMock { + h := &ServiceCIDRHandlerMock{ + updated: make(chan []string, 5), + } + h.process = func(serviceCIDRs []string) { + h.updated <- serviceCIDRs + } + return h +} + +func (h *ServiceCIDRHandlerMock) OnServiceCIDRsChanged(serviceCIDRs []string) { + h.lock.Lock() + defer h.lock.Unlock() + h.state = serviceCIDRs + h.sendServiceCIDRs() +} + +func (h *ServiceCIDRHandlerMock) sendServiceCIDRs() { + serviceCIDRs := append([]string{}, h.state...) + slices.Sort(serviceCIDRs) + h.process(serviceCIDRs) +} + +func (h *ServiceCIDRHandlerMock) ValidateServiceCIDRs(t *testing.T, expectedServiceCIDRs []string) { + // We might get 1 or more updates for N serviceCIDR updates, because we + // over write older snapshots of nodes from the producer go-routine + // if the consumer falls behind. + var serviceCIDRs []string + for { + select { + case serviceCIDRs = <-h.updated: + if reflect.DeepEqual(serviceCIDRs, expectedServiceCIDRs) { + return + } + t.Logf("Expected %#v, Got %#v", expectedServiceCIDRs, serviceCIDRs) + // Unittests will hard timeout in 5m with a stack trace, prevent that + // and surface a clearer reason for failure. + case <-time.After(wait.ForeverTestTimeout): + t.Errorf("Timed out. Expected %#v, Got %#v", expectedServiceCIDRs, serviceCIDRs) + return + } + } +} + +func TestNewServiceCIDRsMultipleHandlersAddRemoveSetAndNotified(t *testing.T) { + _, ctx := klogtesting.NewTestContext(t) + client := fake.NewSimpleClientset() + fakeWatch := watch.NewFake() + client.PrependWatchReactor("servicecidrs", ktesting.DefaultWatchReactor(fakeWatch, nil)) + + stopCh := make(chan struct{}) + defer close(stopCh) + + sharedInformers := informers.NewSharedInformerFactory(client, time.Minute) + + config := NewServiceCIDRConfig(ctx, sharedInformers.Networking().V1().ServiceCIDRs(), time.Minute) + handler := NewServiceCIDRHandlerMock() + handler2 := NewServiceCIDRHandlerMock() + config.RegisterEventHandler(handler) + config.RegisterEventHandler(handler2) + sharedInformers.Start(stopCh) + go config.Run(stopCh) + + serviceCIDRs1 := &networkingv1.ServiceCIDR{ + ObjectMeta: metav1.ObjectMeta{Name: "foo"}, + Spec: networkingv1.ServiceCIDRSpec{ + CIDRs: []string{"1.1.1.0/24"}, + }, + } + serviceCIDRs2 := &networkingv1.ServiceCIDR{ + ObjectMeta: metav1.ObjectMeta{Name: "bar"}, + Spec: networkingv1.ServiceCIDRSpec{ + CIDRs: []string{"2.2.2.0/24", "fc00::/64"}, + }, + } + fakeWatch.Add(serviceCIDRs1) + fakeWatch.Add(serviceCIDRs2) + + serviceCIDRs := []string{"1.1.1.0/24", "2.2.2.0/24", "fc00::/64"} + handler.ValidateServiceCIDRs(t, serviceCIDRs) + handler2.ValidateServiceCIDRs(t, serviceCIDRs) + + // Add one more + serviceCIDRs3 := &networkingv1.ServiceCIDR{ + ObjectMeta: metav1.ObjectMeta{Name: "foobar"}, + Spec: networkingv1.ServiceCIDRSpec{ + CIDRs: []string{"3.3.3.0/24", "2001:db8::/64"}, + }, + } + fakeWatch.Add(serviceCIDRs3) + serviceCIDRs = []string{"1.1.1.0/24", "2.2.2.0/24", "2001:db8::/64", "3.3.3.0/24", "fc00::/64"} + handler.ValidateServiceCIDRs(t, serviceCIDRs) + handler2.ValidateServiceCIDRs(t, serviceCIDRs) + + // Update the "foo" ServiceCIDR + serviceCIDRs1v2 := &networkingv1.ServiceCIDR{ + ObjectMeta: metav1.ObjectMeta{Name: "foo"}, + Spec: networkingv1.ServiceCIDRSpec{ + CIDRs: []string{"4.4.4.0/24"}, + }, + } + fakeWatch.Modify(serviceCIDRs1v2) + serviceCIDRs = []string{"2.2.2.0/24", "2001:db8::/64", "3.3.3.0/24", "4.4.4.0/24", "fc00::/64"} + handler.ValidateServiceCIDRs(t, serviceCIDRs) + handler2.ValidateServiceCIDRs(t, serviceCIDRs) + + // Remove "bar" ServiceCIDR + fakeWatch.Delete(serviceCIDRs2) + serviceCIDRs = []string{"2001:db8::/64", "3.3.3.0/24", "4.4.4.0/24"} + handler.ValidateServiceCIDRs(t, serviceCIDRs) + handler2.ValidateServiceCIDRs(t, serviceCIDRs) +} + // TODO: Add a unittest for interrupts getting processed in a timely manner. -// Currently this module has a circular dependency with config, and so it's -// named config_test, which means even test methods need to be public. This -// is refactoring that we can avoid by resolving the dependency.