github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Wire context for cert controllers

Browse Source 
All the controllers should use context for signalling termination of communication with API server. Once kcm cancels context all the cert controllers which are started via kcm should cancel the APIServer request in flight instead of hanging around.
This commit is contained in:
Ravi Gudimetla
2022-03-07 09:23:52 -05:00
parent 8b84a793b3
commit 72a62f47f7
13 changed files with 73 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
pkg/controller/certificates/certificate_controller.go
View File

@@ -19,6 +19,7 @@ limitations under the License.
package certificates
import (
"context"
"fmt"
"time"
@@ -48,7 +49,7 @@ type CertificateController struct {
csrLister certificateslisters.CertificateSigningRequestLister
csrsSynced cache.InformerSynced
handler func(*certificates.CertificateSigningRequest) error
handler func(context.Context, *certificates.CertificateSigningRequest) error
queue workqueue.RateLimitingInterface
}
@@ -57,7 +58,7 @@ func NewCertificateController(
name string,
kubeClient clientset.Interface,
csrInformer certificatesinformers.CertificateSigningRequestInformer,
handler func(*certificates.CertificateSigningRequest) error,
handler func(context.Context, *certificates.CertificateSigningRequest) error,
) *CertificateController {
// Send events to the apiserver
eventBroadcaster := record.NewBroadcaster()
@@ -111,39 +112,39 @@ func NewCertificateController(
}
// Run the main goroutine responsible for watching and syncing jobs.
func (cc *CertificateController) Run(workers int, stopCh <-chan struct{}) {
func (cc *CertificateController) Run(ctx context.Context, workers int) {
defer utilruntime.HandleCrash()
defer cc.queue.ShutDown()
klog.Infof("Starting certificate controller %q", cc.name)
defer klog.Infof("Shutting down certificate controller %q", cc.name)
if !cache.WaitForNamedCacheSync(fmt.Sprintf("certificate-%s", cc.name), stopCh, cc.csrsSynced) {
if !cache.WaitForNamedCacheSync(fmt.Sprintf("certificate-%s", cc.name), ctx.Done(), cc.csrsSynced) {
return
}
for i := 0; i < workers; i++ {
go wait.Until(cc.worker, time.Second, stopCh)
go wait.UntilWithContext(ctx, cc.worker, time.Second)
}
<-stopCh
<-ctx.Done()
}
// worker runs a thread that dequeues CSRs, handles them, and marks them done.
func (cc *CertificateController) worker() {
for cc.processNextWorkItem() {
func (cc *CertificateController) worker(ctx context.Context) {
for cc.processNextWorkItem(ctx) {
}
}
// processNextWorkItem deals with one key off the queue. It returns false when it's time to quit.
func (cc *CertificateController) processNextWorkItem() bool {
func (cc *CertificateController) processNextWorkItem(ctx context.Context) bool {
cKey, quit := cc.queue.Get()
if quit {
return false
}
defer cc.queue.Done(cKey)
if err := cc.syncFunc(cKey.(string)); err != nil {
if err := cc.syncFunc(ctx, cKey.(string)); err != nil {
cc.queue.AddRateLimited(cKey)
if _, ignorable := err.(ignorableError); !ignorable {
utilruntime.HandleError(fmt.Errorf("Sync %v failed with : %v", cKey, err))
@@ -167,7 +168,7 @@ func (cc *CertificateController) enqueueCertificateRequest(obj interface{}) {
cc.queue.Add(key)
}
func (cc *CertificateController) syncFunc(key string) error {
func (cc *CertificateController) syncFunc(ctx context.Context, key string) error {
startTime := time.Now()
defer func() {
klog.V(4).Infof("Finished syncing certificate request %q (%v)", key, time.Since(startTime))
@@ -188,7 +189,7 @@ func (cc *CertificateController) syncFunc(key string) error {
// need to operate on a copy so we don't mutate the csr in the shared cache
csr = csr.DeepCopy()
return cc.handler(csr)
return cc.handler(ctx, csr)
}
// IgnorableError returns an error that we shouldn't handle (i.e. log) because