feat: add timeout for pausing pipeline. Fixes #992

[dpadhiar]

Explain what this PR does.

Fixes #992

Allows user to add timeout (grace period) for pausing a pipeline. Default is 30 seconds and can be configured in spec.Lifecycle.PauseGracePeriodSeconds.

[whynowy]

@dpadhiar - I tested the change, the annotation causes one extra reconciliation. Can you look into it?

[dpadhiar]

@dpadhiar - I tested the change, the annotation causes one extra reconciliation. Can you look into it?

With a log, I am seeing that using update causes the reconciliation to happen twice when resuming the pipeline. When pausing the pipeline, reconciliation occurs twice to change the Status from Running to Pausing to Paused which is expected.

If we use patch instead of update, reconciliation only occurs once when resuming.

if pl.GetAnnotations() == nil || pl.GetAnnotations()[dfv1.KeyPauseTimestamp] == "" {
		pl.SetAnnotations(map[string]string{dfv1.KeyPauseTimestamp: time.Now().Format(time.RFC3339)})
		body, err := json.Marshal(pl)
		if err != nil {
			return false, err
		}
		err = r.client.Patch(ctx, pl, client.RawPatch(types.MergePatchType, body))
		if err != nil && !apierrors.IsNotFound(err) {
			return false, err
		}
	}

[whynowy]

In this case, we should let it re-queue, right?

[whynowy]

All the rest looks good to me, except the one commented.

[whynowy]

Just a tiny issue here:
if err != nill && IsNotFound, return false, nil

Somehow the pipeline can not be found, and we skip it.

[dpadhiar]

if err != nil {
  if apierrors.IsNotFound(err) {
    return false, nil // skip pl if we can't find it
  } else {
      return true, err // otherwise requeue resume
  }
}

[whynowy]

.

[dpadhiar]

@whynowy Not sure why this jetstream e2e test continues to fail, otherwise should be done.