github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 11:28:58 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #134032 from princepereira/automated-cherry-pick-of-#133953-upstream-release-1.33

Browse Source 
Automated cherry pick of #133953: Fix ClusterIP load balancer disappearing when InternalTrafficPolicy: Local is set.
This commit is contained in:
Kubernetes Prow Robot
2025-09-26 02:48:19 -07:00
committed by GitHub
parent 6c4310711b 22e1fc1a7b
commit 2b83f33f8a
4 changed files with 210 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
pkg/proxy/winkernel/hns.go
View File

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

2
pkg/proxy/winkernel/hns_test.go
View File

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

2
pkg/proxy/winkernel/proxier.go
View File

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

199
pkg/proxy/winkernel/proxier_test.go
View File

@@ -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()