Replace snooze with NewRequeueKey

pkg/reconciler/pipelinerun/controller.go

@@ -18,7 +18,6 @@ package pipelinerun
 
 import (
  "context"
- "time"
 
  "github.com/tektoncd/pipeline/pkg/apis/config"
  "github.com/tektoncd/pipeline/pkg/apis/pipeline"
@@ -35,12 +34,10 @@ import (
  resourceinformer "github.com/tektoncd/pipeline/pkg/client/resource/injection/informers/resource/v1alpha1/pipelineresource"
  cloudeventclient "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
  "github.com/tektoncd/pipeline/pkg/reconciler/volumeclaim"
- "k8s.io/apimachinery/pkg/types"
  "k8s.io/client-go/tools/cache"
  kubeclient "knative.dev/pkg/client/injection/kube/client"
  "knative.dev/pkg/configmap"
  "knative.dev/pkg/controller"
- "knative.dev/pkg/kmeta"
  "knative.dev/pkg/logging"
 )
 
@@ -88,13 +85,6 @@ func NewController(namespace string, images pipeline.Images) func(context.Contex
  }
  })
 
- c.snooze = func(acc kmeta.Accessor, amnt time.Duration) {
- impl.EnqueueKeyAfter(types.NamespacedName{
- Namespace: acc.GetNamespace(),
- Name: acc.GetName(),
- }, amnt)
- }
-
  logger.Info("Setting up event handlers")
  pipelineRunInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 

pkg/reconciler/pipelinerun/pipelinerun.go

@@ -131,8 +131,6 @@ type Reconciler struct {
  cloudEventClient cloudevent.CEClient
  metrics *Recorder
  pvcHandler volumeclaim.PvcHandler
-
- snooze func(kmeta.Accessor, time.Duration)
 }
 
 var (
@@ -221,23 +219,24 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, pr *v1beta1.PipelineRun)
  logger.Errorf("Error while syncing the pipelinerun status: %v", err.Error())
  return c.finishReconcileUpdateEmitEvents(ctx, pr, before, err)
  }
- defer func() {
- if pr.Status.StartTime == nil {
- return
- }
- // Compute the time since the task started.
- elapsed := time.Since(pr.Status.StartTime.Time)
- // Snooze this resource until the timeout has elapsed.
- c.snooze(pr, pr.GetTimeout(ctx)-elapsed)
- }()
 
  // Reconcile this copy of the pipelinerun and then write back any status or label
  // updates regardless of whether the reconciliation errored out.
  if err = c.reconcile(ctx, pr, getPipelineFunc); err != nil {
  logger.Errorf("Reconcile error: %v", err.Error())
  }
 
