Merge pull request #53907 from mikedanese/base-delay

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

sarapprover: increase base delay of per item rate limit

from 5 miliseconds to 1 second

fixes https://github.com/kubernetes/kubernetes/issues/53734

pkg/controller/certificates/BUILD

@@ -15,6 +15,7 @@ go_library(
     deps = [
         "//pkg/controller:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/github.com/juju/ratelimit:go_default_library",
         "//vendor/k8s.io/api/certificates/v1beta1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",

pkg/controller/certificates/approver/sarapprove.go

@@ -115,7 +115,7 @@ func (a *sarApprover) handle(csr *capi.CertificateSigningRequest) error {
 	}
 
 	if len(tried) != 0 {
-		return fmt.Errorf("recognized csr %q as %v but subject access review was not approved", csr.Name, tried)
+		return certificates.IgnorableError("recognized csr %q as %v but subject access review was not approved", csr.Name, tried)
 	}
 
 	return nil

pkg/controller/certificates/certificate_controller.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller"
 
 	"github.com/golang/glog"
+	"github.com/juju/ratelimit"
 )
 
 type CertificateController struct {
@@ -61,7 +62,11 @@ func NewCertificateController(
 
 	cc := &CertificateController{
 		kubeClient: kubeClient,
-		queue:      workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "certificate"),
+		queue: workqueue.NewNamedRateLimitingQueue(workqueue.NewMaxOfRateLimiter(
+			workqueue.NewItemExponentialFailureRateLimiter(200*time.Millisecond, 1000*time.Second),
+			// 10 qps, 100 bucket size.  This is only for retry speed and its only the overall factor (not per item)
+			&workqueue.BucketRateLimiter{Bucket: ratelimit.NewBucketWithRate(float64(10), int64(100))},
+		), "certificate"),
 		handler: handler,
 	}
 
@@ -135,7 +140,11 @@ func (cc *CertificateController) processNextWorkItem() bool {
 
 	if err := cc.syncFunc(cKey.(string)); err != nil {
 		cc.queue.AddRateLimited(cKey)
+		if _, ignorable := err.(ignorableError); !ignorable {
 			utilruntime.HandleError(fmt.Errorf("Sync %v failed with : %v", cKey, err))
+		} else {
+			glog.V(4).Infof("Sync %v failed with : %v", cKey, err)
+		}
 		return true
 	}
 
@@ -181,3 +190,17 @@ func (cc *CertificateController) syncFunc(key string) error {
 
 	return cc.handler(csr)
 }
+
+// IgnorableError returns an error that we shouldn't handle (i.e. log) because
+// it's spammy and usually user error. Instead we will log these errors at a
+// higher log level. We still need to throw these errors to signal that the
+// sync should be retried.
+func IgnorableError(s string, args ...interface{}) ignorableError {
+	return ignorableError(fmt.Sprintf(s, args...))
+}
+
+type ignorableError string
+
+func (e ignorableError) Error() string {
+	return string(e)
+}