Repository: knative/pkg
Pull request #2201 (2021-07-29 21:11:10)

Add a new mechanism for requeuing a key.
# Changes

This is modeled after some of the semantics available in controller-runtime's `Result` type, which allows reconcilers to indicate a desire to reprocess a key either immediately or after a delay.

I worked around our lack of this when I reworked Tekton's timeout handling logic by exposing a "snooze" function to the Reconciler that wrapped `EnqueueAfter`, but this feels like a cleaner API for folks to use in general, and is consistent with our `NewPermanentError` and `NewSkipKey` functions in terms of influencing the queuing behaviors in our reconcilers via wrapped errors.

You can see some discussion of this here: https://knative.slack.com/archives/CA4DNJ9A4/p1627524921161100

/kind enhancement

**Release Note**

```release-note
Introduce a way for Knative-style controllers to requeue keys by returning a wrapped error from Reconcile[Kind].  See NewRequeue{Immediately,After} for more details.
```

**Docs**

Shouldn't have user-facing docs impact.

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/2201#" title="Author self-approved">mattmoor</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[controller/OWNERS](https://github.com/knative/pkg/blob/main/controller/OWNERS)~~ [mattmoor]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
vaikas: UT failed with:
```
--- FAIL: TestAdmissionValidResponseForResource (0.21s)
    logger.go:130: 2021-07-29T21:15:09.052Z	ERROR	webhook/webhook.go:196	failed to fetch secret	{"error": "secret \"webhook-certs\" not found"}
    admission_integration_test.go:139: Failed to get response Get "https://0.0.0.0:43137/bazinga": remote error: tls: unrecognized name
    admission_integration_test.go:176: Admit was called before informers had synced.
    logger.go:130: 2021-07-29T21:15:09.052Z	INFO	webhook/webhook.go:235	Starting to fail readiness probes...
    ```

mattmoor: I haven't touched that and can't rerun it because I'm not an editor.
vaikas: Yeah, I'm rerunning them :)
/lgtm
/approve
/hold