- return c.finishReconcileUpdateEmitEvents(ctx, pr, before, err)
+ if err = c.finishReconcileUpdateEmitEvents(ctx, pr, before, err); err != nil {
+ return err
+ }
+
+ if pr.Status.StartTime != nil {
+ // Compute the time since the task started.
+ elapsed := time.Since(pr.Status.StartTime.Time)
+ // Snooze this resource until the timeout has elapsed.
+ return controller.NewRequeueAfter(pr.GetTimeout(ctx) - elapsed)
+ }
+ return nil
 }
 
 func (c *Reconciler) finishReconcileUpdateEmitEvents(ctx context.Context, pr *v1beta1.PipelineRun, beforeCondition *apis.Condition, previousError error) error {

pkg/reconciler/pipelinerun/pipelinerun_test.go

@@ -2186,7 +2186,9 @@ func TestReconcileCancelledRunFinallyFailsTaskRunCancellation(t *testing.T) {
  failingReactorActivated = false
 
  err = c.Reconciler.Reconcile(testAssets.Ctx, "foo/test-pipeline-fails-to-cancel")
- if err != nil {
+ if err == nil {
+ // No error is ok
+ } else if ok, _ := controller.IsRequeueKey(err); !ok { // Requeue is also fine.
  t.Errorf("Expected to cancel TaskRun successfully!")
  }
 
@@ -6363,6 +6365,8 @@ func (prt PipelineRunTest) reconcileRun(namespace, pipelineRunName string, wantE
  if controller.IsPermanentError(reconcileError) != permanentError {
  prt.Test.Fatalf("Expected the error to be permanent: %v but got: %v", permanentError, controller.IsPermanentError(reconcileError))
  }
+ } else if ok, _ := controller.IsRequeueKey(reconcileError); ok {
+ // This is normal, it happens for timeouts when we otherwise successfully reconcile.
  } else if reconcileError != nil {
  prt.Test.Fatalf("Error reconciling: %s", reconcileError)
  }

pkg/reconciler/taskrun/controller.go

@@ -18,7 +18,6 @@ package taskrun
 
 import (
  "context"
- "time"
 
  "github.com/tektoncd/pipeline/pkg/apis/config"
  "github.com/tektoncd/pipeline/pkg/apis/pipeline"
@@ -32,13 +31,11 @@ import (
  "github.com/tektoncd/pipeline/pkg/pod"
  cloudeventclient "github.com/tektoncd/pipeline/pkg/reconciler/events/cloudevent"
  "github.com/tektoncd/pipeline/pkg/reconciler/volumeclaim"
- "k8s.io/apimachinery/pkg/types"
  "k8s.io/client-go/tools/cache"
  kubeclient "knative.dev/pkg/client/injection/kube/client"
  filteredpodinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/filtered"
  "knative.dev/pkg/configmap"
  "knative.dev/pkg/controller"
- "knative.dev/pkg/kmeta"
  "knative.dev/pkg/logging"
 )
 
@@ -86,13 +83,6 @@ func NewController(namespace string, images pipeline.Images) func(context.Contex
  }
  })
 
- c.snooze = func(acc kmeta.Accessor, amnt time.Duration) {
- impl.EnqueueKeyAfter(types.NamespacedName{
- Namespace: acc.GetNamespace(),
- Name: acc.GetName(),
- }, amnt)
- }
-
  logger.Info("Setting up event handlers")
  taskRunInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 

pkg/reconciler/taskrun/taskrun.go

@@ -72,8 +72,6 @@ type Reconciler struct {
  entrypointCache podconvert.EntrypointCache
  metrics *Recorder
  pvcHandler volumeclaim.PvcHandler
-
- snooze func(kmeta.Accessor, time.Duration)
 }
 
 // Check that our Reconciler implements taskrunreconciler.Interface
@@ -163,15 +161,6 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, tr *v1beta1.TaskRun) pkg
  err := c.failTaskRun(ctx, tr, v1beta1.TaskRunReasonTimedOut, message)
  return c.finishReconcileUpdateEmitEvents(ctx, tr, before, err)
  }
- defer func() {
- if tr.Status.StartTime == nil {
- return
- }
- // Compute the time since the task started.
- elapsed := time.Since(tr.Status.StartTime.Time)
- // Snooze this resource until the timeout has elapsed.
- c.snooze(tr, tr.GetTimeout(ctx)-elapsed)
- }()
 
  // prepare fetches all required resources, validates them together with the
  // taskrun, runs API convertions. Errors that come out of prepare are
@@ -194,7 +183,17 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, tr *v1beta1.TaskRun) pkg
  }
 
  // Emit events (only when ConditionSucceeded was changed)
