github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-08 11:38:15 +00:00
Code Issues Projects Releases Wiki Activity

Fix "Endpoint" vs "Endpoints" in proxy type names

Browse Source 
The use of "Endpoint" vs "Endpoints" in these type names is tricky
because it doesn't always make sense to use the same singular/plural
convention as the corresonding service-related types, since often the
service-related type is referring to a single service while the
endpoint-related type is referring to multiple endpoint IPs.

The "endpointsInfo" types in the iptables and winkernel proxiers are
now "endpointInfo" because they describe a single endpoint IP (and
wrap proxy.BaseEndpointInfo).

"UpdateEndpointMapResult" is now "UpdateEndpointsMapResult", because
it is the result of EndpointsMap.Update (and it's clearly correct for
EndpointsMap to have plural "Endpoints" because it's a map to an array
of proxy.Endpoint objects.)

"EndpointChangeTracker" is now "EndpointsChangeTracker" because it
tracks changes to the full set of endpoints for a particular service
(and the new name matches the existing "endpointsChange" type and
"Proxier.endpointsChanges" fields.)
This commit is contained in:
Dan Winship 2023-10-09 17:21:12 -04:00
parent 59424358cc
commit 6c395eb098
12 changed files with 236 additions and 236 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
pkg/proxy/conntrack/cleanup.go
View File

