Skip to content

Commit

Permalink
Change enqueuing items to use Add instead of AddRateLimited.
Browse files Browse the repository at this point in the history
Calling 'AddRateLimited' appears to indicate that an error has occurred
for the rate limiting queue.  This triggers exponential backoff
behavior, so if a large number of elements (20+) are owned by a resource
and the controller happens to encounter an error, this will cause
exponential backoff to reach the 1000 second limit.
  • Loading branch information
briantkennedy committed Jun 26, 2018
1 parent 05f0100 commit 4ee5550
Show file tree
Hide file tree
Showing 2 changed files with 8 additions and 8 deletions.
14 changes: 7 additions & 7 deletions pkg/controller/eventhandlers/eventhandlers.go
Original file line number Diff line number Diff line change
Expand Up @@ -53,37 +53,37 @@ func (mp MapAndEnqueue) Get(r workqueue.RateLimitingInterface) cache.ResourceEve
return
}
}
mp.addRateLimited(r, obj)
mp.add(r, obj)
},
UpdateFunc: func(old, obj interface{}) {
for _, p := range mp.Predicates {
if !p.HandleUpdate(old, obj) {
return
}
}
mp.addRateLimited(r, obj)
mp.add(r, obj)
},
DeleteFunc: func(obj interface{}) {
for _, p := range mp.Predicates {
if !p.HandleDelete(obj) {
return
}
}
mp.addRateLimited(r, obj)
mp.add(r, obj)
},
}
}

// addRateLimited maps the obj to a string. If the string is non-empty, it is enqueued.
func (mp MapAndEnqueue) addRateLimited(r workqueue.RateLimitingInterface, obj interface{}) {
// add maps the obj to a string. If the string is non-empty, it is enqueued.
func (mp MapAndEnqueue) add(r workqueue.RateLimitingInterface, obj interface{}) {
if mp.Map != nil {
if k := mp.Map(obj); len(k) > 0 {
r.AddRateLimited(k)
r.Add(k)
}
}
if mp.MultiMap != nil {
for _, k := range mp.MultiMap(obj) {
r.AddRateLimited(k.Namespace + "/" + k.Name)
r.Add(k.Namespace + "/" + k.Name)
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/controller/listeningqueue.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ type listeningQueue struct {
func (q *listeningQueue) watchChannel(source <-chan string) error {
go func() {
for msg := range source {
q.AddRateLimited(msg)
q.Add(msg)
}
}()
return nil
Expand Down

0 comments on commit 4ee5550

Please sign in to comment.