dilim/kubernetes
1
0
Fork 0
mirror of https://github.com/kubernetes/kubernetes.git synced 2026-08-08 18:50:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Belatedly add tests of proxy NodeConfig and ServiceCIDRConfig

Browse Source
This commit is contained in:
Dan Winship
2025-07-21 10:23:31 -04:00
parent 963a9acb6d
commit 8acda3631f
1 changed files with 385 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

497
pkg/proxy/config/config_test.go
View File

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