- return c.finishReconcileUpdateEmitEvents(ctx, tr, before, err)
+ if err = c.finishReconcileUpdateEmitEvents(ctx, tr, before, err); err != nil {
+ return err
+ }
+
+ if tr.Status.StartTime != nil {
+ // Compute the time since the task started.
+ elapsed := time.Since(tr.Status.StartTime.Time)
+ // Snooze this resource until the timeout has elapsed.
+ return controller.NewRequeueAfter(tr.GetTimeout(ctx) - elapsed)
+ }
+ return nil
 }
 func (c *Reconciler) stopSidecars(ctx context.Context, tr *v1beta1.TaskRun) (*corev1.Pod, error) {
  logger := logging.FromContext(ctx)

pkg/reconciler/taskrun/taskrun_test.go

@@ -571,7 +571,9 @@ func TestReconcile_ExplicitDefaultSA(t *testing.T) {
  c := testAssets.Controller
  clients := testAssets.Clients
 
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tc.taskRun)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tc.taskRun)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("expected no error. Got error %v", err)
  }
  if len(clients.Kube.Actions()) == 0 {
@@ -764,7 +766,9 @@ func TestReconcile_FeatureFlags(t *testing.T) {
  }, metav1.CreateOptions{}); err != nil {
  t.Fatal(err)
  }
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tc.taskRun)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tc.taskRun)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("expected no error. Got error %v", err)
  }
  if len(clients.Kube.Actions()) == 0 {
@@ -849,7 +853,9 @@ func TestReconcile_CloudEvents(t *testing.T) {
  t.Fatal(err)
  }
 
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(taskRun)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(taskRun)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("expected no error. Got error %v", err)
  }
  if len(clients.Kube.Actions()) == 0 {
@@ -1709,7 +1715,9 @@ func TestReconcile(t *testing.T) {
  t.Fatal(err)
  }
 
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tc.taskRun)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tc.taskRun)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("expected no error. Got error %v", err)
  }
  if len(clients.Kube.Actions()) == 0 {
@@ -1779,7 +1787,9 @@ func TestReconcile_SetsStartTime(t *testing.T) {
  t.Fatal(err)
  }
 
- if err := testAssets.Controller.Reconciler.Reconcile(context.Background(), getRunName(taskRun)); err != nil {
+ if err := testAssets.Controller.Reconciler.Reconcile(context.Background(), getRunName(taskRun)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("expected no error reconciling valid TaskRun but got %v", err)
  }
 
@@ -1933,7 +1943,7 @@ func TestReconcileTaskRunWithPermanentError(t *testing.T) {
  // Such TaskRun enters Reconciler and from within the isDone block, marks the run success so that
  // reconciler does not keep trying to reconcile
  if reconcileErr != nil {
- t.Fatalf("Expected to see no error when reconciling TaskRun with Permanent Error but was not none")
+ t.Fatalf("Expected to see error when reconciling TaskRun with Permanent Error but was not none")
  }
 
  // Check actions
@@ -2036,7 +2046,9 @@ func TestReconcilePodUpdateStatus(t *testing.T) {
  c := testAssets.Controller
  clients := testAssets.Clients
 
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(taskRun)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(taskRun)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Fatalf("Unexpected error when Reconcile() : %v", err)
  }
  newTr, err := clients.Pipeline.TektonV1beta1().TaskRuns(taskRun.Namespace).Get(testAssets.Ctx, taskRun.Name, metav1.GetOptions{})
@@ -2072,7 +2084,9 @@ func TestReconcilePodUpdateStatus(t *testing.T) {
  // lister cache is update to reflect the result of the previous Reconcile.
  testAssets.Informers.TaskRun.Informer().GetIndexer().Add(newTr)
 
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(taskRun)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(taskRun)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Fatalf("Unexpected error when Reconcile(): %v", err)
  }
 
@@ -2514,13 +2528,10 @@ func TestHandlePodCreationError(t *testing.T) {
  taskLister: testAssets.Informers.Task.Lister(),
  clusterTaskLister: testAssets.Informers.ClusterTask.Lister(),
  resourceLister: testAssets.Informers.PipelineResource.Lister(),
- snooze: func(acc kmeta.Accessor, amnt time.Duration) {
- t.Error("Unexpected call to snooze.")
- },
- cloudEventClient: testAssets.Clients.CloudEvents,
- metrics: nil, // Not used
- entrypointCache: nil, // Not used
- pvcHandler: volumeclaim.NewPVCHandler(testAssets.Clients.Kube, testAssets.Logger),
+ cloudEventClient: testAssets.Clients.CloudEvents,
+ metrics: nil, // Not used
+ entrypointCache: nil, // Not used
+ pvcHandler: volumeclaim.NewPVCHandler(testAssets.Clients.Kube, testAssets.Logger),
  }
 
  testcases := []struct {
@@ -2729,7 +2740,9 @@ func TestReconcileCloudEvents(t *testing.T) {
  t.Fatal(err)
  }
 
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tc.taskRun)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tc.taskRun)); err == nil {
+ // No error is ok.
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("expected no error. Got error %v", err)
  }
 
@@ -2947,7 +2960,9 @@ func TestReconcileValidDefaultWorkspace(t *testing.T) {
  t.Fatal(err)
  }
 
- if err := testAssets.Controller.Reconciler.Reconcile(testAssets.Ctx, getRunName(taskRun)); err != nil {
+ if err := testAssets.Controller.Reconciler.Reconcile(testAssets.Ctx, getRunName(taskRun)); err == nil {
+ // No error is ok.
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("Expected no error reconciling valid TaskRun but got %v", err)
  }
 
@@ -3070,7 +3085,9 @@ func TestReconcileValidDefaultWorkspaceOmittedOptionalWorkspace(t *testing.T) {
  t.Fatal(err)
  }
 
- if err := testAssets.Controller.Reconciler.Reconcile(context.Background(), getRunName(taskRunOmittingWorkspace)); err != nil {
+ if err := testAssets.Controller.Reconciler.Reconcile(context.Background(), getRunName(taskRunOmittingWorkspace)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("Unexpected reconcile error for TaskRun %q: %v", taskRunOmittingWorkspace.Name, err)
  }
 
@@ -3299,7 +3316,9 @@ func TestReconcileWorkspaceWithVolumeClaimTemplate(t *testing.T) {
  t.Fatal(err)
  }
 
- if err := testAssets.Controller.Reconciler.Reconcile(context.Background(), getRunName(taskRun)); err != nil {
+ if err := testAssets.Controller.Reconciler.Reconcile(context.Background(), getRunName(taskRun)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("expected no error reconciling valid TaskRun but got %v", err)
  }
 
@@ -3615,13 +3634,10 @@ func TestFailTaskRun(t *testing.T) {
  taskLister: testAssets.Informers.Task.Lister(),
  clusterTaskLister: testAssets.Informers.ClusterTask.Lister(),
  resourceLister: testAssets.Informers.PipelineResource.Lister(),
- snooze: func(acc kmeta.Accessor, amnt time.Duration) {
- t.Error("Unexpected call to snooze.")
- },
- cloudEventClient: testAssets.Clients.CloudEvents,
- metrics: nil, // Not used
- entrypointCache: nil, // Not used
- pvcHandler: volumeclaim.NewPVCHandler(testAssets.Clients.Kube, testAssets.Logger),
+ cloudEventClient: testAssets.Clients.CloudEvents,
+ metrics: nil, // Not used
+ entrypointCache: nil, // Not used
+ pvcHandler: volumeclaim.NewPVCHandler(testAssets.Clients.Kube, testAssets.Logger),
  }
 
  err := c.failTaskRun(testAssets.Ctx, tc.taskRun, tc.reason, tc.message)
@@ -3762,7 +3778,9 @@ func TestPodAdoption(t *testing.T) {
  }
 
  // Reconcile the TaskRun. This creates a Pod.
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tr)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tr)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("Error reconciling TaskRun. Got error %v", err)
  }
 
@@ -3795,7 +3813,9 @@ func TestPodAdoption(t *testing.T) {
  }
 
  // Reconcile the TaskRun again.
- if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tr)); err != nil {
+ if err := c.Reconciler.Reconcile(testAssets.Ctx, getRunName(tr)); err == nil {
+ t.Error("Wanted a wrapped requeue error, but got nil.")
+ } else if ok, _ := controller.IsRequeueKey(err); !ok {
  t.Errorf("Error reconciling TaskRun again. Got error %v", err)
  }