Skip to content

Commit 4505d5b

Browse files
kumarvin123kumarvin123
committed
Changes to Proxy common code
1 parent 71c352d commit 4505d5b

File tree

8 files changed

+61
-41
lines changed

8 files changed

+61
-41
lines changed

‎pkg/proxy/endpoints.go‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,10 @@ func newBaseEndpointInfo(IP string, port int, isLocal bool, topology map[string]
9494

9595
type makeEndpointFunc func(info *BaseEndpointInfo) Endpoint
9696

97+
// This handler is invoked by the apply function on every change. This function should not modify the
98+
// EndpointsMap's but just use the changes for any Proxier specific cleanup.
99+
type processEndpointsMapChangeFunc func(oldEndpointsMap, newEndpointsMap EndpointsMap)
100+
97101
// EndpointChangeTracker carries state about uncommitted changes to an arbitrary number of
98102
// Endpoints, keyed by their namespace and name.
99103
type EndpointChangeTracker struct {
@@ -104,7 +108,8 @@ type EndpointChangeTracker struct {
104108
// items maps a service to is endpointsChange.
105109
items map[types.NamespacedName]*endpointsChange
106110
// makeEndpointInfo allows proxier to inject customized information when processing endpoint.
107-
makeEndpointInfo makeEndpointFunc
111+
makeEndpointInfo makeEndpointFunc
112+
processEndpointsMapChange processEndpointsMapChangeFunc
108113
// endpointSliceCache holds a simplified version of endpoint slices.
109114
endpointSliceCache *EndpointSliceCache
110115
// isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
@@ -116,14 +121,15 @@ type EndpointChangeTracker struct {
116121
}
117122

118123
// NewEndpointChangeTracker initializes an EndpointsChangeMap
119-
func NewEndpointChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc, isIPv6Mode *bool, recorder record.EventRecorder, endpointSlicesEnabled bool) *EndpointChangeTracker {
124+
func NewEndpointChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc, isIPv6Mode *bool, recorder record.EventRecorder, endpointSlicesEnabled bool, processEndpointsMapChange processEndpointsMapChangeFunc) *EndpointChangeTracker {
120125
ect := &EndpointChangeTracker{
121-
hostname: hostname,
122-
items: make(map[types.NamespacedName]*endpointsChange),
123-
makeEndpointInfo: makeEndpointInfo,
124-
isIPv6Mode: isIPv6Mode,
125-
recorder: recorder,
126-
lastChangeTriggerTimes: make(map[types.NamespacedName][]time.Time),
126+
hostname: hostname,
127+
items: make(map[types.NamespacedName]*endpointsChange),
128+
makeEndpointInfo: makeEndpointInfo,
129+
isIPv6Mode: isIPv6Mode,
130+
recorder: recorder,
131+
lastChangeTriggerTimes: make(map[types.NamespacedName][]time.Time),
132+
processEndpointsMapChange: processEndpointsMapChange,
127133
}
128134
if endpointSlicesEnabled {
129135
ect.endpointSliceCache = NewEndpointSliceCache(hostname, isIPv6Mode, recorder, makeEndpointInfo)
@@ -388,6 +394,7 @@ func (ect *EndpointChangeTracker) endpointsToEndpointsMap(endpoints *v1.Endpoint
388394
// The changes map is cleared after applying them.
389395
// In addition it returns (via argument) and resets the lastChangeTriggerTimes for all endpoints
390396
// that were changed and will result in syncing the proxy rules.
397+
// apply triggers processEndpointsMapChange on every change.
391398
func (em EndpointsMap) apply(ect *EndpointChangeTracker, staleEndpoints *[]ServiceEndpoint,
392399
staleServiceNames *[]ServicePortName, lastChangeTriggerTimes *map[types.NamespacedName][]time.Time) {
393400
if ect == nil {
@@ -396,6 +403,9 @@ func (em EndpointsMap) apply(ect *EndpointChangeTracker, staleEndpoints *[]Servi
396403

397404
changes := ect.checkoutChanges()
398405
for _, change := range changes {
406+
if ect.processEndpointsMapChange != nil {
407+
ect.processEndpointsMapChange(change.previous, change.current)
408+
}
399409
em.unmerge(change.previous)
400410
em.merge(change.current)
401411
detectStaleConnections(change.previous, change.current, staleEndpoints, staleServiceNames)

‎pkg/proxy/endpoints_test.go‎

Lines changed: 16 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ func makeTestEndpoints(namespace, name string, eptFunc func(*v1.Endpoints)) *v1.
135135

136136
// This is a coarse test, but it offers some modicum of confidence as the code is evolved.
137137
func TestEndpointsToEndpointsMap(t *testing.T) {
138-
epTracker := NewEndpointChangeTracker("test-hostname", nil, nil, nil, false)
138+
epTracker := NewEndpointChangeTracker("test-hostname", nil, nil, nil, false, nil)
139139

140140
trueVal := true
141141
falseVal := false
@@ -1446,7 +1446,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
14461446
// test starting from an empty state
14471447
"add a simple slice that doesn't already exist": {
14481448
startingSlices: []*discovery.EndpointSlice{},
1449-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1449+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
14501450
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
14511451
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
14521452
paramRemoveSlice: false,
@@ -1469,7 +1469,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
14691469
startingSlices: []*discovery.EndpointSlice{
14701470
generateEndpointSlice("svc1", "ns1", 1, 3, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
14711471
},
1472-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1472+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
14731473
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
14741474
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
14751475
paramRemoveSlice: false,
@@ -1481,7 +1481,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
14811481
startingSlices: []*discovery.EndpointSlice{
14821482
generateEndpointSlice("svc1", "ns1", 1, 3, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
14831483
},
1484-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1484+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
14851485
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
14861486
paramEndpointSlice: fqdnSlice,
14871487
paramRemoveSlice: false,
@@ -1494,7 +1494,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
14941494
generateEndpointSlice("svc1", "ns1", 1, 3, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
14951495
generateEndpointSlice("svc1", "ns1", 2, 2, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
14961496
},
1497-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1497+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
14981498
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
14991499
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 5, 999, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15001500
paramRemoveSlice: false,
@@ -1526,7 +1526,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
15261526
generateEndpointSlice("svc1", "ns1", 1, 3, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15271527
generateEndpointSlice("svc1", "ns1", 2, 2, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15281528
},
1529-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1529+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
15301530
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
15311531
paramEndpointSlice: generateEndpointSliceWithOffset("svc1", "ns1", 3, 1, 5, 999, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80)}),
15321532
paramRemoveSlice: false,
@@ -1556,7 +1556,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
15561556
generateEndpointSlice("svc1", "ns1", 1, 3, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15571557
generateEndpointSlice("svc1", "ns1", 2, 2, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15581558
},
1559-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1559+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
15601560
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
15611561
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 5, 999, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15621562
paramRemoveSlice: true,
@@ -1578,7 +1578,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
15781578
generateEndpointSlice("svc1", "ns1", 1, 5, 999, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15791579
generateEndpointSlice("svc1", "ns1", 2, 2, 999, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15801580
},
1581-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1581+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
15821582
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
15831583
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 3, 5, 999, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15841584
paramRemoveSlice: true,
@@ -1590,7 +1590,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
15901590
startingSlices: []*discovery.EndpointSlice{
15911591
generateEndpointSlice("svc1", "ns1", 1, 3, 999, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15921592
},
1593-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1593+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
15941594
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
15951595
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 1, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
15961596
paramRemoveSlice: false,
@@ -1602,7 +1602,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
16021602
startingSlices: []*discovery.EndpointSlice{
16031603
generateEndpointSlice("svc1", "ns1", 1, 2, 1, []string{"host1", "host2"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
16041604
},
1605-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1605+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
16061606
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
16071607
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 2, 999, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
16081608
paramRemoveSlice: false,
@@ -1624,7 +1624,7 @@ func TestEndpointSliceUpdate(t *testing.T) {
16241624
generateEndpointSlice("svc1", "ns1", 1, 3, 2, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
16251625
generateEndpointSlice("svc1", "ns1", 2, 2, 2, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
16261626
},
1627-
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true),
1627+
endpointChangeTracker: NewEndpointChangeTracker("host1", nil, nil, nil, true, nil),
16281628
namespacedName: types.NamespacedName{Name: "svc1", Namespace: "ns1"},
16291629
paramEndpointSlice: generateEndpointSlice("svc1", "ns1", 1, 3, 3, []string{"host1"}, []*int32{utilpointer.Int32Ptr(80), utilpointer.Int32Ptr(443)}),
16301630
paramRemoveSlice: false,
@@ -1683,20 +1683,20 @@ func TestCheckoutChanges(t *testing.T) {
16831683
pendingSlices []*discovery.EndpointSlice
16841684
}{
16851685
"empty slices": {
1686-
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, true),
1686+
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, true, nil),
16871687
expectedChanges: []*endpointsChange{},
16881688
useEndpointSlices: true,
16891689
appliedSlices: []*discovery.EndpointSlice{},
16901690
pendingSlices: []*discovery.EndpointSlice{},
16911691
},
16921692
"without slices, empty items": {
1693-
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, false),
1693+
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, false, nil),
16941694
expectedChanges: []*endpointsChange{},
16951695
items: map[types.NamespacedName]*endpointsChange{},
16961696
useEndpointSlices: false,
16971697
},
16981698
"without slices, simple items": {
1699-
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, false),
1699+
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, false, nil),
17001700
expectedChanges: []*endpointsChange{{
17011701
previous: EndpointsMap{
17021702
svcPortName0: []Endpoint{newTestEp("10.0.1.1:80", ""), newTestEp("10.0.1.2:80", "")},
@@ -1720,7 +1720,7 @@ func TestCheckoutChanges(t *testing.T) {
17201720
useEndpointSlices: false,
17211721
},
17221722
"adding initial slice": {
1723-
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, true),
1723+
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, true, nil),
17241724
expectedChanges: []*endpointsChange{{
17251725
previous: EndpointsMap{},
17261726
current: EndpointsMap{
@@ -1734,7 +1734,7 @@ func TestCheckoutChanges(t *testing.T) {
17341734
},
17351735
},
17361736
"removing port in update": {
1737-
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, true),
1737+
endpointChangeTracker: NewEndpointChangeTracker("", nil, nil, nil, true, nil),
17381738
expectedChanges: []*endpointsChange{{
17391739
previous: EndpointsMap{
17401740
svcPortName0: []Endpoint{newTestEp("10.0.1.1:80", "host1"), newTestEp("10.0.1.2:80", "host1")},

‎pkg/proxy/iptables/proxier.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -298,9 +298,9 @@ func NewProxier(ipt utiliptables.Interface,
298298
proxier := &Proxier{
299299
portsMap: make(map[utilproxy.LocalPort]utilproxy.Closeable),
300300
serviceMap: make(proxy.ServiceMap),
301-
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder),
301+
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder, nil),
302302
endpointsMap: make(proxy.EndpointsMap),
303-
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, &isIPv6, recorder, endpointSlicesEnabled),
303+
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, &isIPv6, recorder, endpointSlicesEnabled, nil),
304304
syncPeriod: syncPeriod,
305305
iptables: ipt,
306306
masqueradeAll: masqueradeAll,

‎pkg/proxy/iptables/proxier_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -353,9 +353,9 @@ func NewFakeProxier(ipt utiliptables.Interface, endpointSlicesEnabled bool) *Pro
353353
p := &Proxier{
354354
exec: &fakeexec.FakeExec{},
355355
serviceMap: make(proxy.ServiceMap),
356-
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, nil, nil),
356+
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, nil, nil, nil),
357357
endpointsMap: make(proxy.EndpointsMap),
358-
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, newEndpointInfo, nil, nil, endpointSlicesEnabled),
358+
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, newEndpointInfo, nil, nil, endpointSlicesEnabled, nil),
359359
iptables: ipt,
360360
masqueradeMark: "0x4000",
361361
localDetector: detectLocal,

‎pkg/proxy/ipvs/proxier.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -441,9 +441,9 @@ func NewProxier(ipt utiliptables.Interface,
441441
proxier := &Proxier{
442442
portsMap: make(map[utilproxy.LocalPort]utilproxy.Closeable),
443443
serviceMap: make(proxy.ServiceMap),
444-
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder),
444+
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder, nil),
445445
endpointsMap: make(proxy.EndpointsMap),
446-
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, nil, &isIPv6, recorder, endpointSlicesEnabled),
446+
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, nil, &isIPv6, recorder, endpointSlicesEnabled, nil),
447447
syncPeriod: syncPeriod,
448448
minSyncPeriod: minSyncPeriod,
449449
excludeCIDRs: parseExcludedCIDRs(excludeCIDRs),

‎pkg/proxy/ipvs/proxier_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -122,9 +122,9 @@ func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, ipset u
122122
p := &Proxier{
123123
exec: fexec,
124124
serviceMap: make(proxy.ServiceMap),
125-
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, nil, nil),
125+
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, nil, nil, nil),
126126
endpointsMap: make(proxy.EndpointsMap),
127-
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, nil, nil, nil, endpointSlicesEnabled),
127+
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, nil, nil, nil, endpointSlicesEnabled, nil),
128128
excludeCIDRs: excludeCIDRs,
129129
iptables: ipt,
130130
ipvs: ipvs,

‎pkg/proxy/service.go‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,10 @@ func (sct *ServiceChangeTracker) newBaseServiceInfo(port *v1.ServicePort, servic
198198

199199
type makeServicePortFunc func(*v1.ServicePort, *v1.Service, *BaseServiceInfo) ServicePort
200200

201+
// This handler is invoked by the apply function on every change. This function should not modify the
202+
// ServiceMap's but just use the changes for any Proxier specific cleanup.
203+
type processServiceMapChangeFunc func(previous, current ServiceMap)
204+
201205
// serviceChange contains all changes to services that happened since proxy rules were synced. For a single object,
202206
// changes are accumulated, i.e. previous is state from before applying the changes,
203207
// current is state after applying all of the changes.
@@ -214,19 +218,21 @@ type ServiceChangeTracker struct {
214218
// items maps a service to its serviceChange.
215219
items map[types.NamespacedName]*serviceChange
216220
// makeServiceInfo allows proxier to inject customized information when processing service.
217-
makeServiceInfo makeServicePortFunc
221+
makeServiceInfo makeServicePortFunc
222+
processServiceMapChange processServiceMapChangeFunc
218223
// isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
219224
isIPv6Mode *bool
220225
recorder record.EventRecorder
221226
}
222227

223228
// NewServiceChangeTracker initializes a ServiceChangeTracker
224-
func NewServiceChangeTracker(makeServiceInfo makeServicePortFunc, isIPv6Mode *bool, recorder record.EventRecorder) *ServiceChangeTracker {
229+
func NewServiceChangeTracker(makeServiceInfo makeServicePortFunc, isIPv6Mode *bool, recorder record.EventRecorder, processServiceMapChange processServiceMapChangeFunc) *ServiceChangeTracker {
225230
return &ServiceChangeTracker{
226-
items: make(map[types.NamespacedName]*serviceChange),
227-
makeServiceInfo: makeServiceInfo,
228-
isIPv6Mode: isIPv6Mode,
229-
recorder: recorder,
231+
items: make(map[types.NamespacedName]*serviceChange),
232+
makeServiceInfo: makeServiceInfo,
233+
isIPv6Mode: isIPv6Mode,
234+
recorder: recorder,
235+
processServiceMapChange: processServiceMapChange,
230236
}
231237
}
232238

@@ -337,10 +343,14 @@ func (sct *ServiceChangeTracker) serviceToServiceMap(service *v1.Service) Servic
337343

338344
// apply the changes to ServiceMap and update the stale udp cluster IP set. The UDPStaleClusterIP argument is passed in to store the
339345
// udp protocol service cluster ip when service is deleted from the ServiceMap.
346+
// apply triggers processServiceMapChange on every change.
340347
func (sm *ServiceMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP sets.String) {
341348
changes.lock.Lock()
342349
defer changes.lock.Unlock()
343350
for _, change := range changes.items {
351+
if changes.processServiceMapChange != nil {
352+
changes.processServiceMapChange(change.previous, change.current)
353+
}
344354
sm.merge(change.current)
345355
// filter out the Update event of current changes from previous changes before calling unmerge() so that can
346356
// skip deleting the Update events.

0 commit comments

Comments
 (0)