@ -27,23 +27,23 @@ import (
// CleanStaleEntries takes care of flushing stale conntrack entries for services and endpoints. // CleanStaleEntries takes care of flushing stale conntrack entries for services and endpoints.
func CleanStaleEntries(isIPv6 bool, exec utilexec.Interface, svcPortMap proxy.ServicePortMap, func CleanStaleEntries(isIPv6 bool, exec utilexec.Interface, svcPortMap proxy.ServicePortMap,
serviceUpdateResult proxy.UpdateServiceMapResult, endpointUpdateResult proxy.UpdateEndpointMapResult) { serviceUpdateResult proxy.UpdateServiceMapResult, endpointsUpdateResult proxy.UpdateEndpointsMapResult) {
deleteStaleServiceConntrackEntries(isIPv6, exec, svcPortMap, serviceUpdateResult, endpointUpdateResult) deleteStaleServiceConntrackEntries(isIPv6, exec, svcPortMap, serviceUpdateResult, endpointsUpdateResult)
deleteStaleEndpointConntrackEntries(exec, svcPortMap, endpointUpdateResult) deleteStaleEndpointConntrackEntries(exec, svcPortMap, endpointsUpdateResult)
} }
// deleteStaleServiceConntrackEntries takes care of flushing stale conntrack entries related // deleteStaleServiceConntrackEntries takes care of flushing stale conntrack entries related
// to UDP Service IPs. When a service has no endpoints and we drop traffic to it, conntrack // to UDP Service IPs. When a service has no endpoints and we drop traffic to it, conntrack
// may create "black hole" entries for that IP+port. When the service gets endpoints we // may create "black hole" entries for that IP+port. When the service gets endpoints we
// need to delete those entries so further traffic doesn't get dropped. // need to delete those entries so further traffic doesn't get dropped.
func deleteStaleServiceConntrackEntries(isIPv6 bool, exec utilexec.Interface, svcPortMap proxy.ServicePortMap, serviceUpdateResult proxy.UpdateServiceMapResult, endpointUpdateResult proxy.UpdateEndpointMapResult) { func deleteStaleServiceConntrackEntries(isIPv6 bool, exec utilexec.Interface, svcPortMap proxy.ServicePortMap, serviceUpdateResult proxy.UpdateServiceMapResult, endpointsUpdateResult proxy.UpdateEndpointsMapResult) {
conntrackCleanupServiceIPs := serviceUpdateResult.DeletedUDPClusterIPs conntrackCleanupServiceIPs := serviceUpdateResult.DeletedUDPClusterIPs
conntrackCleanupServiceNodePorts := sets.New[int]() conntrackCleanupServiceNodePorts := sets.New[int]()
// merge newly active services gathered from updateEndpointsMap // merge newly active services gathered from endpointsUpdateResult
// a UDP service that changes from 0 to non-0 endpoints is newly active. // a UDP service that changes from 0 to non-0 endpoints is newly active.
for _, svcPortName := range endpointUpdateResult.NewlyActiveUDPServices { for _, svcPortName := range endpointsUpdateResult.NewlyActiveUDPServices {
if svcInfo, ok := svcPortMap[svcPortName]; ok { if svcInfo, ok := svcPortMap[svcPortName]; ok {
klog.V(4).InfoS("Newly-active UDP service may have stale conntrack entries", "servicePortName", svcPortName) klog.V(4).InfoS("Newly-active UDP service may have stale conntrack entries", "servicePortName", svcPortName)
conntrackCleanupServiceIPs.Insert(svcInfo.ClusterIP().String()) conntrackCleanupServiceIPs.Insert(svcInfo.ClusterIP().String())
@ -78,8 +78,8 @@ func deleteStaleServiceConntrackEntries(isIPv6 bool, exec utilexec.Interface, sv
// deleteStaleEndpointConntrackEntries takes care of flushing stale conntrack entries related // deleteStaleEndpointConntrackEntries takes care of flushing stale conntrack entries related
// to UDP endpoints. After a UDP endpoint is removed we must flush any conntrack entries // to UDP endpoints. After a UDP endpoint is removed we must flush any conntrack entries
// for it so that if the same client keeps sending, the packets will get routed to a new endpoint. // for it so that if the same client keeps sending, the packets will get routed to a new endpoint.
func deleteStaleEndpointConntrackEntries(exec utilexec.Interface, svcPortMap proxy.ServicePortMap, endpointUpdateResult proxy.UpdateEndpointMapResult) { func deleteStaleEndpointConntrackEntries(exec utilexec.Interface, svcPortMap proxy.ServicePortMap, endpointsUpdateResult proxy.UpdateEndpointsMapResult) {
for _, epSvcPair := range endpointUpdateResult.DeletedUDPEndpoints { for _, epSvcPair := range endpointsUpdateResult.DeletedUDPEndpoints {
if svcInfo, ok := svcPortMap[epSvcPair.ServicePortName]; ok { if svcInfo, ok := svcPortMap[epSvcPair.ServicePortName]; ok {
endpointIP := proxyutil.IPPart(epSvcPair.Endpoint) endpointIP := proxyutil.IPPart(epSvcPair.Endpoint)
nodePort := svcInfo.NodePort() nodePort := svcInfo.NodePort()

32
pkg/proxy/endpoints.go
View File

@ -147,9 +147,9 @@ type makeEndpointFunc func(info *BaseEndpointInfo, svcPortName *ServicePortName)
// EndpointsMap's but just use the changes for any Proxier specific cleanup. // EndpointsMap's but just use the changes for any Proxier specific cleanup.
type processEndpointsMapChangeFunc func(oldEndpointsMap, newEndpointsMap EndpointsMap) type processEndpointsMapChangeFunc func(oldEndpointsMap, newEndpointsMap EndpointsMap)
// EndpointChangeTracker carries state about uncommitted changes to an arbitrary number of // EndpointsChangeTracker carries state about uncommitted changes to an arbitrary number of
// Endpoints, keyed by their namespace and name. // Endpoints, keyed by their namespace and name.
type EndpointChangeTracker struct { type EndpointsChangeTracker struct {
// lock protects lastChangeTriggerTimes // lock protects lastChangeTriggerTimes
lock sync.Mutex lock sync.Mutex
@ -159,16 +159,16 @@ type EndpointChangeTracker struct {
// Map from the Endpoints namespaced-name to the times of the triggers that caused the endpoints // Map from the Endpoints namespaced-name to the times of the triggers that caused the endpoints
// object to change. Used to calculate the network-programming-latency. // object to change. Used to calculate the network-programming-latency.
lastChangeTriggerTimes map[types.NamespacedName][]time.Time lastChangeTriggerTimes map[types.NamespacedName][]time.Time
// record the time when the endpointChangeTracker was created so we can ignore the endpoints // record the time when the endpointsChangeTracker was created so we can ignore the endpoints
// that were generated before, because we can't estimate the network-programming-latency on those. // that were generated before, because we can't estimate the network-programming-latency on those.
// This is specially problematic on restarts, because we process all the endpoints that may have been // This is specially problematic on restarts, because we process all the endpoints that may have been
// created hours or days before. // created hours or days before.
trackerStartTime time.Time trackerStartTime time.Time
} }
// NewEndpointChangeTracker initializes an EndpointsChangeMap // NewEndpointsChangeTracker initializes an EndpointsChangeTracker
func NewEndpointChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc, ipFamily v1.IPFamily, recorder events.EventRecorder, processEndpointsMapChange processEndpointsMapChangeFunc) *EndpointChangeTracker { func NewEndpointsChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc, ipFamily v1.IPFamily, recorder events.EventRecorder, processEndpointsMapChange processEndpointsMapChangeFunc) *EndpointsChangeTracker {
return &EndpointChangeTracker{ return &EndpointsChangeTracker{
lastChangeTriggerTimes: make(map[types.NamespacedName][]time.Time), lastChangeTriggerTimes: make(map[types.NamespacedName][]time.Time),
trackerStartTime: time.Now(), trackerStartTime: time.Now(),
processEndpointsMapChange: processEndpointsMapChange, processEndpointsMapChange: processEndpointsMapChange,
@ -177,9 +177,9 @@ func NewEndpointChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc
} }
// EndpointSliceUpdate updates given service's endpoints change map based on the <previous, current> endpoints pair. // EndpointSliceUpdate updates given service's endpoints change map based on the <previous, current> endpoints pair.
// It returns true if items changed, otherwise return false. Will add/update/delete items of EndpointsChangeMap. // It returns true if items changed, otherwise return false. Will add/update/delete items of EndpointsChangeTracker.
// If removeSlice is true, slice will be removed, otherwise it will be added or updated. // If removeSlice is true, slice will be removed, otherwise it will be added or updated.
func (ect *EndpointChangeTracker) EndpointSliceUpdate(endpointSlice *discovery.EndpointSlice, removeSlice bool) bool { func (ect *EndpointsChangeTracker) EndpointSliceUpdate(endpointSlice *discovery.EndpointSlice, removeSlice bool) bool {
if !supportedEndpointSliceAddressTypes.Has(string(endpointSlice.AddressType)) { if !supportedEndpointSliceAddressTypes.Has(string(endpointSlice.AddressType)) {
klog.V(4).InfoS("EndpointSlice address type not supported by kube-proxy", "addressType", endpointSlice.AddressType) klog.V(4).InfoS("EndpointSlice address type not supported by kube-proxy", "addressType", endpointSlice.AddressType)
return false return false
@ -225,13 +225,13 @@ func (ect *EndpointChangeTracker) EndpointSliceUpdate(endpointSlice *discovery.E
// PendingChanges returns a set whose keys are the names of the services whose endpoints // PendingChanges returns a set whose keys are the names of the services whose endpoints
// have changed since the last time ect was used to update an EndpointsMap. (You must call // have changed since the last time ect was used to update an EndpointsMap. (You must call
// this _before_ calling em.Update(ect).) // this _before_ calling em.Update(ect).)
func (ect *EndpointChangeTracker) PendingChanges() sets.Set[string] { func (ect *EndpointsChangeTracker) PendingChanges() sets.Set[string] {
return ect.endpointSliceCache.pendingChanges() return ect.endpointSliceCache.pendingChanges()
} }
// checkoutChanges returns a list of pending endpointsChanges and marks them as // checkoutChanges returns a list of pending endpointsChanges and marks them as
// applied. // applied.
func (ect *EndpointChangeTracker) checkoutChanges() []*endpointsChange { func (ect *EndpointsChangeTracker) checkoutChanges() []*endpointsChange {
metrics.EndpointChangesPending.Set(0) metrics.EndpointChangesPending.Set(0)
return ect.endpointSliceCache.checkoutChanges() return ect.endpointSliceCache.checkoutChanges()
@ -239,7 +239,7 @@ func (ect *EndpointChangeTracker) checkoutChanges() []*endpointsChange {
// checkoutTriggerTimes applies the locally cached trigger times to a map of // checkoutTriggerTimes applies the locally cached trigger times to a map of
// trigger times that have been passed in and empties the local cache. // trigger times that have been passed in and empties the local cache.
func (ect *EndpointChangeTracker) checkoutTriggerTimes(lastChangeTriggerTimes *map[types.NamespacedName][]time.Time) { func (ect *EndpointsChangeTracker) checkoutTriggerTimes(lastChangeTriggerTimes *map[types.NamespacedName][]time.Time) {
ect.lock.Lock() ect.lock.Lock()
defer ect.lock.Unlock() defer ect.lock.Unlock()
@ -284,8 +284,8 @@ type endpointsChange struct {
current EndpointsMap current EndpointsMap
} }
// UpdateEndpointMapResult is the updated results after applying endpoints changes. // UpdateEndpointsMapResult is the updated results after applying endpoints changes.
type UpdateEndpointMapResult struct { type UpdateEndpointsMapResult struct {
// DeletedUDPEndpoints identifies UDP endpoints that have just been deleted. // DeletedUDPEndpoints identifies UDP endpoints that have just been deleted.
// Existing conntrack NAT entries pointing to these endpoints must be deleted to // Existing conntrack NAT entries pointing to these endpoints must be deleted to
// ensure that no further traffic for the Service gets delivered to them. // ensure that no further traffic for the Service gets delivered to them.
@ -304,7 +304,7 @@ type UpdateEndpointMapResult struct {
} }
// Update updates endpointsMap base on the given changes. // Update updates endpointsMap base on the given changes.
func (em EndpointsMap) Update(changes *EndpointChangeTracker) (result UpdateEndpointMapResult) { func (em EndpointsMap) Update(changes *EndpointsChangeTracker) (result UpdateEndpointsMapResult) {
result.DeletedUDPEndpoints = make([]ServiceEndpoint, 0) result.DeletedUDPEndpoints = make([]ServiceEndpoint, 0)
result.NewlyActiveUDPServices = make([]ServicePortName, 0) result.NewlyActiveUDPServices = make([]ServicePortName, 0)
result.LastChangeTriggerTimes = make(map[types.NamespacedName][]time.Time) result.LastChangeTriggerTimes = make(map[types.NamespacedName][]time.Time)
@ -321,7 +321,7 @@ type EndpointsMap map[ServicePortName][]Endpoint
// and clear the changes map. In addition it returns (via argument) and resets the // and clear the changes map. In addition it returns (via argument) and resets the
// lastChangeTriggerTimes for all endpoints that were changed and will result in syncing // lastChangeTriggerTimes for all endpoints that were changed and will result in syncing
// the proxy rules. apply triggers processEndpointsMapChange on every change. // the proxy rules. apply triggers processEndpointsMapChange on every change.
func (em EndpointsMap) apply(ect *EndpointChangeTracker, deletedUDPEndpoints *[]ServiceEndpoint, func (em EndpointsMap) apply(ect *EndpointsChangeTracker, deletedUDPEndpoints *[]ServiceEndpoint,
newlyActiveUDPServices *[]ServicePortName, lastChangeTriggerTimes *map[types.NamespacedName][]time.Time) { newlyActiveUDPServices *[]ServicePortName, lastChangeTriggerTimes *map[types.NamespacedName][]time.Time) {
if ect == nil { if ect == nil {
return return
@ -396,7 +396,7 @@ func (em EndpointsMap) LocalReadyEndpoints() map[types.NamespacedName]int {
} }
// detectStaleConntrackEntries detects services that may be associated with stale conntrack entries. // detectStaleConntrackEntries detects services that may be associated with stale conntrack entries.
// (See UpdateEndpointMapResult.DeletedUDPEndpoints and .NewlyActiveUDPServices.) // (See UpdateEndpointsMapResult.DeletedUDPEndpoints and .NewlyActiveUDPServices.)
func detectStaleConntrackEntries(oldEndpointsMap, newEndpointsMap EndpointsMap, deletedUDPEndpoints *[]ServiceEndpoint, newlyActiveUDPServices *[]ServicePortName) { func detectStaleConntrackEntries(oldEndpointsMap, newEndpointsMap EndpointsMap, deletedUDPEndpoints *[]ServiceEndpoint, newlyActiveUDPServices *[]ServicePortName) {
// Find the UDP endpoints that we were sending traffic to in oldEndpointsMap, but // Find the UDP endpoints that we were sending traffic to in oldEndpointsMap, but
// are no longer sending to newEndpointsMap. The proxier should make sure that // are no longer sending to newEndpointsMap. The proxier should make sure that

130
pkg/proxy/endpoints_test.go
View File

@ -1340,7 +1340,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
testCases := map[string]struct { testCases := map[string]struct {
startingSlices []*discovery.EndpointSlice startingSlices []*discovery.EndpointSlice
endpointChangeTracker *EndpointChangeTracker endpointsChangeTracker *EndpointsChangeTracker
namespacedName types.NamespacedName namespacedName types.NamespacedName
paramEndpointSlice *discovery.EndpointSlice paramEndpointSlice *discovery.EndpointSlice
paramRemoveSlice bool paramRemoveSlice bool
@ -1350,12 +1350,12 @@ func TestEndpointSliceUpdate(t *testing.T) {
}{ }{
// test starting from an empty state // test starting from an empty state
"add a simple slice that doesn't already exist": { "add a simple slice that doesn't already exist": {
startingSlices: []*discovery.EndpointSlice{}, startingSlices: []*discovery.EndpointSlice{},
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: false, paramRemoveSlice: false,
expectedReturnVal: true, expectedReturnVal: true,
expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{ expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{
makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): { makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): {
&BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: false, Ready: true, Serving: true, Terminating: false}, &BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: false, Ready: true, Serving: true, Terminating: false},
@ -1375,7 +1375,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
startingSlices: []*discovery.EndpointSlice{ startingSlices: []*discovery.EndpointSlice{
generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: false, paramRemoveSlice: false,
@ -1388,7 +1388,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
startingSlices: []*discovery.EndpointSlice{ startingSlices: []*discovery.EndpointSlice{
generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: fqdnSlice, paramEndpointSlice: fqdnSlice,
paramRemoveSlice: false, paramRemoveSlice: false,
@ -1402,11 +1402,11 @@ func TestEndpointSliceUpdate(t *testing.T) {
generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
generateEndpointSlice("svc1", "ns1", 2, 2, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 2, 2, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: false, paramRemoveSlice: false,
expectedReturnVal: true, expectedReturnVal: true,
expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{ expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{
makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): { makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): {
&BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false}, &BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false},
@ -1435,11 +1435,11 @@ func TestEndpointSliceUpdate(t *testing.T) {
generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
generateEndpointSlice("svc1", "ns1", 2, 2, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 2, 2, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSliceWithOffset("svc1", "ns1", 3, 1, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80)}), paramEndpointSlice: generateEndpointSliceWithOffset("svc1", "ns1", 3, 1, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80)}),
paramRemoveSlice: false, paramRemoveSlice: false,
expectedReturnVal: true, expectedReturnVal: true,
expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{ expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{
makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): { makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): {
&BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false}, &BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false},
@ -1466,11 +1466,11 @@ func TestEndpointSliceUpdate(t *testing.T) {
generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
generateEndpointSlice("svc1", "ns1", 2, 2, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 2, 2, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: true, paramRemoveSlice: true,
expectedReturnVal: true, expectedReturnVal: true,
expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{ expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{
makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): { makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): {
&BaseEndpointInfo{Endpoint: "10.0.2.1:80", IsLocal: false, Ready: true, Serving: true, Terminating: false}, &BaseEndpointInfo{Endpoint: "10.0.2.1:80", IsLocal: false, Ready: true, Serving: true, Terminating: false},
@ -1489,7 +1489,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
generateEndpointSlice("svc1", "ns1", 1, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
generateEndpointSlice("svc1", "ns1", 2, 2, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 2, 2, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 3, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 3, 5, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: true, paramRemoveSlice: true,
@ -1502,11 +1502,11 @@ func TestEndpointSliceUpdate(t *testing.T) {
startingSlices: []*discovery.EndpointSlice{ startingSlices: []*discovery.EndpointSlice{
generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 3, 999, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 1, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 1, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: false, paramRemoveSlice: false,
expectedReturnVal: true, expectedReturnVal: true,
expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{ expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{
makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): { makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): {
&BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: false, Serving: false, Terminating: false}, &BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: false, Serving: false, Terminating: false},
@ -1526,11 +1526,11 @@ func TestEndpointSliceUpdate(t *testing.T) {
startingSlices: []*discovery.EndpointSlice{ startingSlices: []*discovery.EndpointSlice{
generateEndpointSlice("svc1", "ns1", 1, 2, 1, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 2, 1, 999, []string{"host1", "host2"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 2, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 2, 999, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: false, paramRemoveSlice: false,
expectedReturnVal: true, expectedReturnVal: true,
expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{ expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{
makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): { makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): {
&BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false}, &BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false},
@ -1549,11 +1549,11 @@ func TestEndpointSliceUpdate(t *testing.T) {
generateEndpointSlice("svc1", "ns1", 1, 3, 2, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 3, 2, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
generateEndpointSlice("svc1", "ns1", 2, 2, 2, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 2, 2, 2, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 3, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 3, 999, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: false, paramRemoveSlice: false,
expectedReturnVal: true, expectedReturnVal: true,
expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{ expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{
makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): { makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): {
&BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false}, &BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false},
@ -1578,11 +1578,11 @@ func TestEndpointSliceUpdate(t *testing.T) {
generateEndpointSlice("svc1", "ns1", 1, 3, 2, 2, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 1, 3, 2, 2, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
generateEndpointSlice("svc1", "ns1", 2, 2, 2, 2, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), generateEndpointSlice("svc1", "ns1", 2, 2, 2, 2, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
}, },
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("host1", nil, v1.IPv4Protocol, nil, nil),
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"}, namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 3, 2, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}), paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 3, 2, []string{"host1"}, []*int32{pointer.Int32(80), pointer.Int32(443)}),
paramRemoveSlice: false, paramRemoveSlice: false,
expectedReturnVal: true, expectedReturnVal: true,
expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{ expectedCurrentChange: map[ServicePortName][]*BaseEndpointInfo{
makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): { makeServicePortName("ns1", "svc1", "port-0", v1.ProtocolTCP): {
&BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false}, &BaseEndpointInfo{Endpoint: "10.0.1.1:80", IsLocal: true, Ready: true, Serving: true, Terminating: false},
@ -1605,19 +1605,19 @@ func TestEndpointSliceUpdate(t *testing.T) {
for name, tc := range testCases { for name, tc := range testCases {
t.Run(name, func(t *testing.T) { t.Run(name, func(t *testing.T) {
initializeCache(tc.endpointChangeTracker.endpointSliceCache, tc.startingSlices) initializeCache(tc.endpointsChangeTracker.endpointSliceCache, tc.startingSlices)
got := tc.endpointChangeTracker.EndpointSliceUpdate(tc.paramEndpointSlice, tc.paramRemoveSlice) got := tc.endpointsChangeTracker.EndpointSliceUpdate(tc.paramEndpointSlice, tc.paramRemoveSlice)
if !reflect.DeepEqual(got, tc.expectedReturnVal) { if !reflect.DeepEqual(got, tc.expectedReturnVal) {
t.Errorf("EndpointSliceUpdate return value got: %v, want %v", got, tc.expectedReturnVal) t.Errorf("EndpointSliceUpdate return value got: %v, want %v", got, tc.expectedReturnVal)
} }
pendingChanges := tc.endpointChangeTracker.PendingChanges() pendingChanges := tc.endpointsChangeTracker.PendingChanges()
if !pendingChanges.Equal(tc.expectedChangedEndpoints) { if !pendingChanges.Equal(tc.expectedChangedEndpoints) {
t.Errorf("expected changed endpoints %q, got %q", tc.expectedChangedEndpoints.UnsortedList(), pendingChanges.UnsortedList()) t.Errorf("expected changed endpoints %q, got %q", tc.expectedChangedEndpoints.UnsortedList(), pendingChanges.UnsortedList())
} }
changes := tc.endpointChangeTracker.checkoutChanges() changes := tc.endpointsChangeTracker.checkoutChanges()
if tc.expectedCurrentChange == nil { if tc.expectedCurrentChange == nil {
if len(changes) != 0 { if len(changes) != 0 {
t.Errorf("Expected %s to have no changes", tc.namespacedName) t.Errorf("Expected %s to have no changes", tc.namespacedName)
@ -1637,20 +1637,20 @@ func TestCheckoutChanges(t *testing.T) {
svcPortName1 := ServicePortName{types.NamespacedName{Namespace: "ns1", Name: "svc1"}, "port-1", v1.ProtocolTCP} svcPortName1 := ServicePortName{types.NamespacedName{Namespace: "ns1", Name: "svc1"}, "port-1", v1.ProtocolTCP}
testCases := map[string]struct { testCases := map[string]struct {
endpointChangeTracker *EndpointChangeTracker endpointsChangeTracker *EndpointsChangeTracker
expectedChanges []*endpointsChange expectedChanges []*endpointsChange
items map[types.NamespacedName]*endpointsChange items map[types.NamespacedName]*endpointsChange
appliedSlices []*discovery.EndpointSlice appliedSlices []*discovery.EndpointSlice
pendingSlices []*discovery.EndpointSlice pendingSlices []*discovery.EndpointSlice
}{ }{
"empty slices": { "empty slices": {
endpointChangeTracker: NewEndpointChangeTracker("", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("", nil, v1.IPv4Protocol, nil, nil),
expectedChanges: []*endpointsChange{}, expectedChanges: []*endpointsChange{},
appliedSlices: []*discovery.EndpointSlice{}, appliedSlices: []*discovery.EndpointSlice{},
pendingSlices: []*discovery.EndpointSlice{}, pendingSlices: []*discovery.EndpointSlice{},
}, },
"adding initial slice": { "adding initial slice": {
endpointChangeTracker: NewEndpointChangeTracker("", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("", nil, v1.IPv4Protocol, nil, nil),
expectedChanges: []*endpointsChange{{ expectedChanges: []*endpointsChange{{
previous: EndpointsMap{}, previous: EndpointsMap{},
current: EndpointsMap{ current: EndpointsMap{
@ -1663,7 +1663,7 @@ func TestCheckoutChanges(t *testing.T) {
}, },
}, },
"removing port in update": { "removing port in update": {
endpointChangeTracker: NewEndpointChangeTracker("", nil, v1.IPv4Protocol, nil, nil), endpointsChangeTracker: NewEndpointsChangeTracker("", nil, v1.IPv4Protocol, nil, nil),
expectedChanges: []*endpointsChange{{ expectedChanges: []*endpointsChange{{
previous: EndpointsMap{ previous: EndpointsMap{
svcPortName0: []Endpoint{newTestEp("10.0.1.1:80", "host1", true, true, false), newTestEp("10.0.1.2:80", "host1", true, true, false), newTestEp("10.0.1.3:80", "host1", false, false, false)}, svcPortName0: []Endpoint{newTestEp("10.0.1.1:80", "host1", true, true, false), newTestEp("10.0.1.2:80", "host1", true, true, false), newTestEp("10.0.1.3:80", "host1", false, false, false)},
@ -1685,13 +1685,13 @@ func TestCheckoutChanges(t *testing.T) {
for name, tc := range testCases { for name, tc := range testCases {
t.Run(name, func(t *testing.T) { t.Run(name, func(t *testing.T) {
for _, slice := range tc.appliedSlices { for _, slice := range tc.appliedSlices {
tc.endpointChangeTracker.EndpointSliceUpdate(slice, false) tc.endpointsChangeTracker.EndpointSliceUpdate(slice, false)
} }
tc.endpointChangeTracker.checkoutChanges() tc.endpointsChangeTracker.checkoutChanges()
for _, slice := range tc.pendingSlices { for _, slice := range tc.pendingSlices {
tc.endpointChangeTracker.EndpointSliceUpdate(slice, false) tc.endpointsChangeTracker.EndpointSliceUpdate(slice, false)
} }
changes := tc.endpointChangeTracker.checkoutChanges() changes := tc.endpointsChangeTracker.checkoutChanges()
if len(tc.expectedChanges) != len(changes) { if len(tc.expectedChanges) != len(changes) {
t.Fatalf("Expected %d changes, got %d", len(tc.expectedChanges), len(changes)) t.Fatalf("Expected %d changes, got %d", len(tc.expectedChanges), len(changes))
@ -1730,7 +1730,7 @@ func compareEndpointsMapsStr(t *testing.T, newMap EndpointsMap, expected map[Ser
for i := range expected[x] { for i := range expected[x] {
newEp, ok := newMap[x][i].(*BaseEndpointInfo) newEp, ok := newMap[x][i].(*BaseEndpointInfo)
if !ok { if !ok {
t.Fatalf("Failed to cast endpointsInfo") t.Fatalf("Failed to cast endpointInfo")
} }
if !endpointEqual(newEp, expected[x][i]) { if !endpointEqual(newEp, expected[x][i]) {
t.Fatalf("expected new[%v][%d] to be %v, got %v"+ t.Fatalf("expected new[%v][%d] to be %v, got %v"+

20
pkg/proxy/iptables/proxier.go
View File

@ -122,15 +122,15 @@ func newServiceInfo(port *v1.ServicePort, service *v1.Service, bsvcPortInfo *pro
} }
// internal struct for endpoints information // internal struct for endpoints information
type endpointsInfo struct { type endpointInfo struct {
*proxy.BaseEndpointInfo *proxy.BaseEndpointInfo
ChainName utiliptables.Chain ChainName utiliptables.Chain
} }
// returns a new proxy.Endpoint which abstracts a endpointsInfo // returns a new proxy.Endpoint which abstracts a endpointInfo
func newEndpointInfo(baseInfo *proxy.BaseEndpointInfo, svcPortName *proxy.ServicePortName) proxy.Endpoint { func newEndpointInfo(baseInfo *proxy.BaseEndpointInfo, svcPortName *proxy.ServicePortName) proxy.Endpoint {
return &endpointsInfo{ return &endpointInfo{
BaseEndpointInfo: baseInfo, BaseEndpointInfo: baseInfo,
ChainName: servicePortEndpointChainName(svcPortName.String(), strings.ToLower(string(svcPortName.Protocol)), baseInfo.Endpoint), ChainName: servicePortEndpointChainName(svcPortName.String(), strings.ToLower(string(svcPortName.Protocol)), baseInfo.Endpoint),
} }
@ -143,7 +143,7 @@ type Proxier struct {
// services that happened since iptables was synced. For a single object, // services that happened since iptables was synced. For a single object,
// changes are accumulated, i.e. previous is state from before all of them, // changes are accumulated, i.e. previous is state from before all of them,
// current is state after applying all of those. // current is state after applying all of those.
endpointsChanges *proxy.EndpointChangeTracker endpointsChanges *proxy.EndpointsChangeTracker
serviceChanges *proxy.ServiceChangeTracker serviceChanges *proxy.ServiceChangeTracker
mu sync.Mutex // protects the following fields mu sync.Mutex // protects the following fields
@ -265,7 +265,7 @@ func NewProxier(ipFamily v1.IPFamily,
svcPortMap: make(proxy.ServicePortMap), svcPortMap: make(proxy.ServicePortMap),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, ipFamily, recorder, nil), endpointsChanges: proxy.NewEndpointsChangeTracker(hostname, newEndpointInfo, ipFamily, recorder, nil),
needFullSync: true, needFullSync: true,
syncPeriod: syncPeriod, syncPeriod: syncPeriod,
iptables: ipt, iptables: ipt,
@ -952,7 +952,7 @@ func (proxier *Proxier) syncProxyRules() {
// Note the endpoint chains that will be used // Note the endpoint chains that will be used
for _, ep := range allLocallyReachableEndpoints { for _, ep := range allLocallyReachableEndpoints {
if epInfo, ok := ep.(*endpointsInfo); ok { if epInfo, ok := ep.(*endpointInfo); ok {
activeNATChains[epInfo.ChainName] = true activeNATChains[epInfo.ChainName] = true
} }
} }
@ -1345,9 +1345,9 @@ func (proxier *Proxier) syncProxyRules() {
// Generate the per-endpoint chains. // Generate the per-endpoint chains.
for _, ep := range allLocallyReachableEndpoints { for _, ep := range allLocallyReachableEndpoints {
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if !ok { if !ok {
klog.ErrorS(nil, "Failed to cast endpointsInfo", "endpointsInfo", ep) klog.ErrorS(nil, "Failed to cast endpointInfo", "endpointInfo", ep)
continue continue
} }
@ -1556,7 +1556,7 @@ func (proxier *Proxier) writeServiceToEndpointRules(natRules proxyutil.LineBuffe
// First write session affinity rules, if applicable. // First write session affinity rules, if applicable.
if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP { if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
for _, ep := range endpoints { for _, ep := range endpoints {
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if !ok { if !ok {
continue continue
} }
@ -1578,7 +1578,7 @@ func (proxier *Proxier) writeServiceToEndpointRules(natRules proxyutil.LineBuffe
// Now write loadbalancing rules. // Now write loadbalancing rules.
numEndpoints := len(endpoints) numEndpoints := len(endpoints)
for i, ep := range endpoints { for i, ep := range endpoints {
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if !ok { if !ok {
continue continue
} }

72
pkg/proxy/iptables/proxier_test.go
View File

@ -330,7 +330,7 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier {
svcPortMap: make(proxy.ServicePortMap), svcPortMap: make(proxy.ServicePortMap),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipfamily, nil, nil), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipfamily, nil, nil),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, newEndpointInfo, ipfamily, nil, nil), endpointsChanges: proxy.NewEndpointsChangeTracker(testHostname, newEndpointInfo, ipfamily, nil, nil),
needFullSync: true, needFullSync: true,
iptables: ipt, iptables: ipt,
masqueradeMark: "0x4000", masqueradeMark: "0x4000",
@ -3371,7 +3371,7 @@ func makeServiceMap(proxier *Proxier, allServices ...*v1.Service) {
proxier.servicesSynced = true proxier.servicesSynced = true
} }
func compareEndpointsMapsExceptChainName(t *testing.T, tci int, newMap proxy.EndpointsMap, expected map[proxy.ServicePortName][]*endpointsInfo) { func compareEndpointsMapsExceptChainName(t *testing.T, tci int, newMap proxy.EndpointsMap, expected map[proxy.ServicePortName][]*endpointInfo) {
if len(newMap) != len(expected) { if len(newMap) != len(expected) {
t.Errorf("[%d] expected %d results, got %d: %v", tci, len(expected), len(newMap), newMap) t.Errorf("[%d] expected %d results, got %d: %v", tci, len(expected), len(newMap), newMap)
} }
@ -3380,9 +3380,9 @@ func compareEndpointsMapsExceptChainName(t *testing.T, tci int, newMap proxy.End
t.Errorf("[%d] expected %d endpoints for %v, got %d", tci, len(expected[x]), x, len(newMap[x])) t.Errorf("[%d] expected %d endpoints for %v, got %d", tci, len(expected[x]), x, len(newMap[x]))
} else { } else {
for i := range expected[x] { for i := range expected[x] {
newEp, ok := newMap[x][i].(*endpointsInfo) newEp, ok := newMap[x][i].(*endpointInfo)
if !ok { if !ok {
t.Errorf("Failed to cast endpointsInfo") t.Errorf("Failed to cast endpointInfo")
continue continue
} }
if newEp.Endpoint != expected[x][i].Endpoint || if newEp.Endpoint != expected[x][i].Endpoint ||
@ -3731,16 +3731,16 @@ func TestUpdateEndpointsMap(t *testing.T) {
name string name string
previousEndpoints []*discovery.EndpointSlice previousEndpoints []*discovery.EndpointSlice
currentEndpoints []*discovery.EndpointSlice currentEndpoints []*discovery.EndpointSlice
oldEndpoints map[proxy.ServicePortName][]*endpointsInfo oldEndpoints map[proxy.ServicePortName][]*endpointInfo
expectedResult map[proxy.ServicePortName][]*endpointsInfo expectedResult map[proxy.ServicePortName][]*endpointInfo
expectedDeletedUDPEndpoints []proxy.ServiceEndpoint expectedDeletedUDPEndpoints []proxy.ServiceEndpoint
expectedNewlyActiveUDPServices map[proxy.ServicePortName]bool expectedNewlyActiveUDPServices map[proxy.ServicePortName]bool
expectedLocalEndpoints map[types.NamespacedName]int expectedLocalEndpoints map[types.NamespacedName]int
}{{ }{{
// Case[0]: nothing // Case[0]: nothing
name: "nothing", name: "nothing",
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{}, oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{},
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{}, expectedResult: map[proxy.ServicePortName][]*endpointInfo{},
expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{}, expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{},
expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{}, expectedNewlyActiveUDPServices: map[proxy.ServicePortName]bool{},
expectedLocalEndpoints: map[types.NamespacedName]int{}, expectedLocalEndpoints: map[types.NamespacedName]int{},
@ -3749,12 +3749,12 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "no change, named port, local", name: "no change, named port, local",
previousEndpoints: namedPortLocal, previousEndpoints: namedPortLocal,
currentEndpoints: namedPortLocal, currentEndpoints: namedPortLocal,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
@ -3769,7 +3769,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "no change, multiple subsets", name: "no change, multiple subsets",
previousEndpoints: multipleSubsets, previousEndpoints: multipleSubsets,
currentEndpoints: multipleSubsets, currentEndpoints: multipleSubsets,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -3777,7 +3777,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:12", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:12", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -3793,7 +3793,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "no change, multiple subsets, multiple ports, local", name: "no change, multiple subsets, multiple ports, local",
previousEndpoints: multipleSubsetsMultiplePortsLocal, previousEndpoints: multipleSubsetsMultiplePortsLocal,
currentEndpoints: multipleSubsetsMultiplePortsLocal, currentEndpoints: multipleSubsetsMultiplePortsLocal,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
@ -3804,7 +3804,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.3:13", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.3:13", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
@ -3825,7 +3825,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "no change, multiple endpoints, subsets, IPs, and ports", name: "no change, multiple endpoints, subsets, IPs, and ports",
previousEndpoints: multipleSubsetsIPsPorts, previousEndpoints: multipleSubsetsIPsPorts,
currentEndpoints: multipleSubsetsIPsPorts, currentEndpoints: multipleSubsetsIPsPorts,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
@ -3851,7 +3851,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.2.2.2:22", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.2.2.2:22", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
@ -3888,8 +3888,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "add an Endpoints", name: "add an Endpoints",
previousEndpoints: []*discovery.EndpointSlice{nil}, previousEndpoints: []*discovery.EndpointSlice{nil},
currentEndpoints: namedPortLocal, currentEndpoints: namedPortLocal,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{}, oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{},
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
@ -3906,12 +3906,12 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "remove an Endpoints", name: "remove an Endpoints",
previousEndpoints: namedPortLocal, previousEndpoints: namedPortLocal,
currentEndpoints: []*discovery.EndpointSlice{nil}, currentEndpoints: []*discovery.EndpointSlice{nil},
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{}, expectedResult: map[proxy.ServicePortName][]*endpointInfo{},
expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{ expectedDeletedUDPEndpoints: []proxy.ServiceEndpoint{{
Endpoint: "10.1.1.1:11", Endpoint: "10.1.1.1:11",
ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP), ServicePortName: makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP),
@ -3923,12 +3923,12 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "add an IP and port", name: "add an IP and port",
previousEndpoints: namedPort, previousEndpoints: namedPort,
currentEndpoints: namedPortsLocalNoLocal, currentEndpoints: namedPortsLocalNoLocal,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
@ -3950,7 +3950,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "remove an IP and port", name: "remove an IP and port",
previousEndpoints: namedPortsLocalNoLocal, previousEndpoints: namedPortsLocalNoLocal,
currentEndpoints: namedPort, currentEndpoints: namedPort,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:11", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
@ -3960,7 +3960,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:12", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:12", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -3982,12 +3982,12 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "add a subset", name: "add a subset",
previousEndpoints: []*discovery.EndpointSlice{namedPort[0], nil}, previousEndpoints: []*discovery.EndpointSlice{namedPort[0], nil},
currentEndpoints: multipleSubsetsWithLocal, currentEndpoints: multipleSubsetsWithLocal,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -4007,7 +4007,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "remove a subset", name: "remove a subset",
previousEndpoints: multipleSubsets, previousEndpoints: multipleSubsets,
currentEndpoints: []*discovery.EndpointSlice{namedPort[0], nil}, currentEndpoints: []*discovery.EndpointSlice{namedPort[0], nil},
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -4015,7 +4015,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:12", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.2:12", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -4031,12 +4031,12 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "rename a port", name: "rename a port",
previousEndpoints: namedPort, previousEndpoints: namedPort,
currentEndpoints: namedPortRenamed, currentEndpoints: namedPortRenamed,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11-2", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11-2", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -4054,12 +4054,12 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "renumber a port", name: "renumber a port",
previousEndpoints: namedPort, previousEndpoints: namedPort,
currentEndpoints: namedPortRenumbered, currentEndpoints: namedPortRenumbered,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:22", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:22", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -4075,7 +4075,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "complex add and remove", name: "complex add and remove",
previousEndpoints: complexBefore, previousEndpoints: complexBefore,
currentEndpoints: complexAfter, currentEndpoints: complexAfter,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },
@ -4094,7 +4094,7 @@ func TestUpdateEndpointsMap(t *testing.T) {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.4.4.6:45", IsLocal: true, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.4.4.6:45", IsLocal: true, Ready: true, Serving: true, Terminating: false}},
}, },
}, },
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.11:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.11:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
@ -4141,8 +4141,8 @@ func TestUpdateEndpointsMap(t *testing.T) {
name: "change from 0 endpoint address to 1 unnamed port", name: "change from 0 endpoint address to 1 unnamed port",
previousEndpoints: emptyEndpointSlices, previousEndpoints: emptyEndpointSlices,
currentEndpoints: namedPort, currentEndpoints: namedPort,
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{}, oldEndpoints: map[proxy.ServicePortName][]*endpointInfo{},
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{ expectedResult: map[proxy.ServicePortName][]*endpointInfo{
makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): { makeServicePortName("ns1", "ep1", "p11", v1.ProtocolUDP): {
{BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}}, {BaseEndpointInfo: &proxy.BaseEndpointInfo{Endpoint: "10.1.1.1:11", IsLocal: false, Ready: true, Serving: true, Terminating: false}},
}, },

4
pkg/proxy/ipvs/proxier.go
View File

@ -219,7 +219,7 @@ type Proxier struct {
// services that happened since last syncProxyRules call. For a single object, // services that happened since last syncProxyRules call. For a single object,
// changes are accumulated, i.e. previous is state from before all of them, // changes are accumulated, i.e. previous is state from before all of them,
// current is state after applying all of those. // current is state after applying all of those.
endpointsChanges *proxy.EndpointChangeTracker endpointsChanges *proxy.EndpointsChangeTracker
serviceChanges *proxy.ServiceChangeTracker serviceChanges *proxy.ServiceChangeTracker
mu sync.Mutex // protects the following fields mu sync.Mutex // protects the following fields
@ -413,7 +413,7 @@ func NewProxier(ipFamily v1.IPFamily,
svcPortMap: make(proxy.ServicePortMap), svcPortMap: make(proxy.ServicePortMap),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, nil, ipFamily, recorder, nil), endpointsChanges: proxy.NewEndpointsChangeTracker(hostname, nil, ipFamily, recorder, nil),
initialSync: true, initialSync: true,
syncPeriod: syncPeriod, syncPeriod: syncPeriod,
minSyncPeriod: minSyncPeriod, minSyncPeriod: minSyncPeriod,

4
pkg/proxy/ipvs/proxier_test.go
View File

@ -154,7 +154,7 @@ func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, ipset u
svcPortMap: make(proxy.ServicePortMap), svcPortMap: make(proxy.ServicePortMap),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, nil, nil), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, nil, nil),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, nil, ipFamily, nil, nil), endpointsChanges: proxy.NewEndpointsChangeTracker(testHostname, nil, ipFamily, nil, nil),
excludeCIDRs: excludeCIDRs, excludeCIDRs: excludeCIDRs,
iptables: ipt, iptables: ipt,
ipvs: ipvs, ipvs: ipvs,
@ -4358,7 +4358,7 @@ raid10 57344 0 - Live 0xffffffffc0597000`,
} }
// The majority of EndpointSlice specific tests are not ipvs specific and focus on // The majority of EndpointSlice specific tests are not ipvs specific and focus on
// the shared EndpointChangeTracker and EndpointSliceCache. This test ensures that the // the shared EndpointsChangeTracker and EndpointSliceCache. This test ensures that the
// ipvs proxier supports translating EndpointSlices to ipvs output. // ipvs proxier supports translating EndpointSlices to ipvs output.
func TestEndpointSliceE2E(t *testing.T) { func TestEndpointSliceE2E(t *testing.T) {
ipt := iptablestest.NewFake() ipt := iptablestest.NewFake()

4
pkg/proxy/service_test.go
View File

@ -595,7 +595,7 @@ func TestServiceToServiceMap(t *testing.T) {
} }
type FakeProxier struct { type FakeProxier struct {
endpointsChanges *EndpointChangeTracker endpointsChanges *EndpointsChangeTracker
serviceChanges *ServiceChangeTracker serviceChanges *ServiceChangeTracker
svcPortMap ServicePortMap svcPortMap ServicePortMap
endpointsMap EndpointsMap endpointsMap EndpointsMap
@ -607,7 +607,7 @@ func newFakeProxier(ipFamily v1.IPFamily, t time.Time) *FakeProxier {
svcPortMap: make(ServicePortMap), svcPortMap: make(ServicePortMap),
serviceChanges: NewServiceChangeTracker(nil, ipFamily, nil, nil), serviceChanges: NewServiceChangeTracker(nil, ipFamily, nil, nil),
endpointsMap: make(EndpointsMap), endpointsMap: make(EndpointsMap),
endpointsChanges: &EndpointChangeTracker{ endpointsChanges: &EndpointsChangeTracker{
lastChangeTriggerTimes: make(map[types.NamespacedName][]time.Time), lastChangeTriggerTimes: make(map[types.NamespacedName][]time.Time),
trackerStartTime: t, trackerStartTime: t,
processEndpointsMapChange: nil, processEndpointsMapChange: nil,

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

@ -32,13 +32,13 @@ import (
type HostNetworkService interface { type HostNetworkService interface {
getNetworkByName(name string) (*hnsNetworkInfo, error) getNetworkByName(name string) (*hnsNetworkInfo, error)
getAllEndpointsByNetwork(networkName string) (map[string]*endpointsInfo, error) getAllEndpointsByNetwork(networkName string) (map[string]*endpointInfo, error)
getEndpointByID(id string) (*endpointsInfo, error) getEndpointByID(id string) (*endpointInfo, error)
getEndpointByIpAddress(ip string, networkName string) (*endpointsInfo, error) getEndpointByIpAddress(ip string, networkName string) (*endpointInfo, error)
getEndpointByName(id string) (*endpointsInfo, error) getEndpointByName(id string) (*endpointInfo, error)
createEndpoint(ep *endpointsInfo, networkName string) (*endpointsInfo, error) createEndpoint(ep *endpointInfo, networkName string) (*endpointInfo, error)
deleteEndpoint(hnsID string) error deleteEndpoint(hnsID string) error
getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error) getLoadBalancer(endpoints []endpointInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error)
getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerInfo, error) getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerInfo, error)
deleteLoadBalancer(hnsID string) error deleteLoadBalancer(hnsID string) error
} }
@ -87,7 +87,7 @@ func (hns hns) getNetworkByName(name string) (*hnsNetworkInfo, error) {
}, nil }, nil
} }
func (hns hns) getAllEndpointsByNetwork(networkName string) (map[string]*(endpointsInfo), error) { func (hns hns) getAllEndpointsByNetwork(networkName string) (map[string]*(endpointInfo), error) {
hcnnetwork, err := hns.hcn.GetNetworkByName(networkName) hcnnetwork, err := hns.hcn.GetNetworkByName(networkName)
if err != nil { if err != nil {
klog.ErrorS(err, "failed to get HNS network by name", "name", networkName) klog.ErrorS(err, "failed to get HNS network by name", "name", networkName)
@ -97,7 +97,7 @@ func (hns hns) getAllEndpointsByNetwork(networkName string) (map[string]*(endpoi
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to list endpoints: %w", err) return nil, fmt.Errorf("failed to list endpoints: %w", err)
} }
endpointInfos := make(map[string]*(endpointsInfo)) endpointInfos := make(map[string]*(endpointInfo))
for _, ep := range endpoints { for _, ep := range endpoints {
if len(ep.IpConfigurations) == 0 { if len(ep.IpConfigurations) == 0 {
@ -108,7 +108,7 @@ func (hns hns) getAllEndpointsByNetwork(networkName string) (map[string]*(endpoi
// Add to map with key endpoint ID or IP address // Add to map with key endpoint ID or IP address
// Storing this is expensive in terms of memory, however there is a bug in Windows Server 2019 that can cause two endpoints to be created with the same IP address. // Storing this is expensive in terms of memory, however there is a bug in Windows Server 2019 that can cause two endpoints to be created with the same IP address.
// TODO: Store by IP only and remove any lookups by endpoint ID. // TODO: Store by IP only and remove any lookups by endpoint ID.
endpointInfos[ep.Id] = &endpointsInfo{ endpointInfos[ep.Id] = &endpointInfo{
ip: ep.IpConfigurations[0].IpAddress, ip: ep.IpConfigurations[0].IpAddress,
isLocal: uint32(ep.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, isLocal: uint32(ep.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0,
macAddress: ep.MacAddress, macAddress: ep.MacAddress,
@ -127,7 +127,7 @@ func (hns hns) getAllEndpointsByNetwork(networkName string) (map[string]*(endpoi
// If ipFamilyPolicy is RequireDualStack or PreferDualStack, then there will be 2 IPS (iPV4 and IPV6) // If ipFamilyPolicy is RequireDualStack or PreferDualStack, then there will be 2 IPS (iPV4 and IPV6)
// in the endpoint list // in the endpoint list
endpointDualstack := &endpointsInfo{ endpointDualstack := &endpointInfo{
ip: ep.IpConfigurations[1].IpAddress, ip: ep.IpConfigurations[1].IpAddress,
isLocal: uint32(ep.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, isLocal: uint32(ep.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0,
macAddress: ep.MacAddress, macAddress: ep.MacAddress,
@ -145,12 +145,12 @@ func (hns hns) getAllEndpointsByNetwork(networkName string) (map[string]*(endpoi
return endpointInfos, nil return endpointInfos, nil
} }
func (hns hns) getEndpointByID(id string) (*endpointsInfo, error) { func (hns hns) getEndpointByID(id string) (*endpointInfo, error) {
hnsendpoint, err := hns.hcn.GetEndpointByID(id) hnsendpoint, err := hns.hcn.GetEndpointByID(id)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return &endpointsInfo{ //TODO: fill out PA return &endpointInfo{ //TODO: fill out PA
ip: hnsendpoint.IpConfigurations[0].IpAddress, ip: hnsendpoint.IpConfigurations[0].IpAddress,
isLocal: uint32(hnsendpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote isLocal: uint32(hnsendpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote
macAddress: hnsendpoint.MacAddress, macAddress: hnsendpoint.MacAddress,
@ -158,7 +158,7 @@ func (hns hns) getEndpointByID(id string) (*endpointsInfo, error) {
hns: hns, hns: hns,
}, nil }, nil
} }
func (hns hns) getEndpointByIpAddress(ip string, networkName string) (*endpointsInfo, error) { func (hns hns) getEndpointByIpAddress(ip string, networkName string) (*endpointInfo, error) {
hnsnetwork, err := hns.hcn.GetNetworkByName(networkName) hnsnetwork, err := hns.hcn.GetNetworkByName(networkName)
if err != nil { if err != nil {
klog.ErrorS(err, "Error getting network by name") klog.ErrorS(err, "Error getting network by name")
@ -179,7 +179,7 @@ func (hns hns) getEndpointByIpAddress(ip string, networkName string) (*endpoints
} }
} }
if equal && strings.EqualFold(endpoint.HostComputeNetwork, hnsnetwork.Id) { if equal && strings.EqualFold(endpoint.HostComputeNetwork, hnsnetwork.Id) {
return &endpointsInfo{ return &endpointInfo{
ip: ip, ip: ip,
isLocal: uint32(endpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote isLocal: uint32(endpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote
macAddress: endpoint.MacAddress, macAddress: endpoint.MacAddress,
@ -190,12 +190,12 @@ func (hns hns) getEndpointByIpAddress(ip string, networkName string) (*endpoints
} }
return nil, fmt.Errorf("Endpoint %v not found on network %s", ip, networkName) return nil, fmt.Errorf("Endpoint %v not found on network %s", ip, networkName)
} }
func (hns hns) getEndpointByName(name string) (*endpointsInfo, error) { func (hns hns) getEndpointByName(name string) (*endpointInfo, error) {
hnsendpoint, err := hns.hcn.GetEndpointByName(name) hnsendpoint, err := hns.hcn.GetEndpointByName(name)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return &endpointsInfo{ //TODO: fill out PA return &endpointInfo{ //TODO: fill out PA
ip: hnsendpoint.IpConfigurations[0].IpAddress, ip: hnsendpoint.IpConfigurations[0].IpAddress,
isLocal: uint32(hnsendpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote isLocal: uint32(hnsendpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, //TODO: Change isLocal to isRemote
macAddress: hnsendpoint.MacAddress, macAddress: hnsendpoint.MacAddress,
@ -203,7 +203,7 @@ func (hns hns) getEndpointByName(name string) (*endpointsInfo, error) {
hns: hns, hns: hns,
}, nil }, nil
} }
func (hns hns) createEndpoint(ep *endpointsInfo, networkName string) (*endpointsInfo, error) { func (hns hns) createEndpoint(ep *endpointInfo, networkName string) (*endpointInfo, error) {
hnsNetwork, err := hns.hcn.GetNetworkByName(networkName) hnsNetwork, err := hns.hcn.GetNetworkByName(networkName)
if err != nil { if err != nil {
return nil, err return nil, err
@ -251,7 +251,7 @@ func (hns hns) createEndpoint(ep *endpointsInfo, networkName string) (*endpoints
return nil, err return nil, err
} }
} }
return &endpointsInfo{ return &endpointInfo{
ip: createdEndpoint.IpConfigurations[0].IpAddress, ip: createdEndpoint.IpConfigurations[0].IpAddress,
isLocal: uint32(createdEndpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0, isLocal: uint32(createdEndpoint.Flags&hcn.EndpointFlagsRemoteEndpoint) == 0,
macAddress: createdEndpoint.MacAddress, macAddress: createdEndpoint.MacAddress,
@ -273,7 +273,7 @@ func (hns hns) deleteEndpoint(hnsID string) error {
} }
// findLoadBalancerID will construct a id from the provided loadbalancer fields // findLoadBalancerID will construct a id from the provided loadbalancer fields
func findLoadBalancerID(endpoints []endpointsInfo, vip string, protocol, internalPort, externalPort uint16) (loadBalancerIdentifier, error) { func findLoadBalancerID(endpoints []endpointInfo, vip string, protocol, internalPort, externalPort uint16) (loadBalancerIdentifier, error) {
// Compute hash from backends (endpoint IDs) // Compute hash from backends (endpoint IDs)
hash, err := hashEndpoints(endpoints) hash, err := hashEndpoints(endpoints)
if err != nil { if err != nil {
@ -315,7 +315,7 @@ func (hns hns) getAllLoadBalancers() (map[loadBalancerIdentifier]*loadBalancerIn
return loadBalancers, nil return loadBalancers, nil
} }
func (hns hns) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancerFlags, sourceVip string, vip string, protocol uint16, internalPort uint16, externalPort uint16, previousLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) (*loadBalancerInfo, error) { 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 var id loadBalancerIdentifier
vips := []string{} vips := []string{}
// Compute hash from backends (endpoint IDs) // Compute hash from backends (endpoint IDs)
@ -424,7 +424,7 @@ func (hns hns) deleteLoadBalancer(hnsID string) error {
} }
// Calculates a hash from the given endpoint IDs. // Calculates a hash from the given endpoint IDs.
func hashEndpoints[T string | endpointsInfo](endpoints []T) (hash [20]byte, err error) { func hashEndpoints[T string | endpointInfo](endpoints []T) (hash [20]byte, err error) {
var id string var id string
// Recover in case something goes wrong. Return error and null byte array. // Recover in case something goes wrong. Return error and null byte array.
defer func() { defer func() {
@ -437,7 +437,7 @@ func hashEndpoints[T string | endpointsInfo](endpoints []T) (hash [20]byte, err
// Iterate over endpoints, compute hash // Iterate over endpoints, compute hash
for _, ep := range endpoints { for _, ep := range endpoints {
switch x := any(ep).(type) { switch x := any(ep).(type) {
case endpointsInfo: case endpointInfo:
id = strings.ToUpper(x.hnsID) id = strings.ToUpper(x.hnsID)
case string: case string:
id = x id = x

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

@ -203,7 +203,7 @@ func TestCreateEndpointLocal(t *testing.T) {
hns := hns{hcn: newHcnImpl()} hns := hns{hcn: newHcnImpl()}
Network := mustTestNetwork(t) Network := mustTestNetwork(t)
endpoint := &endpointsInfo{ endpoint := &endpointInfo{
ip: epIpAddress, ip: epIpAddress,
macAddress: epMacAddress, macAddress: epMacAddress,
isLocal: true, isLocal: true,
@ -242,7 +242,7 @@ func TestCreateEndpointRemote(t *testing.T) {
Network := mustTestNetwork(t) Network := mustTestNetwork(t)
providerAddress := epPaAddress providerAddress := epPaAddress
endpoint := &endpointsInfo{ endpoint := &endpointInfo{
ip: epIpAddressRemote, ip: epIpAddressRemote,
macAddress: epMacAddress, macAddress: epMacAddress,
isLocal: false, isLocal: false,
@ -350,11 +350,11 @@ func TestGetLoadBalancerExisting(t *testing.T) {
if err != nil { if err != nil {
t.Error(err) t.Error(err)
} }
endpoint := &endpointsInfo{ endpoint := &endpointInfo{
ip: Endpoint.IpConfigurations[0].IpAddress, ip: Endpoint.IpConfigurations[0].IpAddress,
hnsID: Endpoint.Id, hnsID: Endpoint.Id,
} }
endpoints := []endpointsInfo{*endpoint} endpoints := []endpointInfo{*endpoint}
hash, err := hashEndpoints(endpoints) hash, err := hashEndpoints(endpoints)
if err != nil { if err != nil {
t.Error(err) t.Error(err)
@ -409,11 +409,11 @@ func TestGetLoadBalancerNew(t *testing.T) {
if err != nil { if err != nil {
t.Error(err) t.Error(err)
} }
endpoint := &endpointsInfo{ endpoint := &endpointInfo{
ip: Endpoint.IpConfigurations[0].IpAddress, ip: Endpoint.IpConfigurations[0].IpAddress,
hnsID: Endpoint.Id, hnsID: Endpoint.Id,
} }
endpoints := []endpointsInfo{*endpoint} endpoints := []endpointInfo{*endpoint}
lb, err := hns.getLoadBalancer(endpoints, loadBalancerFlags{}, sourceVip, serviceVip, protocol, internalPort, externalPort, lbs) lb, err := hns.getLoadBalancer(endpoints, loadBalancerFlags{}, sourceVip, serviceVip, protocol, internalPort, externalPort, lbs)
if err != nil { if err != nil {
t.Error(err) t.Error(err)
@ -523,7 +523,7 @@ func TestHashEndpoints(t *testing.T) {
if err != nil { if err != nil {
t.Error(err) t.Error(err)
} }
endpointInfoA := &endpointsInfo{ endpointInfoA := &endpointInfo{
ip: endpointA.IpConfigurations[0].IpAddress, ip: endpointA.IpConfigurations[0].IpAddress,
hnsID: endpointA.Id, hnsID: endpointA.Id,
} }
@ -543,12 +543,12 @@ func TestHashEndpoints(t *testing.T) {
if err != nil { if err != nil {
t.Error(err) t.Error(err)
} }
endpointInfoB := &endpointsInfo{ endpointInfoB := &endpointInfo{
ip: endpointB.IpConfigurations[0].IpAddress, ip: endpointB.IpConfigurations[0].IpAddress,
hnsID: endpointB.Id, hnsID: endpointB.Id,
} }
endpoints := []endpointsInfo{*endpointInfoA, *endpointInfoB} endpoints := []endpointInfo{*endpointInfoA, *endpointInfoB}
endpointsReverse := []endpointsInfo{*endpointInfoB, *endpointInfoA} endpointsReverse := []endpointInfo{*endpointInfoB, *endpointInfoA}
h1, err := hashEndpoints(endpoints) h1, err := hashEndpoints(endpoints)
if err != nil { if err != nil {
t.Error(err) t.Error(err)

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

@ -123,7 +123,7 @@ type serviceInfo struct {
hnsID string hnsID string
nodePorthnsID string nodePorthnsID string
policyApplied bool policyApplied bool
remoteEndpoint *endpointsInfo remoteEndpoint *endpointInfo
hns HostNetworkService hns HostNetworkService
preserveDIP bool preserveDIP bool
localTrafficDSR bool localTrafficDSR bool
@ -273,7 +273,7 @@ func (t DualStackCompatTester) DualStackCompatible(networkName string) bool {
} }
// internal struct for endpoints information // internal struct for endpoints information
type endpointsInfo struct { type endpointInfo struct {
ip string ip string
port uint16 port uint16
isLocal bool isLocal bool
@ -290,57 +290,57 @@ type endpointsInfo struct {
} }
// String is part of proxy.Endpoint interface. // String is part of proxy.Endpoint interface.
func (info *endpointsInfo) String() string { func (info *endpointInfo) String() string {
return net.JoinHostPort(info.ip, strconv.Itoa(int(info.port))) return net.JoinHostPort(info.ip, strconv.Itoa(int(info.port)))
} }
// GetIsLocal is part of proxy.Endpoint interface. // GetIsLocal is part of proxy.Endpoint interface.
func (info *endpointsInfo) GetIsLocal() bool { func (info *endpointInfo) GetIsLocal() bool {
return info.isLocal return info.isLocal
} }
// IsReady returns true if an endpoint is ready and not terminating. // IsReady returns true if an endpoint is ready and not terminating.
func (info *endpointsInfo) IsReady() bool { func (info *endpointInfo) IsReady() bool {
return info.ready return info.ready
} }
// IsServing returns true if an endpoint is ready, regardless of it's terminating state. // IsServing returns true if an endpoint is ready, regardless of it's terminating state.
func (info *endpointsInfo) IsServing() bool { func (info *endpointInfo) IsServing() bool {
return info.serving return info.serving
} }
// IsTerminating returns true if an endpoint is terminating. // IsTerminating returns true if an endpoint is terminating.
func (info *endpointsInfo) IsTerminating() bool { func (info *endpointInfo) IsTerminating() bool {
return info.terminating return info.terminating
} }
// GetZoneHint returns the zone hint for the endpoint. // GetZoneHint returns the zone hint for the endpoint.
func (info *endpointsInfo) GetZoneHints() sets.Set[string] { func (info *endpointInfo) GetZoneHints() sets.Set[string] {
return sets.Set[string]{} return sets.Set[string]{}
} }
// IP returns just the IP part of the endpoint, it's a part of proxy.Endpoint interface. // IP returns just the IP part of the endpoint, it's a part of proxy.Endpoint interface.
func (info *endpointsInfo) IP() string { func (info *endpointInfo) IP() string {
return info.ip return info.ip
} }
// Port returns just the Port part of the endpoint. // Port returns just the Port part of the endpoint.
func (info *endpointsInfo) Port() (int, error) { func (info *endpointInfo) Port() (int, error) {
return int(info.port), nil return int(info.port), nil
} }
// Equal is part of proxy.Endpoint interface. // Equal is part of proxy.Endpoint interface.
func (info *endpointsInfo) Equal(other proxy.Endpoint) bool { func (info *endpointInfo) Equal(other proxy.Endpoint) bool {
return info.String() == other.String() && info.GetIsLocal() == other.GetIsLocal() return info.String() == other.String() && info.GetIsLocal() == other.GetIsLocal()
} }
// GetNodeName returns the NodeName for this endpoint. // GetNodeName returns the NodeName for this endpoint.
func (info *endpointsInfo) GetNodeName() string { func (info *endpointInfo) GetNodeName() string {
return "" return ""
} }
// GetZone returns the Zone for this endpoint. // GetZone returns the Zone for this endpoint.
func (info *endpointsInfo) GetZone() string { func (info *endpointInfo) GetZone() string {
return "" return ""
} }
@ -407,7 +407,7 @@ func (proxier *Proxier) onEndpointsMapChange(svcPortName *proxy.ServicePortName,
if exists { if exists {
// Cleanup Endpoints references // Cleanup Endpoints references
for _, ep := range epInfos { for _, ep := range epInfos {
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if ok { if ok {
epInfo.Cleanup() epInfo.Cleanup()
@ -448,7 +448,7 @@ func (proxier *Proxier) onServiceMapChange(svcPortName *proxy.ServicePortName) {
} }
} }
// returns a new proxy.Endpoint which abstracts a endpointsInfo // returns a new proxy.Endpoint which abstracts a endpointInfo
func (proxier *Proxier) newEndpointInfo(baseInfo *proxy.BaseEndpointInfo, _ *proxy.ServicePortName) proxy.Endpoint { func (proxier *Proxier) newEndpointInfo(baseInfo *proxy.BaseEndpointInfo, _ *proxy.ServicePortName) proxy.Endpoint {
portNumber, err := baseInfo.Port() portNumber, err := baseInfo.Port()
@ -457,7 +457,7 @@ func (proxier *Proxier) newEndpointInfo(baseInfo *proxy.BaseEndpointInfo, _ *pro
portNumber = 0 portNumber = 0
} }
info := &endpointsInfo{ info := &endpointInfo{
ip: baseInfo.IP(), ip: baseInfo.IP(),
port: uint16(portNumber), port: uint16(portNumber),
isLocal: baseInfo.GetIsLocal(), isLocal: baseInfo.GetIsLocal(),
@ -474,8 +474,8 @@ func (proxier *Proxier) newEndpointInfo(baseInfo *proxy.BaseEndpointInfo, _ *pro
return info return info
} }
func newSourceVIP(hns HostNetworkService, network string, ip string, mac string, providerAddress string) (*endpointsInfo, error) { func newSourceVIP(hns HostNetworkService, network string, ip string, mac string, providerAddress string) (*endpointInfo, error) {
hnsEndpoint := &endpointsInfo{ hnsEndpoint := &endpointInfo{
ip: ip, ip: ip,
isLocal: true, isLocal: true,
macAddress: mac, macAddress: mac,
@ -489,15 +489,15 @@ func newSourceVIP(hns HostNetworkService, network string, ip string, mac string,
return ep, err return ep, err
} }
func (ep *endpointsInfo) DecrementRefCount() { func (ep *endpointInfo) DecrementRefCount() {
klog.V(3).InfoS("Decrementing Endpoint RefCount", "endpointsInfo", ep) klog.V(3).InfoS("Decrementing Endpoint RefCount", "endpointInfo", ep)
if !ep.GetIsLocal() && ep.refCount != nil && *ep.refCount > 0 { if !ep.GetIsLocal() && ep.refCount != nil && *ep.refCount > 0 {
*ep.refCount-- *ep.refCount--
} }
} }
func (ep *endpointsInfo) Cleanup() { func (ep *endpointInfo) Cleanup() {
klog.V(3).InfoS("Endpoint cleanup", "endpointsInfo", ep) klog.V(3).InfoS("Endpoint cleanup", "endpointInfo", ep)
if !ep.GetIsLocal() && ep.refCount != nil { if !ep.GetIsLocal() && ep.refCount != nil {
*ep.refCount-- *ep.refCount--
@ -601,7 +601,7 @@ type Proxier struct {
// services that happened since policies were synced. For a single object, // services that happened since policies were synced. For a single object,
// changes are accumulated, i.e. previous is state from before all of them, // changes are accumulated, i.e. previous is state from before all of them,
// current is state after applying all of those. // current is state after applying all of those.
endpointsChanges *proxy.EndpointChangeTracker endpointsChanges *proxy.EndpointsChangeTracker
serviceChanges *proxy.ServiceChangeTracker serviceChanges *proxy.ServiceChangeTracker
endPointsRefCount endPointsReferenceCountMap endPointsRefCount endPointsReferenceCountMap
mu sync.Mutex // protects the following fields mu sync.Mutex // protects the following fields
@ -802,7 +802,7 @@ func NewProxier(
} }
serviceChanges := proxy.NewServiceChangeTracker(proxier.newServiceInfo, ipFamily, recorder, proxier.serviceMapChange) serviceChanges := proxy.NewServiceChangeTracker(proxier.newServiceInfo, ipFamily, recorder, proxier.serviceMapChange)
endPointChangeTracker := proxy.NewEndpointChangeTracker(hostname, proxier.newEndpointInfo, ipFamily, recorder, proxier.endpointsMapChange) endPointChangeTracker := proxy.NewEndpointsChangeTracker(hostname, proxier.newEndpointInfo, ipFamily, recorder, proxier.endpointsMapChange)
proxier.endpointsChanges = endPointChangeTracker proxier.endpointsChanges = endPointChangeTracker
proxier.serviceChanges = serviceChanges proxier.serviceChanges = serviceChanges
@ -868,7 +868,7 @@ func (svcInfo *serviceInfo) cleanupAllPolicies(endpoints []proxy.Endpoint, mapSt
} }
// Cleanup Endpoints references // Cleanup Endpoints references
for _, ep := range endpoints { for _, ep := range endpoints {
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if ok { if ok {
if winProxyOptimization { if winProxyOptimization {
epInfo.DecrementRefCount() epInfo.DecrementRefCount()
@ -1062,7 +1062,7 @@ func isNetworkNotFoundError(err error) bool {
// If atleast one is not terminating, then return false // If atleast one is not terminating, then return false
func (proxier *Proxier) isAllEndpointsTerminating(svcName proxy.ServicePortName, isLocalTrafficDSR bool) bool { func (proxier *Proxier) isAllEndpointsTerminating(svcName proxy.ServicePortName, isLocalTrafficDSR bool) bool {
for _, epInfo := range proxier.endpointsMap[svcName] { for _, epInfo := range proxier.endpointsMap[svcName] {
ep, ok := epInfo.(*endpointsInfo) ep, ok := epInfo.(*endpointInfo)
if !ok { if !ok {
continue continue
} }
@ -1087,7 +1087,7 @@ func (proxier *Proxier) isAllEndpointsTerminating(svcName proxy.ServicePortName,
// If atleast one is serving, then return false // If atleast one is serving, then return false
func (proxier *Proxier) isAllEndpointsNonServing(svcName proxy.ServicePortName, isLocalTrafficDSR bool) bool { func (proxier *Proxier) isAllEndpointsNonServing(svcName proxy.ServicePortName, isLocalTrafficDSR bool) bool {
for _, epInfo := range proxier.endpointsMap[svcName] { for _, epInfo := range proxier.endpointsMap[svcName] {
ep, ok := epInfo.(*endpointsInfo) ep, ok := epInfo.(*endpointInfo)
if !ok { if !ok {
continue continue
} }
@ -1102,7 +1102,7 @@ func (proxier *Proxier) isAllEndpointsNonServing(svcName proxy.ServicePortName,
} }
// updateQueriedEndpoints updates the queriedEndpoints map with newly created endpoint details // updateQueriedEndpoints updates the queriedEndpoints map with newly created endpoint details
func updateQueriedEndpoints(newHnsEndpoint *endpointsInfo, queriedEndpoints map[string]*endpointsInfo) { func updateQueriedEndpoints(newHnsEndpoint *endpointInfo, queriedEndpoints map[string]*endpointInfo) {
// store newly created endpoints in queriedEndpoints // store newly created endpoints in queriedEndpoints
queriedEndpoints[newHnsEndpoint.hnsID] = newHnsEndpoint queriedEndpoints[newHnsEndpoint.hnsID] = newHnsEndpoint
queriedEndpoints[newHnsEndpoint.ip] = newHnsEndpoint queriedEndpoints[newHnsEndpoint.ip] = newHnsEndpoint
@ -1130,7 +1130,7 @@ func (proxier *Proxier) syncProxyRules() {
hnsNetworkName := proxier.network.name hnsNetworkName := proxier.network.name
hns := proxier.hns hns := proxier.hns
var gatewayHnsendpoint *endpointsInfo var gatewayHnsendpoint *endpointInfo
if proxier.forwardHealthCheckVip { if proxier.forwardHealthCheckVip {
gatewayHnsendpoint, _ = hns.getEndpointByName(proxier.rootHnsEndpointName) gatewayHnsendpoint, _ = hns.getEndpointByName(proxier.rootHnsEndpointName)
} }
@ -1153,7 +1153,7 @@ func (proxier *Proxier) syncProxyRules() {
endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges) endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)
deletedUDPClusterIPs := serviceUpdateResult.DeletedUDPClusterIPs deletedUDPClusterIPs := serviceUpdateResult.DeletedUDPClusterIPs
// merge stale services gathered from updateEndpointsMap // merge stale services gathered from EndpointsMap.Update
for _, svcPortName := range endpointUpdateResult.NewlyActiveUDPServices { for _, svcPortName := range endpointUpdateResult.NewlyActiveUDPServices {
if svcInfo, ok := proxier.svcPortMap[svcPortName]; ok && svcInfo != nil && svcInfo.Protocol() == v1.ProtocolUDP { if svcInfo, ok := proxier.svcPortMap[svcPortName]; ok && svcInfo != nil && svcInfo.Protocol() == v1.ProtocolUDP {
klog.V(2).InfoS("Newly-active UDP service may have stale conntrack entries", "servicePortName", svcPortName) klog.V(2).InfoS("Newly-active UDP service may have stale conntrack entries", "servicePortName", svcPortName)
@ -1168,7 +1168,7 @@ func (proxier *Proxier) syncProxyRules() {
} }
if queriedEndpoints == nil { if queriedEndpoints == nil {
klog.V(4).InfoS("No existing endpoints found in HNS") klog.V(4).InfoS("No existing endpoints found in HNS")
queriedEndpoints = make(map[string]*(endpointsInfo)) queriedEndpoints = make(map[string]*(endpointInfo))
} }
queriedLoadBalancers, err := hns.getAllLoadBalancers() queriedLoadBalancers, err := hns.getAllLoadBalancers()
if queriedLoadBalancers == nil { if queriedLoadBalancers == nil {
@ -1208,7 +1208,7 @@ func (proxier *Proxier) syncProxyRules() {
serviceVipEndpoint := queriedEndpoints[svcInfo.ClusterIP().String()] serviceVipEndpoint := queriedEndpoints[svcInfo.ClusterIP().String()]
if serviceVipEndpoint == nil { if serviceVipEndpoint == nil {
klog.V(4).InfoS("No existing remote endpoint", "IP", svcInfo.ClusterIP()) klog.V(4).InfoS("No existing remote endpoint", "IP", svcInfo.ClusterIP())
hnsEndpoint := &endpointsInfo{ hnsEndpoint := &endpointInfo{
ip: svcInfo.ClusterIP().String(), ip: svcInfo.ClusterIP().String(),
isLocal: false, isLocal: false,
macAddress: proxier.hostMac, macAddress: proxier.hostMac,
@ -1228,8 +1228,8 @@ func (proxier *Proxier) syncProxyRules() {
} }
} }
var hnsEndpoints []endpointsInfo var hnsEndpoints []endpointInfo
var hnsLocalEndpoints []endpointsInfo var hnsLocalEndpoints []endpointInfo
klog.V(4).InfoS("Applying Policy", "serviceInfo", svcName) klog.V(4).InfoS("Applying Policy", "serviceInfo", svcName)
// Create Remote endpoints for every endpoint, corresponding to the service // Create Remote endpoints for every endpoint, corresponding to the service
containsPublicIP := false containsPublicIP := false
@ -1249,9 +1249,9 @@ func (proxier *Proxier) syncProxyRules() {
} }
for _, epInfo := range proxier.endpointsMap[svcName] { for _, epInfo := range proxier.endpointsMap[svcName] {
ep, ok := epInfo.(*endpointsInfo) ep, ok := epInfo.(*endpointInfo)
if !ok { if !ok {
klog.ErrorS(nil, "Failed to cast endpointsInfo", "serviceName", svcName) klog.ErrorS(nil, "Failed to cast endpointInfo", "serviceName", svcName)
continue continue
} }
@ -1274,7 +1274,7 @@ func (proxier *Proxier) syncProxyRules() {
} }
var newHnsEndpoint *endpointsInfo var newHnsEndpoint *endpointInfo
hnsNetworkName := proxier.network.name hnsNetworkName := proxier.network.name
var err error var err error
@ -1319,7 +1319,7 @@ func (proxier *Proxier) syncProxyRules() {
providerAddress = proxier.nodeIP.String() providerAddress = proxier.nodeIP.String()
} }
hnsEndpoint := &endpointsInfo{ hnsEndpoint := &endpointInfo{
ip: ep.ip, ip: ep.ip,
isLocal: false, isLocal: false,
macAddress: conjureMac("02-11", netutils.ParseIPSloppy(ep.ip)), macAddress: conjureMac("02-11", netutils.ParseIPSloppy(ep.ip)),
@ -1328,13 +1328,13 @@ func (proxier *Proxier) syncProxyRules() {
newHnsEndpoint, err = hns.createEndpoint(hnsEndpoint, hnsNetworkName) newHnsEndpoint, err = hns.createEndpoint(hnsEndpoint, hnsNetworkName)
if err != nil { if err != nil {
klog.ErrorS(err, "Remote endpoint creation failed", "endpointsInfo", hnsEndpoint) klog.ErrorS(err, "Remote endpoint creation failed", "endpointInfo", hnsEndpoint)
continue continue
} }
updateQueriedEndpoints(newHnsEndpoint, queriedEndpoints) updateQueriedEndpoints(newHnsEndpoint, queriedEndpoints)
} else { } else {
hnsEndpoint := &endpointsInfo{ hnsEndpoint := &endpointInfo{
ip: ep.ip, ip: ep.ip,
isLocal: false, isLocal: false,
macAddress: ep.macAddress, macAddress: ep.macAddress,
@ -1371,7 +1371,7 @@ func (proxier *Proxier) syncProxyRules() {
} }
// Save the hnsId for reference // Save the hnsId for reference
klog.V(1).InfoS("Hns endpoint resource", "endpointsInfo", newHnsEndpoint) klog.V(1).InfoS("Hns endpoint resource", "endpointInfo", newHnsEndpoint)
hnsEndpoints = append(hnsEndpoints, *newHnsEndpoint) hnsEndpoints = append(hnsEndpoints, *newHnsEndpoint)
if newHnsEndpoint.GetIsLocal() { if newHnsEndpoint.GetIsLocal() {
@ -1384,10 +1384,10 @@ func (proxier *Proxier) syncProxyRules() {
ep.hnsID = newHnsEndpoint.hnsID ep.hnsID = newHnsEndpoint.hnsID
klog.V(3).InfoS("Endpoint resource found", "endpointsInfo", ep) klog.V(3).InfoS("Endpoint resource found", "endpointInfo", ep)
} }
klog.V(3).InfoS("Associated endpoints for service", "endpointsInfo", hnsEndpoints, "serviceName", svcName) klog.V(3).InfoS("Associated endpoints for service", "endpointInfo", hnsEndpoints, "serviceName", svcName)
if len(svcInfo.hnsID) > 0 { if len(svcInfo.hnsID) > 0 {
// This should not happen // This should not happen
@ -1399,7 +1399,7 @@ func (proxier *Proxier) syncProxyRules() {
if len(hnsEndpoints) == 0 { if len(hnsEndpoints) == 0 {
if svcInfo.winProxyOptimization { if svcInfo.winProxyOptimization {
// Deleting loadbalancers when there are no endpoints to serve. // Deleting loadbalancers when there are no endpoints to serve.
klog.V(3).InfoS("Cleanup existing ", "endpointsInfo", hnsEndpoints, "serviceName", svcName) klog.V(3).InfoS("Cleanup existing ", "endpointInfo", hnsEndpoints, "serviceName", svcName)
svcInfo.deleteLoadBalancerPolicy(proxier.mapStaleLoadbalancers) svcInfo.deleteLoadBalancerPolicy(proxier.mapStaleLoadbalancers)
} }
klog.ErrorS(nil, "Endpoint information not available for service, not applying any policy", "serviceName", svcName) klog.ErrorS(nil, "Endpoint information not available for service, not applying any policy", "serviceName", svcName)
@ -1560,10 +1560,10 @@ func (proxier *Proxier) syncProxyRules() {
nodeport = svcInfo.HealthCheckNodePort() nodeport = svcInfo.HealthCheckNodePort()
} }
proxier.deleteExistingLoadBalancer(hns, svcInfo.winProxyOptimization, &lbIngressIP.healthCheckHnsID, sourceVip, Enum(svcInfo.Protocol()), uint16(svcInfo.targetPort), uint16(svcInfo.Port()), []endpointsInfo{*gatewayHnsendpoint}, queriedLoadBalancers) proxier.deleteExistingLoadBalancer(hns, svcInfo.winProxyOptimization, &lbIngressIP.healthCheckHnsID, sourceVip, Enum(svcInfo.Protocol()), uint16(svcInfo.targetPort), uint16(svcInfo.Port()), []endpointInfo{*gatewayHnsendpoint}, queriedLoadBalancers)
hnsHealthCheckLoadBalancer, err := hns.getLoadBalancer( hnsHealthCheckLoadBalancer, err := hns.getLoadBalancer(
[]endpointsInfo{*gatewayHnsendpoint}, []endpointInfo{*gatewayHnsendpoint},
loadBalancerFlags{isDSR: false, useMUX: svcInfo.preserveDIP, preserveDIP: svcInfo.preserveDIP}, loadBalancerFlags{isDSR: false, useMUX: svcInfo.preserveDIP, preserveDIP: svcInfo.preserveDIP},
sourceVip, sourceVip,
lbIngressIP.ip, lbIngressIP.ip,
@ -1623,7 +1623,7 @@ func (proxier *Proxier) syncProxyRules() {
// deleteExistingLoadBalancer checks whether loadbalancer delete is needed or not. // deleteExistingLoadBalancer checks whether loadbalancer delete is needed or not.
// If it is needed, the function will delete the existing loadbalancer and return true, else false. // If it is needed, the function will delete the existing loadbalancer and return true, else false.
func (proxier *Proxier) deleteExistingLoadBalancer(hns HostNetworkService, winProxyOptimization bool, lbHnsID *string, sourceVip string, protocol, intPort, extPort uint16, endpoints []endpointsInfo, queriedLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) bool { func (proxier *Proxier) deleteExistingLoadBalancer(hns HostNetworkService, winProxyOptimization bool, lbHnsID *string, sourceVip string, protocol, intPort, extPort uint16, endpoints []endpointInfo, queriedLoadBalancers map[loadBalancerIdentifier]*loadBalancerInfo) bool {
if !winProxyOptimization || *lbHnsID == "" { if !winProxyOptimization || *lbHnsID == "" {
// Loadbalancer delete not needed // Loadbalancer delete not needed

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

@ -123,8 +123,8 @@ func NewFakeProxier(syncPeriod time.Duration, minSyncPeriod time.Duration, clust
} }
serviceChanges := proxy.NewServiceChangeTracker(proxier.newServiceInfo, v1.IPv4Protocol, nil, proxier.serviceMapChange) serviceChanges := proxy.NewServiceChangeTracker(proxier.newServiceInfo, v1.IPv4Protocol, nil, proxier.serviceMapChange)
endpointChangeTracker := proxy.NewEndpointChangeTracker(hostname, proxier.newEndpointInfo, v1.IPv4Protocol, nil, proxier.endpointsMapChange) endpointsChangeTracker := proxy.NewEndpointsChangeTracker(hostname, proxier.newEndpointInfo, v1.IPv4Protocol, nil, proxier.endpointsMapChange)
proxier.endpointsChanges = endpointChangeTracker proxier.endpointsChanges = endpointsChangeTracker
proxier.serviceChanges = serviceChanges proxier.serviceChanges = serviceChanges
return proxier return proxier
@ -232,9 +232,9 @@ func TestCreateRemoteEndpointOverlay(t *testing.T) {
proxier.syncProxyRules() proxier.syncProxyRules()
ep := proxier.endpointsMap[svcPortName][0] ep := proxier.endpointsMap[svcPortName][0]
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if !ok { if !ok {
t.Errorf("Failed to cast endpointsInfo %q", svcPortName.String()) t.Errorf("Failed to cast endpointInfo %q", svcPortName.String())
} else { } else {
if epInfo.hnsID != "EPID-3" { if epInfo.hnsID != "EPID-3" {
@ -296,9 +296,9 @@ func TestCreateRemoteEndpointL2Bridge(t *testing.T) {
proxier.setInitialized(true) proxier.setInitialized(true)
proxier.syncProxyRules() proxier.syncProxyRules()
ep := proxier.endpointsMap[svcPortName][0] ep := proxier.endpointsMap[svcPortName][0]
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if !ok { if !ok {
t.Errorf("Failed to cast endpointsInfo %q", svcPortName.String()) t.Errorf("Failed to cast endpointInfo %q", svcPortName.String())
} else { } else {
if epInfo.hnsID != endpointGuid1 { if epInfo.hnsID != endpointGuid1 {
@ -389,9 +389,9 @@ func TestSharedRemoteEndpointDelete(t *testing.T) {
proxier.setInitialized(true) proxier.setInitialized(true)
proxier.syncProxyRules() proxier.syncProxyRules()
ep := proxier.endpointsMap[svcPortName1][0] ep := proxier.endpointsMap[svcPortName1][0]
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if !ok { if !ok {
t.Errorf("Failed to cast endpointsInfo %q", svcPortName1.String()) t.Errorf("Failed to cast endpointInfo %q", svcPortName1.String())
} else { } else {
if epInfo.hnsID != endpointGuid1 { if epInfo.hnsID != endpointGuid1 {
@ -439,9 +439,9 @@ func TestSharedRemoteEndpointDelete(t *testing.T) {
proxier.syncProxyRules() proxier.syncProxyRules()
ep = proxier.endpointsMap[svcPortName1][0] ep = proxier.endpointsMap[svcPortName1][0]
epInfo, ok = ep.(*endpointsInfo) epInfo, ok = ep.(*endpointInfo)
if !ok { if !ok {
t.Errorf("Failed to cast endpointsInfo %q", svcPortName1.String()) t.Errorf("Failed to cast endpointInfo %q", svcPortName1.String())
} else { } else {
if epInfo.hnsID != endpointGuid1 { if epInfo.hnsID != endpointGuid1 {
@ -534,9 +534,9 @@ func TestSharedRemoteEndpointUpdate(t *testing.T) {
proxier.setInitialized(true) proxier.setInitialized(true)
proxier.syncProxyRules() proxier.syncProxyRules()
ep := proxier.endpointsMap[svcPortName1][0] ep := proxier.endpointsMap[svcPortName1][0]
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if !ok { if !ok {
t.Errorf("Failed to cast endpointsInfo %q", svcPortName1.String()) t.Errorf("Failed to cast endpointInfo %q", svcPortName1.String())
} else { } else {
if epInfo.hnsID != endpointGuid1 { if epInfo.hnsID != endpointGuid1 {
@ -613,10 +613,10 @@ func TestSharedRemoteEndpointUpdate(t *testing.T) {
proxier.syncProxyRules() proxier.syncProxyRules()
ep = proxier.endpointsMap[svcPortName1][0] ep = proxier.endpointsMap[svcPortName1][0]
epInfo, ok = ep.(*endpointsInfo) epInfo, ok = ep.(*endpointInfo)
if !ok { if !ok {
t.Errorf("Failed to cast endpointsInfo %q", svcPortName1.String()) t.Errorf("Failed to cast endpointInfo %q", svcPortName1.String())
} else { } else {
if epInfo.hnsID != endpointGuid1 { if epInfo.hnsID != endpointGuid1 {
@ -913,9 +913,9 @@ func TestEndpointSlice(t *testing.T) {
} }
ep := proxier.endpointsMap[svcPortName][0] ep := proxier.endpointsMap[svcPortName][0]
epInfo, ok := ep.(*endpointsInfo) epInfo, ok := ep.(*endpointInfo)
if !ok { if !ok {
t.Errorf("Failed to cast endpointsInfo %q", svcPortName.String()) t.Errorf("Failed to cast endpointInfo %q", svcPortName.String())
} else { } else {
if epInfo.hnsID != "EPID-3" { if epInfo.hnsID != "EPID-3" {