From 22e1fc1a7b75493d61d2b5ea8eeac32fb26d1c4d Mon Sep 17 00:00:00 2001 From: Prince Pereira Date: Mon, 8 Sep 2025 04:26:56 -0700 Subject: [PATCH] Fix ClusterIP load balancer disappearing when InternalTrafficPolicy: Local is set. --- pkg/proxy/winkernel/hns.go | 20 +-- pkg/proxy/winkernel/hns_test.go | 2 + pkg/proxy/winkernel/proxier.go | 2 +- pkg/proxy/winkernel/proxier_test.go | 199 +++++++++++++++++++++++++++- 4 files changed, 210 insertions(+), 13 deletions(-) diff --git a/pkg/proxy/winkernel/hns.go b/pkg/proxy/winkernel/hns.go index d6e425f8b77..80bb208adff 100644 --- a/pkg/proxy/winkernel/hns.go +++ b/pkg/proxy/winkernel/hns.go @@ -349,17 +349,21 @@ func (hns hns) getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerIn func (hns hns) getLoadBalancer(endpoints []endpointInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error) { var id loadBalancerIdentifier vips := []string{} - // Compute hash from backends (endpoint IDs) - hash, err := hashEndpoints(endpoints) - if err != nil { - klog.V(2).ErrorS(err, "Error hashing endpoints", "endpoints", endpoints) - return nil, err + id, lbIdErr := findLoadBalancerID( + endpoints, + vip, + protocol, + internalPort, + externalPort, + ) + + if lbIdErr != nil { + klog.V(2).ErrorS(lbIdErr, "Error hashing endpoints", "endpoints", endpoints) + return nil, lbIdErr } + if len(vip) > 0 { - id = loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, vip: vip, endpointsHash: hash} vips = append(vips, vip) - } else { - id = loadBalancerIdentifier{protocol: protocol, internalPort: internalPort, externalPort: externalPort, endpointsHash: hash} } if lb, found := previousLoadBalancers[id]; found { diff --git a/pkg/proxy/winkernel/hns_test.go b/pkg/proxy/winkernel/hns_test.go index ca436394b01..b859ad639ef 100644 --- a/pkg/proxy/winkernel/hns_test.go +++ b/pkg/proxy/winkernel/hns_test.go @@ -39,6 +39,8 @@ const ( epIpv6Address = "192::3" epIpAddressB = "192.168.1.4" epIpAddressRemote = "192.168.2.3" + epIpAddressLocal1 = "192.168.4.4" + epIpAddressLocal2 = "192.168.4.5" epPaAddress = "10.0.0.3" protocol = 6 internalPort = 80 diff --git a/pkg/proxy/winkernel/proxier.go b/pkg/proxy/winkernel/proxier.go index 91fb6fad8d4..f8600494d0d 100644 --- a/pkg/proxy/winkernel/proxier.go +++ b/pkg/proxy/winkernel/proxier.go @@ -1520,7 +1520,7 @@ func (proxier *Proxier) syncProxyRules() { } if !proxier.requiresUpdateLoadbalancer(svcInfo.hnsID, len(clusterIPEndpoints)) { - proxier.deleteExistingLoadBalancer(hns, svcInfo.winProxyOptimization, &svcInfo.hnsID, svcInfo.ClusterIP().String(), Enum(svcInfo.Protocol()), uint16(svcInfo.targetPort), uint16(svcInfo.Port()), hnsEndpoints, queriedLoadBalancers) + proxier.deleteExistingLoadBalancer(hns, svcInfo.winProxyOptimization, &svcInfo.hnsID, svcInfo.ClusterIP().String(), Enum(svcInfo.Protocol()), uint16(svcInfo.targetPort), uint16(svcInfo.Port()), clusterIPEndpoints, queriedLoadBalancers) if len(clusterIPEndpoints) > 0 { // If all endpoints are terminating, then no need to create Cluster IP LoadBalancer diff --git a/pkg/proxy/winkernel/proxier_test.go b/pkg/proxy/winkernel/proxier_test.go index f84526a412a..ed11ea61cbd 100644 --- a/pkg/proxy/winkernel/proxier_test.go +++ b/pkg/proxy/winkernel/proxier_test.go @@ -27,7 +27,7 @@ import ( "testing" "github.com/Microsoft/hnslib/hcn" - + "github.com/stretchr/testify/assert" v1 "k8s.io/api/core/v1" discovery "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -47,15 +47,19 @@ import ( const ( testNodeName = "test-node" testNetwork = "TestNetwork" - ipAddress = "10.0.0.1" prefixLen = 24 macAddress = "00-11-22-33-44-55" + macAddressLocal1 = "00-11-22-33-44-56" + macAddressLocal2 = "00-11-22-33-44-57" destinationPrefix = "192.168.2.0/24" providerAddress = "10.0.0.3" guid = "123ABC" + networkId = "123ABC" endpointGuid1 = "EPID-1" loadbalancerGuid1 = "LBID-1" - endpointLocal = "EP-LOCAL" + loadbalancerGuid2 = "LBID-2" + endpointLocal1 = "EP-LOCAL-1" + endpointLocal2 = "EP-LOCAL-2" endpointGw = "EP-GW" epIpAddressGw = "192.168.2.1" epMacAddressGw = "00-11-22-33-44-66" @@ -412,6 +416,193 @@ func TestDsrNotAppliedToClusterTrafficPolicy(t *testing.T) { } } +// TestClusterIPSvcWithITPLocal tests the following scenarios for a ClusterIP service with InternalTrafficPolicy=Local: +// 1. When a local endpoint is added to the service, the service should continue to use the local endpoints and existing loadbalancer. +// If no existing loadbalancer is present, a new loadbalancer should be created. +// 2. When one more local endpoint is added to the service, the service should delete existing loadbalancer and create a new loadbalancer. +// 3. When a remote endpoint is added to the service, the service should continue to use the local endpoints and existing loadbalancer, +// since it's a InternalTrafficPolicy=Local service. +func TestClusterIPSvcWithITPLocal(t *testing.T) { + proxier := NewFakeProxier(t, "testhost", netutils.ParseIPSloppy("10.0.0.1"), "L2Bridge", true) + if proxier == nil { + t.Fatal("Failed to create proxier") + } + + svcIP := "10.20.30.41" + svcPort := 80 + svcPortName := proxy.ServicePortName{ + NamespacedName: makeNSN("ns1", "svc1"), + Port: "p80", + Protocol: v1.ProtocolTCP, + } + + itpLocal := v1.ServiceInternalTrafficPolicyLocal + + makeServiceMap(proxier, + makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *v1.Service) { + svc.Spec.Type = v1.ServiceTypeClusterIP + svc.Spec.ClusterIP = svcIP + svc.Spec.InternalTrafficPolicy = &itpLocal // Setting the InternalTrafficPolicy to Local + svc.Spec.Ports = []v1.ServicePort{{ + Name: svcPortName.Port, + Port: int32(svcPort), + Protocol: v1.ProtocolTCP, + }} + }), + ) + + populateEndpointSlices(proxier, + makeTestEndpointSlice(svcPortName.Namespace, svcPortName.Name, 1, func(eps *discovery.EndpointSlice) { + eps.AddressType = discovery.AddressTypeIPv4 + eps.Endpoints = []discovery.Endpoint{ + { + Addresses: []string{epIpAddressLocal1}, // Local Endpoint 1 + }, + } + eps.Ports = []discovery.EndpointPort{{ + Name: ptr.To(svcPortName.Port), + Port: ptr.To(int32(svcPort)), + Protocol: ptr.To(v1.ProtocolTCP), + }} + }), + ) + + hcn := (proxier.hcn).(*fakehcn.HcnMock) + // Populating the endpoint to the cache, since it's a local endpoint and local endpoints are managed by CNI and not KubeProxy + // Populating here marks the endpoint to local + hcn.PopulateQueriedEndpoints(endpointLocal1, networkId, epIpAddressLocal1, macAddressLocal1, prefixLen) + + proxier.setInitialized(true) + + // Test 1: When a local endpoint is added to the service, the service should continue to use the local endpoints and existing loadbalancer. + // If no existing loadbalancer is present, a new loadbalancer should be created. + proxier.syncProxyRules() + + ep := proxier.endpointsMap[svcPortName][0] + epInfo, ok := ep.(*endpointInfo) + assert.True(t, ok, fmt.Sprintf("Failed to cast endpointInfo %q", svcPortName.String())) + assert.NotEmpty(t, epInfo.hnsID, fmt.Sprintf("Expected HNS ID to be set for endpoint %s, but got empty value", epIpAddressRemote)) + + svc := proxier.svcPortMap[svcPortName] + svcInfo, ok := svc.(*serviceInfo) + assert.True(t, ok, "Failed to cast serviceInfo %q", svcPortName.String()) + assert.Equal(t, svcInfo.hnsID, loadbalancerGuid1, fmt.Sprintf("%v does not match %v", svcInfo.hnsID, loadbalancerGuid1)) + lb, err := proxier.hcn.GetLoadBalancerByID(loadbalancerGuid1) + assert.Equal(t, nil, err, fmt.Sprintf("Failed to fetch loadbalancer: %s. Error: %v", loadbalancerGuid1, err)) + assert.NotNil(t, lb, "Loadbalancer object should not be nil") + + // Test 2: When one more local endpoint is added to the service, the service should delete existing loadbalancer and create a new loadbalancer. + + proxier.setInitialized(false) + + proxier.OnEndpointSliceUpdate( + makeTestEndpointSlice(svcPortName.Namespace, svcPortName.Name, 1, func(eps *discovery.EndpointSlice) { + eps.AddressType = discovery.AddressTypeIPv4 + eps.Endpoints = []discovery.Endpoint{{ + Addresses: []string{epIpAddressLocal1}, + }} + eps.Ports = []discovery.EndpointPort{{ + Name: ptr.To(svcPortName.Port), + Port: ptr.To(int32(svcPort)), + Protocol: ptr.To(v1.ProtocolTCP), + }} + }), + makeTestEndpointSlice(svcPortName.Namespace, svcPortName.Name, 1, func(eps *discovery.EndpointSlice) { + eps.AddressType = discovery.AddressTypeIPv4 + eps.Endpoints = []discovery.Endpoint{ + { + Addresses: []string{epIpAddressLocal1}, + }, + { + Addresses: []string{epIpAddressLocal2}, // Adding one more local endpoint + }, + } + eps.Ports = []discovery.EndpointPort{{ + Name: ptr.To(svcPortName.Port), + Port: ptr.To(int32(svcPort)), + Protocol: ptr.To(v1.ProtocolTCP), + }} + })) + + proxier.mu.Lock() + proxier.endpointSlicesSynced = true + proxier.mu.Unlock() + + proxier.setInitialized(true) + + // Creating the second local endpoint + hcn.PopulateQueriedEndpoints(endpointLocal2, networkId, epIpAddressLocal2, macAddressLocal2, prefixLen) + // Reinitiating the syncProxyRules to create new loadbalancer with the new local endpoint + proxier.syncProxyRules() + svc = proxier.svcPortMap[svcPortName] + svcInfo, ok = svc.(*serviceInfo) + assert.True(t, ok, "Failed to cast serviceInfo %q", svcPortName.String()) + assert.Equal(t, svcInfo.hnsID, loadbalancerGuid2, fmt.Sprintf("%v does not match %v", svcInfo.hnsID, loadbalancerGuid2)) + lb, err = proxier.hcn.GetLoadBalancerByID(loadbalancerGuid2) + assert.Equal(t, nil, err, fmt.Sprintf("Failed to fetch loadbalancer: %s. Error: %v", loadbalancerGuid2, err)) + assert.NotNil(t, lb, "Loadbalancer object should not be nil") + + lb, _ = proxier.hcn.GetLoadBalancerByID(loadbalancerGuid1) + assert.Nil(t, lb, fmt.Sprintf("Loadbalancer object should be nil: %s", loadbalancerGuid1)) + + // Test 3: When a remote endpoint is added to the service, the service should continue to use the local endpoints and existing loadbalancer, + // since it's a InternalTrafficPolicy=Local service. + + proxier.setInitialized(false) + + proxier.OnEndpointSliceUpdate( + makeTestEndpointSlice(svcPortName.Namespace, svcPortName.Name, 1, func(eps *discovery.EndpointSlice) { + eps.AddressType = discovery.AddressTypeIPv4 + eps.Endpoints = []discovery.Endpoint{ + { + Addresses: []string{epIpAddressLocal1}, + }, + { + Addresses: []string{epIpAddressLocal2}, + }, + } + eps.Ports = []discovery.EndpointPort{{ + Name: ptr.To(svcPortName.Port), + Port: ptr.To(int32(svcPort)), + Protocol: ptr.To(v1.ProtocolTCP), + }} + }), + makeTestEndpointSlice(svcPortName.Namespace, svcPortName.Name, 1, func(eps *discovery.EndpointSlice) { + eps.AddressType = discovery.AddressTypeIPv4 + eps.Endpoints = []discovery.Endpoint{ + { + Addresses: []string{epIpAddressLocal1}, + }, + { + Addresses: []string{epIpAddressLocal2}, // Adding one more local endpoint + }, + { + Addresses: []string{epIpAddressRemote}, // Adding one more remote endpoint to the slice + }, + } + eps.Ports = []discovery.EndpointPort{{ + Name: ptr.To(svcPortName.Port), + Port: ptr.To(int32(svcPort)), + Protocol: ptr.To(v1.ProtocolTCP), + }} + })) + + proxier.mu.Lock() + proxier.endpointSlicesSynced = true + proxier.mu.Unlock() + + proxier.setInitialized(true) + + proxier.syncProxyRules() + svc = proxier.svcPortMap[svcPortName] + svcInfo, ok = svc.(*serviceInfo) + assert.True(t, ok, "Failed to cast serviceInfo %q", svcPortName.String()) + assert.Equal(t, svcInfo.hnsID, loadbalancerGuid2, fmt.Sprintf("%v does not match %v", svcInfo.hnsID, loadbalancerGuid2)) + lb, err = proxier.hcn.GetLoadBalancerByID(loadbalancerGuid2) + assert.Equal(t, nil, err, fmt.Sprintf("Failed to fetch loadbalancer: %s. Error: %v", loadbalancerGuid2, err)) + assert.NotNil(t, lb, "Loadbalancer object should not be nil") +} + func TestSharedRemoteEndpointDelete(t *testing.T) { proxier := NewFakeProxier(t, testNodeName, netutils.ParseIPSloppy("10.0.0.1"), "L2Bridge", true) if proxier == nil { @@ -1197,7 +1388,7 @@ func TestCreateDsrLoadBalancer(t *testing.T) { hcn := (proxier.hcn).(*fakehcn.HcnMock) proxier.rootHnsEndpointName = endpointGw - hcn.PopulateQueriedEndpoints(endpointLocal, guid, epIpAddressRemote, macAddress, prefixLen) + hcn.PopulateQueriedEndpoints(endpointLocal1, guid, epIpAddressRemote, macAddress, prefixLen) hcn.PopulateQueriedEndpoints(endpointGw, guid, epIpAddressGw, epMacAddressGw, prefixLen) proxier.setInitialized(true) proxier.syncProxyRules()