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

ensure that indexer has latest instance of an object #42

Merged
merged 1 commit into from
May 5, 2021
Merged

ensure that indexer has latest instance of an object #42

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 9 additions & 14 deletions controller/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,9 +158,9 @@ func NewObjectStorageControllerWithClientset(identity string, leaderLockName str

ResyncPeriod: 30 * time.Second,
// leader election
LeaseDuration: 60 * time.Second,
RenewDeadline: 30 * time.Second,
RetryPeriod: 15 * time.Second,
LeaseDuration: 150 * time.Second,
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It was being renewed too aggressively.

RenewDeadline: 120 * time.Second,
RetryPeriod: 60 * time.Second,

opMap: &sync.Map{},
}, nil
Expand Down Expand Up @@ -258,7 +258,7 @@ func (c *ObjectStorageController) processNextItem(ctx context.Context) bool {

op, ok := c.opMap.Load(uuid)
if !ok {
panic("unreachable code")
return true
}

// Ensure that multiple operations on different versions of the same object
Expand All @@ -270,22 +270,16 @@ func (c *ObjectStorageController) processNextItem(ctx context.Context) bool {
case addOp:
add := *o.AddFunc
err = add(ctx, o.Object)
if err == nil {
o.Indexer.Add(o.Object)
}
o.Indexer.Add(o.Object)
case updateOp:
update := *o.UpdateFunc
err = update(ctx, o.OldObject, o.NewObject)
if err == nil {
o.Indexer.Update(o.NewObject)
}
o.Indexer.Update(o.NewObject)
case deleteOp:
delete := *o.DeleteFunc
err = delete(ctx, o.Object)
if err == nil {
o.Indexer.Delete(o.Object)
c.opMap.Delete(uuid)
}
o.Indexer.Delete(o.Object)
c.opMap.Delete(uuid)
default:
panic("unknown item in queue")
}
Expand Down Expand Up @@ -321,6 +315,7 @@ func (c *ObjectStorageController) GetOpLock(op types.UID) *sync.Mutex {
// handleErr checks if an error happened and makes sure we will retry later.
func (c *ObjectStorageController) handleErr(err error, uuid types.UID) {
if err == nil {
c.opMap.Delete(uuid)
return
}
c.queue.AddRateLimited(uuid)
Expand Down