Respect timeout from pr by default

By default we will respect the timeout from the pr as defined by user or
the tekton controller.

We properly handle exiting on error if there is any timeouts.

Fixes #430

Signed-off-by: Chmouel Boudjnah <[email protected]>

Author: chmouel

config/302-pac-configmap.yaml

@@ -42,8 +42,9 @@ data:
   # Add extra IPS (ie: 127.0.0.1) or networks (127.0.0.0/16) separated by commas.
   bitbucket-cloud-additional-source-ip: ""
 
-  # The default time to wait for a pipelineRun
-  default-pipelinerun-timeout: "2h"
+  # Global setting for pipelinerun timeout regardless of how the user
+  # pipelinerun timeout.
+  # default-pipelinerun-timeout: "2h"
 
 kind: ConfigMap
 metadata:

docs/content/docs/install/settings.md

@@ -40,3 +40,15 @@ There is a few things you can configure through the configmap
 
   The base URL for the [tekton hub](https://github.com/tektoncd/hub/)
   API. default to the [public hub](https://hub.tekton.dev/): <https://api.hub.tekton.dev/v1>
+
+* `default-pipelinerun-timeout`
+
+  Let specify a global timeout across all pipeline. If you define the value
+  every PipelineRun running under Pipelines-as-Code
+
+  If it hasn't been set the timeout setting ordering will be :
+
+  * Picked up from the timeout setting [as
+    defined](https://tekton.dev/docs/pipelines/pipelineruns/#configuring-a-failure-timeout)
+    in the `PipelineRun` spec.timeout.
+  * If it hasn't defined it will use the default tekton controller timeout setting (default: 1 hour).

pkg/adapter/adapter.go

@@ -129,7 +129,10 @@ func (l listener) handleEvent() http.HandlerFunc {
 		localRequest := request.Clone(request.Context())
 
 		go func() {
-			s.processEvent(ctx, localRequest, payload)
+			err := s.processEvent(ctx, localRequest, payload)
+			if err != nil {
+				l.run.Clients.Log.Errorf("an error occurred: %v", err)
+			}
 		}()
 
 		l.writeResponse(response, http.StatusAccepted, "accepted")

pkg/adapter/sinker.go

@@ -20,12 +20,12 @@ type sinker struct {
 	event *info.Event
 }
 
-func (s *sinker) processEvent(ctx context.Context, request *http.Request, payload []byte) {
+func (s *sinker) processEvent(ctx context.Context, request *http.Request, payload []byte) error {
 	var err error
 	s.event, err = s.vcx.ParsePayload(ctx, s.run, request, string(payload))
 	if err != nil {
 		s.run.Clients.Log.Errorf("failed to parse event: %v", err)
-		return
+		return err
 	}
 	s.event.Request = &info.Request{
 		Header:  request.Header,
@@ -44,4 +44,5 @@ func (s *sinker) processEvent(ctx context.Context, request *http.Request, payloa
 			s.run.Clients.Log.Errorf("Cannot create status: %s %s", err, createStatusErr)
 		}
 	}
+	return err
 }

pkg/kubeinteraction/wait.go

@@ -15,7 +15,6 @@ import (
 
 const (
 	interval = 1 * time.Second
-	timeout  = 120 * time.Minute
 )
 
 type ConditionAccessorFn func(ca knativeapi.ConditionAccessor) (bool, error)
@@ -83,11 +82,11 @@ func PipelineRunSucceed(name string) ConditionAccessorFn {
 	return Succeed(name)
 }
 
-func PollImmediateWithContext(ctx context.Context, fn func() (bool, error)) error {
-	return wait.PollImmediate(interval, timeout, func() (bool, error) {
+func PollImmediateWithContext(ctx context.Context, pollTimeout time.Duration, fn func() (bool, error)) error {
+	return wait.PollImmediate(interval, pollTimeout, func() (bool, error) {
 		select {
 		case <-ctx.Done():
-			return true, ctx.Err()
+			return true, fmt.Errorf("polling timed out, pipelinerun has exceeded its timeout: %v", pollTimeout)
 		default:
 		}
 		return fn()
@@ -101,7 +100,8 @@ func PollImmediateWithContext(ctx context.Context, fn func() (bool, error)) erro
 func waitForPipelineRunState(ctx context.Context, tektonbeta1 tektonv1beta1client.TektonV1beta1Interface, pr *v1beta1.PipelineRun, polltimeout time.Duration, inState ConditionAccessorFn) error {
 	ctx, cancel := context.WithTimeout(ctx, polltimeout)
 	defer cancel()
-	return PollImmediateWithContext(ctx, func() (bool, error) {
+
+	return PollImmediateWithContext(ctx, polltimeout, func() (bool, error) {
 		r, err := tektonbeta1.PipelineRuns(pr.Namespace).Get(ctx, pr.Name, metav1.GetOptions{})
 		if err != nil {
 			return true, err

pkg/params/info/pac.go

@@ -17,7 +17,7 @@ type PacOpts struct {
 	TektonDashboardURL        string
 	HubURL                    string
 	RemoteTasks               bool
-	DefaultPipelineRunTimeout time.Duration
+	DefaultPipelineRunTimeout *time.Duration
 
 	// bitbucket cloud specific fields
 	BitbucketCloudCheckSourceIP      bool

pkg/params/run.go

@@ -33,6 +33,7 @@ func (r *Run) GetConfigFromConfigMap(ctx context.Context) error {
 	if ns == "" {
 		return fmt.Errorf("failed to find pipelines-as-code installation namespace")
 	}
+	// TODO: move this to kubeinteractions class so we can add unittests.
 	cfg, err := r.Clients.Kube.CoreV1().ConfigMaps(ns).Get(ctx, info.PACConfigmapName, v1.GetOptions{})
 	if err != nil {
 		return err
@@ -72,14 +73,10 @@ func (r *Run) GetConfigFromConfigMap(ctx context.Context) error {
 	if timeout, ok := cfg.Data["default-pipelinerun-timeout"]; ok {
 		parsedTimeout, err := time.ParseDuration(timeout)
 		if err != nil {
-			r.Clients.Log.Infof("failed to parse default-pipelinerun-timeout: %s, using %v as default timeout",
-				cfg.Data["default-pipelinerun-timeout"], info.DefaultPipelineRunTimeout)
-			r.Info.Pac.DefaultPipelineRunTimeout = info.DefaultPipelineRunTimeout
+			r.Clients.Log.Errorf("failed to parse default-pipelinerun-timeout: %s", cfg.Data["default-pipelinerun-timeout"])
 		} else {
-			r.Info.Pac.DefaultPipelineRunTimeout = parsedTimeout
+			r.Info.Pac.DefaultPipelineRunTimeout = &parsedTimeout
 		}
-	} else {
-		r.Info.Pac.DefaultPipelineRunTimeout = info.DefaultPipelineRunTimeout
 	}
 
 	if check, ok := cfg.Data["bitbucket-cloud-check-source-ip"]; ok {

pkg/pipelineascode/pipelineascode.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"strconv"
+	"time"
 
 	apipac "github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/formatting"
@@ -173,9 +174,18 @@ func Run(ctx context.Context, cs *params.Run, providerintf provider.Interface, k
 		return fmt.Errorf("cannot create a in_progress status on the provider platform: %w", err)
 	}
 
+	var duration time.Duration
+	if cs.Info.Pac.DefaultPipelineRunTimeout != nil {
+		duration = *cs.Info.Pac.DefaultPipelineRunTimeout
+	} else {
+		// Tekton Pipeline controller should always set this value.
+		duration = pr.Spec.Timeout.Duration + 1*time.Minute
+	}
 	cs.Clients.Log.Infof("Waiting for PipelineRun %s/%s to Succeed in a maximum time of %s minutes",
-		pr.Namespace, pr.Name, formatting.HumanDuration(cs.Info.Pac.DefaultPipelineRunTimeout))
-	if err := k8int.WaitForPipelineRunSucceed(ctx, cs.Clients.Tekton.TektonV1beta1(), pr, cs.Info.Pac.DefaultPipelineRunTimeout); err != nil {
+		pr.Namespace, pr.Name, formatting.HumanDuration(duration))
+	if err := k8int.WaitForPipelineRunSucceed(ctx, cs.Clients.Tekton.TektonV1beta1(), pr, duration); err != nil {
+		// if we have a timeout from the pipeline run, we would not know it. We would need to get the PR status to know.
+		// maybe something to improve in the future.
 		return fmt.Errorf("pipelinerun %s in namespace %s has a failed status: %w",
 			pipelineRun.GetGenerateName(), repo.GetNamespace(), err)
 	}

pkg/pipelineascode/pipelinesascode_github_test.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/google/go-github/v43/github"
 	"github.com/openshift-pipelines/pipelines-as-code/pkg/apis/pipelinesascode/v1alpha1"
@@ -332,6 +333,7 @@ func TestRun(t *testing.T) {
 			stdata, _ := testclient.SeedTestData(t, ctx, tdata)
 			tdc := testDynamic.Options{}
 			dc, _ := tdc.Client()
+			duration, _ := time.ParseDuration("0h0m5s")
 			cs := &params.Run{
 				Clients: clients.Clients{
 					PipelineAsCode: stdata.PipelineAsCode,
@@ -343,7 +345,8 @@ func TestRun(t *testing.T) {
 				},
 				Info: info.Info{
 					Pac: &info.PacOpts{
-						SecretAutoCreation: true,
+						SecretAutoCreation:        true,
+						DefaultPipelineRunTimeout: &duration,
 					},
 				},
 			}

test/pkg/wait/wait.go

@@ -23,7 +23,7 @@ type Opts struct {
 func UntilRepositoryUpdated(ctx context.Context, clients clients.Clients, opts Opts) error {
 	ctx, cancel := context.WithTimeout(ctx, opts.PollTimeout)
 	defer cancel()
-	return kubeinteraction.PollImmediateWithContext(ctx, func() (bool, error) {
+	return kubeinteraction.PollImmediateWithContext(ctx, opts.PollTimeout, func() (bool, error) {
 		clients.PipelineAsCode.PipelinesascodeV1alpha1().Repositories(opts.Namespace)
 		r, err := clients.PipelineAsCode.PipelinesascodeV1alpha1().Repositories(opts.Namespace).Get(ctx, opts.RepoName,
 			metav1.GetOptions{})