@@ -130,6 +130,8 @@ func (c *Controller) Start(stop <-chan struct{}) error {
130
130
c .Queue = c .MakeQueue ()
131
131
defer c .Queue .ShutDown () // needs to be outside the iife so that we shutdown after the stop channel is closed
132
132
133
+ ctx , cancel := context .WithCancel (context .Background ())
134
+
133
135
err := func () error {
134
136
defer c .mu .Unlock ()
135
137
@@ -170,8 +172,12 @@ func (c *Controller) Start(stop <-chan struct{}) error {
170
172
// Launch workers to process resources
171
173
c .Log .Info ("Starting workers" , "worker count" , c .MaxConcurrentReconciles )
172
174
for i := 0 ; i < c .MaxConcurrentReconciles ; i ++ {
173
- // Process work items
174
- go wait .Until (c .worker , c .JitterPeriod , stop )
175
+ // Run a worker thread that just dequeues items, processes them, and marks them done.
176
+ // It enforces that the reconcileHandler is never invoked concurrently with the same object.
177
+ go wait .Until (func () {
178
+ for c .processNextWorkItem (ctx ) {
179
+ }
180
+ }, c .JitterPeriod , stop )
175
181
}
176
182
177
183
c .Started = true
@@ -180,22 +186,17 @@ func (c *Controller) Start(stop <-chan struct{}) error {
180
186
if err != nil {
181
187
return err
182
188
}
189
+ context .WithCancel (context .Background ())
183
190
184
191
<- stop
192
+ cancel ()
185
193
c .Log .Info ("Stopping workers" )
186
194
return nil
187
195
}
188
196
189
- // worker runs a worker thread that just dequeues items, processes them, and marks them done.
190
- // It enforces that the reconcileHandler is never invoked concurrently with the same object.
191
- func (c * Controller ) worker () {
192
- for c .processNextWorkItem () {
193
- }
194
- }
195
-
196
197
// processNextWorkItem will read a single work item off the workqueue and
197
198
// attempt to process it, by calling the reconcileHandler.
198
- func (c * Controller ) processNextWorkItem () bool {
199
+ func (c * Controller ) processNextWorkItem (ctx context. Context ) bool {
199
200
obj , shutdown := c .Queue .Get ()
200
201
if shutdown {
201
202
// Stop working
@@ -210,10 +211,10 @@ func (c *Controller) processNextWorkItem() bool {
210
211
// period.
211
212
defer c .Queue .Done (obj )
212
213
213
- return c .reconcileHandler (obj )
214
+ return c .reconcileHandler (ctx , obj )
214
215
}
215
216
216
- func (c * Controller ) reconcileHandler (obj interface {}) bool {
217
+ func (c * Controller ) reconcileHandler (ctx context. Context , obj interface {}) bool {
217
218
// Update metrics after processing each item
218
219
reconcileStartTS := time .Now ()
219
220
defer func () {
@@ -233,7 +234,7 @@ func (c *Controller) reconcileHandler(obj interface{}) bool {
233
234
}
234
235
235
236
log := c .Log .WithValues ("name" , req .Name , "namespace" , req .Namespace )
236
- ctx := logf .IntoContext (context . Background () , log )
237
+ ctx := logf .IntoContext (ctx , log )
237
238
238
239
// RunInformersAndControllers the syncHandler, passing it the namespace/Name string of the
239
240
// resource to be synced.
0 commit comments