Skip to content
This repository was archived by the owner on Dec 6, 2024. It is now read-only.

Commit 9bf426e

Browse files
wlan0wlan0
committed
fix controller logic to prevent concurrent map reads and writes
1 parent 5e1814a commit 9bf426e

File tree

3 files changed

+27
-29
lines changed

3 files changed

+27
-29
lines changed

Diff for: ‎controller/controller.go

+24-29
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@ type ObjectStorageController struct {
103103

104104
lockerLock sync.Mutex
105105
locker map[types.UID]*sync.Mutex
106-
opMap map[types.UID]interface{}
106+
opMap *sync.Map
107107
}
108108

109109
func NewDefaultObjectStorageController(identity string, leaderLockName string, threads int) (*ObjectStorageController, error) {
@@ -162,7 +162,7 @@ func NewObjectStorageControllerWithClientset(identity string, leaderLockName str
162162
RenewDeadline: 15 * time.Second,
163163
RetryPeriod: 5 * time.Second,
164164

165-
opMap: map[types.UID]interface{}{},
165+
opMap: &sync.Map{},
166166
}, nil
167167
}
168168

@@ -253,18 +253,19 @@ func (c *ObjectStorageController) processNextItem(ctx context.Context) bool {
253253

254254
uuid := uuidInterface.(types.UID)
255255
var err error
256-
// With the lock below in place, we can safely tell the queue that we are done
257-
// processing this item. The lock will ensure that multiple items of the same
258-
// name and kind do not get processed simultaneously
256+
259257
defer c.queue.Done(uuid)
260258

259+
op, ok := c.opMap.Load(uuid)
260+
if !ok {
261+
panic("unreachable code")
262+
}
263+
261264
// Ensure that multiple operations on different versions of the same object
262265
// do not happen in parallel
263266
c.OpLock(uuid)
264267
defer c.OpUnlock(uuid)
265268

266-
op := c.opMap[uuid]
267-
268269
switch o := op.(type) {
269270
case addOp:
270271
add := *o.AddFunc
@@ -332,12 +333,12 @@ func (c *ObjectStorageController) GetOpLock(op types.UID) *sync.Mutex {
332333
}
333334

334335
// handleErr checks if an error happened and makes sure we will retry later.
335-
func (c *ObjectStorageController) handleErr(err error, op interface{}) {
336+
func (c *ObjectStorageController) handleErr(err error, uuid types.UID) {
336337
if err == nil {
337-
c.queue.Forget(op)
338+
c.opMap.Delete(uuid)
338339
return
339340
}
340-
c.queue.AddRateLimited(op)
341+
c.queue.AddRateLimited(uuid)
341342
}
342343

343344
func (c *ObjectStorageController) runController(ctx context.Context) {
@@ -349,7 +350,7 @@ func (c *ObjectStorageController) runController(ctx context.Context) {
349350
cfg := &cache.Config{
350351
Queue: cache.NewDeltaFIFOWithOptions(cache.DeltaFIFOOptions{
351352
KnownObjects: indexer,
352-
EmitDeltaTypeReplaced: true,
353+
EmitDeltaTypeReplaced: false,
353354
}),
354355
ListerWatcher: lw,
355356
ObjectType: objType,
@@ -370,15 +371,14 @@ func (c *ObjectStorageController) runController(ctx context.Context) {
370371
}
371372

372373
uuid := d.Object.(metav1.Object).GetUID()
373-
c.OpLock(uuid)
374-
defer c.OpUnlock(uuid)
375-
c.opMap[uuid] = updateOp{
374+
375+
c.opMap.Store(uuid, updateOp{
376376
OldObject: old,
377377
NewObject: d.Object,
378378
UpdateFunc: &update,
379379
Key: key,
380380
Indexer: indexer,
381-
}
381+
})
382382
c.queue.Add(uuid)
383383
} else {
384384
key, err := cache.MetaNamespaceKeyFunc(d.Object)
@@ -387,20 +387,17 @@ func (c *ObjectStorageController) runController(ctx context.Context) {
387387
}
388388

389389
uuid := d.Object.(metav1.Object).GetUID()
390-
c.OpLock(uuid)
391-
defer c.OpUnlock(uuid)
392-
393-
// If an update to the k8s object happens before add has succeeded
394-
if op, ok := c.opMap[uuid]; ok {
395-
if _, ok := op.(updateOp); ok {
396-
return fmt.Errorf("cannot add already added object: %s", key)
397-
}
398-
}
399-
c.opMap[uuid] = addOp{
390+
391+
if op, ok := c.opMap.LoadOrStore(uuid, addOp{
400392
Object: d.Object,
401393
AddFunc: &add,
402394
Key: key,
403395
Indexer: indexer,
396+
}); ok { // If an update to the k8s object happens before add has succeeded
397+
if _, ok := op.(updateOp); ok {
398+
err := fmt.Errorf("cannot add already added object: %s", key)
399+
return err
400+
}
404401
}
405402
c.queue.Add(uuid)
406403
}
@@ -411,14 +408,12 @@ func (c *ObjectStorageController) runController(ctx context.Context) {
411408
}
412409

413410
uuid := d.Object.(metav1.Object).GetUID()
414-
c.OpLock(uuid)
415-
defer c.OpUnlock(uuid)
416-
c.opMap[uuid] = deleteOp{
411+
c.opMap.Store(uuid, deleteOp{
417412
Object: d.Object,
418413
DeleteFunc: &delete,
419414
Key: key,
420415
Indexer: indexer,
421-
}
416+
})
422417
c.queue.Add(uuid)
423418
}
424419
}

Diff for: ‎go.mod

+1
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ require (
66
github.com/go-openapi/spec v0.19.12
77
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
88
github.com/onsi/ginkgo v1.14.2 // indirect
9+
github.com/spf13/cobra v0.0.5
910
github.com/spf13/viper v1.3.2
1011
golang.org/x/time v0.0.0-20191024005414-555d28b269f0
1112
k8s.io/api v0.19.4

Diff for: ‎go.sum

+2
Original file line numberDiff line numberDiff line change
@@ -228,6 +228,7 @@ github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:
228228
github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
229229
github.com/imdario/mergo v0.3.9 h1:UauaLniWCFHWd+Jp9oCEkTBj8VO/9DKg3PV3VCNMDIg=
230230
github.com/imdario/mergo v0.3.9/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
231+
github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
231232
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
232233
github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
233234
github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
@@ -325,6 +326,7 @@ github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTd
325326
github.com/spf13/cast v1.3.0 h1:oget//CVOEoFewqQxwr0Ej5yjygnqGkvggSE/gB35Q8=
326327
github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
327328
github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
329+
github.com/spf13/cobra v0.0.5 h1:f0B+LkLX6DtmRH1isoNA9VTtNUK9K8xYd28JNNfOv/s=
328330
github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
329331
github.com/spf13/jwalterweatherman v1.0.0 h1:XHEdyB+EcvlqZamSM4ZOMGlc93t6AcsBEu9Gc1vn7yk=
330332
github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=

0 commit comments

Comments
 (0)