-
Notifications
You must be signed in to change notification settings - Fork 39.3k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
client-go workqueue example #44320
client-go workqueue example #44320
Conversation
Hi @rmohr. Thanks for your PR. I'm waiting for a kubernetes member to verify that this patch is reasonable to test. If it is, they should reply with Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. I understand the commands that are listed here. |
Follow up for kubernetes/client-go#65. |
glog.Info("Starting Pod controller") | ||
|
||
go c.informer.Run(stopCh) | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You need to wait for your caches to sync before running any worker. See for example here
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Since I only have one cache, I don't think so. The key is dirtied for every entry anyway, right?
But maybe it would make the example more complete. What do you think?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The number of caches does not matter. When a cluster restarts, controllers that don't wait for their caches to sync may end up doing all sorts of weird things.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Really, could you elaborate on that?
I think the informer has a well defined behaviour when you are using it's callback. So whenever something is added/updated/deleted in the underlying store, we dirty the key via the callbacks. This means that when we process the key in the controller, and fetch the item from the informer store, we get the object from this event or a newer version. Or do I miss something here?
But I totally agree, that when you have "secondary" caches, it is vital to sync.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You are right, I was thinking of a bug we had with a secondary cache. Yet, I think it still makes sense to have the check here for all caches.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Agreed.
// Tricky what to do in this situation. One thing we can do, is enqueueing it a few times to | ||
// add some backoff delays on the invalid key. This way we avoid hotlooping | ||
// on invalid keys. | ||
if queue.NumRequeues(key) < 5 { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I can't see where the business logic of your controller exists. As a convention we use a separate function for it which surfaces an error at this level and then another function that is your error handling code which also facilitates requeues vs drops of objects. See #8 in https://github.com/kubernetes/community/blob/master/contributors/devel/controllers.md
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thx. Should be addressed now. @Kargakis could you have another look?
} | ||
|
||
runtime.HandleError(err) | ||
glog.Infof("Dropping deployment %q out of the queue: %v", key, err) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You are missing c.queue.Forget(key)
here
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
|
||
// Wait for all involved caches to be synced, before processing items from the queue is started | ||
if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) { | ||
glog.Fatal("Timed out waiting for caches to sync") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
instead of Fatal, use utilruntime.HandleError everywhere. You should also return if this path is accessed
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
|
||
func (c *Controller) Run(threadiness int, stopCh chan struct{}) { | ||
// Let the workers stop when we are done | ||
defer c.queue.ShutDown() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Missing a defer utilruntime.HandleCrash()
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
/lgtm |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
A few nits. Thanks.
// create the workqueue | ||
queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()) | ||
|
||
// Bind the workqueue to a cache with the help of an informer. This way we make sure than whenever the cache |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
s/than/that
controller := NewController(queue, indexer, informer) | ||
|
||
// We can now warm up the cache for initial synchronization | ||
// Le's suppose that we knew about a pod mypod on our last run, so we add it to the cache |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
s/Le/Let
|
||
// Bind the workqueue to a cache with the help of an informer. This way we make sure than whenever the cache | ||
// is updated, the pod key is added to the workqueue. Note than when we finally process the item from the | ||
// workqueue we might see a newer version of the Pod than the version which was responsible for triggering the update. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: could you wrap this line?
flag.StringVar(&master, "master", "", "master url") | ||
flag.Parse() | ||
|
||
// Create the connection |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: s/Create/creates
|
||
func (c *Controller) handleErr(err error, key interface{}) { | ||
if err == nil { | ||
c.queue.Forget(key) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you add a line of comment for Forget and later at AddRateLimited?
@caesarxuchao comments are updated 👍 |
/lgtm |
Thanks for the example, @rmohr ! |
need to run |
@rmohr Could you run hack/update-bazel.sh and squash the commits? |
@caesarxuchao squashed and ran hack/update-bazel.sh |
} | ||
c.queue.Forget(key) | ||
runtime.HandleError(err) | ||
glog.Infof("Dropping deployment %q out of the queue: %v", key, err) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
s/deployment/pod/
} | ||
|
||
if c.queue.NumRequeues(key) < 5 { | ||
glog.Infof("Error syncing controller %v: %v", key, err) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
s/controller/pod/
return true | ||
} | ||
|
||
func (c *Controller) syncToStdout(key string) error { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Add a comment here that this function is responsible for handling the business logic of the controller
Demonstrates how to compose a controller out of cache.Controller, cache.Indexer and a workqueue.
/lgtm |
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: caesarxuchao, kargakis, rmohr
Needs approval from an approver in each of these OWNERS Files:
You can indicate your approval by writing |
Automatic merge from submit-queue |
Demonstrates how to compose a controller out of cache.Controller,
cache.Indexer and a workqueue.