Just adding a hold so the other folks from the thread have a chance to take a look. If they don't have concerns, I'll remove the hold tmw morning.
@markusthoemmes  

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative/pkg/pull/2201#" title="Author self-approved">mattmoor</a>*, *<a href="https://github.com/knative/pkg/pull/2201#issuecomment-889489115" title="Approved">vaikas</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative%2Fpkg).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[controller/OWNERS](https://github.com/knative/pkg/blob/main/controller/OWNERS)~~ [mattmoor,vaikas]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative/pkg/pull/2201?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative) Report
> Merging [#2201](https://codecov.io/gh/knative/pkg/pull/2201?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative) (083056d) into [main](https://codecov.io/gh/knative/pkg/commit/889b5670e173d58d2dd62916956973e9f3b30ce7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative) (889b567) will **increase** coverage by `0.05%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative/pkg/pull/2201/graphs/tree.svg?width=650&height=150&src=pr&token=rlqpDnfxbO&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative)](https://codecov.io/gh/knative/pkg/pull/2201?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative)

```diff
@@            Coverage Diff             @@
##             main    #2201      +/-   ##
==========================================
+ Coverage   65.98%   66.04%   +0.05%     
==========================================
  Files         220      220              
  Lines        9207     9220      +13     
==========================================
+ Hits         6075     6089      +14     
+ Misses       2861     2860       -1     
  Partials      271      271              
```


| [Impacted Files](https://codecov.io/gh/knative/pkg/pull/2201?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative) | Coverage Δ | |
|---|---|---|
| [controller/controller.go](https://codecov.io/gh/knative/pkg/pull/2201/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative#diff-Y29udHJvbGxlci9jb250cm9sbGVyLmdv) | `87.59% <100.00%> (+1.03%)` | :arrow_up: |
| [test/gcs/mock/mock.go](https://codecov.io/gh/knative/pkg/pull/2201/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative#diff-dGVzdC9nY3MvbW9jay9tb2NrLmdv) | `91.39% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative/pkg/pull/2201?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative/pkg/pull/2201?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative). Last update [889b567...083056d](https://codecov.io/gh/knative/pkg/pull/2201?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative).

vaikas: /unhold


Patch:
diff --git a/controller/controller.go b/controller/controller.go
index 593d00a47..7ac98b2e0 100644
--- a/controller/controller.go
+++ b/controller/controller.go
@@ -555,6 +555,12 @@ func (c *Impl) handleErr(err error, key types.NamespacedName, startTime time.Tim
 		c.workQueue.Forget(key)
 		return
 	}
+	if ok, delay := IsRequeueKey(err); ok {
+		c.workQueue.AddAfter(key, delay)
+		c.logger.Debugf("Requeuing key %s (by request) after %v (depth: %d)", safeKey(key), delay, c.workQueue.Len())
+		return
+	}
+
 	c.logger.Errorw("Reconcile error", zap.Duration("duration", time.Since(startTime)), zap.Error(err))
 
 	// Re-queue the key if it's a transient error.
@@ -666,6 +672,49 @@ func (err permanentError) Unwrap() error {
 	return err.e
 }
 
+// NewRequeueImmediately returns a new instance of requeueKeyError.
+// Users can return this type of error to immediately requeue a key.
+func NewRequeueImmediately() error {
+	return requeueKeyError{}
+}
+
+// NewRequeueAfter returns a new instance of requeueKeyError.
+// Users can return this type of error to requeue a key after a delay.
+func NewRequeueAfter(dur time.Duration) error {
+	return requeueKeyError{duration: dur}
+}
+
+// requeueKeyError is an error that indicates the reconciler wants to reprocess
+// the key after a particular duration (possibly zero).
+// We should re-queue keys with the desired duration when this is returned by Reconcile.
+type requeueKeyError struct {
+	duration time.Duration
+}
+
+var _ error = requeueKeyError{}
+
+// Error implements the Error() interface of error.
+func (err requeueKeyError) Error() string {
+	return fmt.Sprintf("requeue after: %s", err.duration)
+}
+
+// IsRequeueKey returns true if the given error is a requeueKeyError.
+func IsRequeueKey(err error) (bool, time.Duration) {
+	rqe := requeueKeyError{}
+	if errors.As(err, &rqe) {
+		return true, rqe.duration
+	}
+	return false, 0
+}
+
+// Is implements the Is() interface of error. It returns whether the target
+// error can be treated as equivalent to a requeueKeyError.
+func (requeueKeyError) Is(target error) bool {
+	//nolint: errorlint // This check is actually fine.
+	_, ok := target.(requeueKeyError)
+	return ok
+}
+
 // Informer is the group of methods that a type must implement to be passed to
 // StartInformers.
 type Informer interface {


Test patch:
diff --git a/controller/controller_test.go b/controller/controller_test.go
index 222d7821f..7937b16e9 100644
--- a/controller/controller_test.go
+++ b/controller/controller_test.go
@@ -1128,6 +1128,26 @@ func TestPermanentError(t *testing.T) {
 	}
 }
 
+func TestRequeueKey(t *testing.T) {
+	err := new(fakeError)
+	reqErr := NewRequeueImmediately()
+	if ok, _ := IsRequeueKey(reqErr); !ok {
+		t.Errorf("Expected type %T to be a requeueKeyError", reqErr)
+	}
+	if ok, _ := IsRequeueKey(err); ok {
+		t.Errorf("Expected type %T to not be a requeueKeyError", err)
+	}
+
+	want := 10 * time.Minute
+	reqErr = NewRequeueAfter(want)
+	wrapReqErr := fmt.Errorf("wrapped: %w", reqErr)
+	if ok, got := IsRequeueKey(wrapReqErr); !ok {
+		t.Error("Expected wrapped requeueKeyError to be equivalent to a requeueKeyError")
+	} else if want != got {
+		t.Errorf("IsRequeueKey() = (true, %v), wanted (true, %v)", got, want)
+	}
+}
+
 type errorReconciler struct{}
 
 func (er *errorReconciler) Reconcile(context.Context, string) error {
@@ -1232,6 +1252,70 @@ func TestStartAndShutdownWithPermanentErroringWork(t *testing.T) {
 	checkStats(t, reporter, 1, 0, 1, falseString)
 }
 
+type requeueAfterReconciler struct {
+	duration time.Duration
+}
+
+func (er *requeueAfterReconciler) Reconcile(context.Context, string) error {
+	return NewRequeueAfter(er.duration)
+}
+
+func TestStartAndShutdownWithRequeuingWork(t *testing.T) {
+	tests := []struct {
+		name     string
+		minimum  int
+		duration time.Duration
+	}{{
+		name:    "no duration",
+		minimum: 100,
+	}, {
+		name:     "small duration",
+		minimum:  2,
+		duration: 100 * time.Millisecond,
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			r := &requeueAfterReconciler{duration: test.duration}
+			reporter := &FakeStatsReporter{}
+			impl := NewImplWithStats(r, TestLogger(t), "Testing", reporter)
+
+			ctx, cancel := context.WithCancel(context.Background())
+			doneCh := make(chan struct{})
+			go func() {
+				defer close(doneCh)
+				StartAll(ctx, impl)
+			}()
+			t.Cleanup(func() {
+				cancel()
+				<-doneCh
+			})
+
+			impl.EnqueueKey(types.NamespacedName{Namespace: "foo", Name: "bar"})
+
+			select {
+			case <-time.After(20 * time.Millisecond):
+				// We don't expect completion before the context is cancelled.
+			case <-doneCh:
+				t.Error("StartAll finished early.")
+			}
+			cancel()
+
+			select {
+			case <-time.After(time.Second):
+				t.Error("Timed out waiting for controller to finish.")
+			case <-doneCh:
+				// We expect the work to complete.
+			}
+
+			// Check that the work was not requeued in RateLimiter.
+			if got, wantAtLeast := impl.WorkQueue().NumRequeues(types.NamespacedName{Namespace: "foo", Name: "bar"}), test.minimum; got >= wantAtLeast {
+				t.Errorf("Requeue count = %v, wanted at least %v", got, wantAtLeast)
+			}
+		})
+	}
+}
+
 func drainWorkQueue(wq workqueue.RateLimitingInterface) (hasQueue []types.NamespacedName) {
 	for {
 		key, shutdown := wq.Get()