@@ -130,6 +130,9 @@ 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
+ defer cancel ()
135
+
133
136
err := func () error {
134
137
defer c .mu .Unlock ()
135
138
@@ -170,8 +173,12 @@ func (c *Controller) Start(stop <-chan struct{}) error {
170
173
// Launch workers to process resources
171
174
c .Log .Info ("Starting workers" , "worker count" , c .MaxConcurrentReconciles )
172
175
for i := 0 ; i < c .MaxConcurrentReconciles ; i ++ {
173
- // Process work items
174
- go wait .Until (c .worker , c .JitterPeriod , stop )
176
+ // Run a worker thread that just dequeues items, processes them, and marks them done.
177
+ // It enforces that the reconcileHandler is never invoked concurrently with the same object.
178
+ go wait .Until (func () {
179
+ for c .processNextWorkItem (ctx ) {
180
+ }
181
+ }, c .JitterPeriod , stop )
175
182
}
176
183
177
184
c .Started = true
@@ -180,22 +187,16 @@ func (c *Controller) Start(stop <-chan struct{}) error {
180
187
if err != nil {
181
188
return err
182
189
}
190
+ context .WithCancel (context .Background ())
183
191
184
192
<- stop
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