fix: Annotate PLR when when started status is reported (#2208)

* fix: Annotate PLR when started status is reported

The use of the state label (which is mutable) for deciding when to report
to the SCM that the PLR was started is flaky.

It was seen that the reconciler get events about PLRs with unexpected
value for the state label. For example, after the status is reported
to the SCM, and the label value is patched to "started", after serval
reconcile iterations the label had the "queued" value again. This can
happen because of unsafe patching done by controllers
(not just the PAC controllers) which reconciles PLRs.

Introduce a new annotation for indicating the the status was reported
to the SCM. By adding an annotation which is set once, we remove the
risk that its value will get overwritten by other controllers
(since maps are merged when patched, values are not getting removed
unless explicitly defined in the patch
- https://datatracker.ietf.org/doc/html/rfc7386#section-2)

In addition, at the start of each reconcile loop, ensure that we
operate on the latest version of the PLR and not using a stale value
from the cache.

Assisted-By: Cursor
Signed-off-by: Gal Ben Haim <[email protected]>

Author: gbenhaim

pkg/apis/pipelinesascode/keys/keys.go

@@ -23,43 +23,44 @@ import (
 )
 
 const (
-	ControllerInfo     = pipelinesascode.GroupName + "/controller-info"
-	Task               = pipelinesascode.GroupName + "/task"
-	Pipeline           = pipelinesascode.GroupName + "/pipeline"
-	URLOrg             = pipelinesascode.GroupName + "/url-org"
-	URLRepository      = pipelinesascode.GroupName + "/url-repository"
-	SHA                = pipelinesascode.GroupName + "/sha"
-	Sender             = pipelinesascode.GroupName + "/sender"
-	EventType          = pipelinesascode.GroupName + "/event-type"
-	Branch             = pipelinesascode.GroupName + "/branch"
-	SourceBranch       = pipelinesascode.GroupName + "/source-branch"
-	Repository         = pipelinesascode.GroupName + "/repository"
-	GitProvider        = pipelinesascode.GroupName + "/git-provider"
-	State              = pipelinesascode.GroupName + "/state"
-	ShaTitle           = pipelinesascode.GroupName + "/sha-title"
-	ShaURL             = pipelinesascode.GroupName + "/sha-url"
-	RepoURL            = pipelinesascode.GroupName + "/repo-url"
-	SourceRepoURL      = pipelinesascode.GroupName + "/source-repo-url"
-	PullRequest        = pipelinesascode.GroupName + "/pull-request"
-	InstallationID     = pipelinesascode.GroupName + "/installation-id"
-	GHEURL             = pipelinesascode.GroupName + "/ghe-url"
-	SourceProjectID    = pipelinesascode.GroupName + "/source-project-id"
-	TargetProjectID    = pipelinesascode.GroupName + "/target-project-id"
-	OriginalPRName     = pipelinesascode.GroupName + "/original-prname"
-	GitAuthSecret      = pipelinesascode.GroupName + "/git-auth-secret"
-	CheckRunID         = pipelinesascode.GroupName + "/check-run-id"
-	OnEvent            = pipelinesascode.GroupName + "/on-event"
-	OnComment          = pipelinesascode.GroupName + "/on-comment"
-	OnTargetBranch     = pipelinesascode.GroupName + "/on-target-branch"
-	OnPathChange       = pipelinesascode.GroupName + "/on-path-change"
-	OnLabel            = pipelinesascode.GroupName + "/on-label"
-	OnPathChangeIgnore = pipelinesascode.GroupName + "/on-path-change-ignore"
-	OnCelExpression    = pipelinesascode.GroupName + "/on-cel-expression"
-	TargetNamespace    = pipelinesascode.GroupName + "/target-namespace"
-	MaxKeepRuns        = pipelinesascode.GroupName + "/max-keep-runs"
-	CancelInProgress   = pipelinesascode.GroupName + "/cancel-in-progress"
-	LogURL             = pipelinesascode.GroupName + "/log-url"
-	ExecutionOrder     = pipelinesascode.GroupName + "/execution-order"
+	ControllerInfo         = pipelinesascode.GroupName + "/controller-info"
+	Task                   = pipelinesascode.GroupName + "/task"
+	Pipeline               = pipelinesascode.GroupName + "/pipeline"
+	URLOrg                 = pipelinesascode.GroupName + "/url-org"
+	URLRepository          = pipelinesascode.GroupName + "/url-repository"
+	SHA                    = pipelinesascode.GroupName + "/sha"
+	Sender                 = pipelinesascode.GroupName + "/sender"
+	EventType              = pipelinesascode.GroupName + "/event-type"
+	Branch                 = pipelinesascode.GroupName + "/branch"
+	SourceBranch           = pipelinesascode.GroupName + "/source-branch"
+	Repository             = pipelinesascode.GroupName + "/repository"
+	GitProvider            = pipelinesascode.GroupName + "/git-provider"
+	State                  = pipelinesascode.GroupName + "/state"
+	ShaTitle               = pipelinesascode.GroupName + "/sha-title"
+	ShaURL                 = pipelinesascode.GroupName + "/sha-url"
+	RepoURL                = pipelinesascode.GroupName + "/repo-url"
+	SourceRepoURL          = pipelinesascode.GroupName + "/source-repo-url"
+	PullRequest            = pipelinesascode.GroupName + "/pull-request"
+	InstallationID         = pipelinesascode.GroupName + "/installation-id"
+	GHEURL                 = pipelinesascode.GroupName + "/ghe-url"
+	SourceProjectID        = pipelinesascode.GroupName + "/source-project-id"
+	TargetProjectID        = pipelinesascode.GroupName + "/target-project-id"
+	OriginalPRName         = pipelinesascode.GroupName + "/original-prname"
+	GitAuthSecret          = pipelinesascode.GroupName + "/git-auth-secret"
+	CheckRunID             = pipelinesascode.GroupName + "/check-run-id"
+	OnEvent                = pipelinesascode.GroupName + "/on-event"
+	OnComment              = pipelinesascode.GroupName + "/on-comment"
+	OnTargetBranch         = pipelinesascode.GroupName + "/on-target-branch"
+	OnPathChange           = pipelinesascode.GroupName + "/on-path-change"
+	OnLabel                = pipelinesascode.GroupName + "/on-label"
+	OnPathChangeIgnore     = pipelinesascode.GroupName + "/on-path-change-ignore"
+	OnCelExpression        = pipelinesascode.GroupName + "/on-cel-expression"
+	TargetNamespace        = pipelinesascode.GroupName + "/target-namespace"
+	MaxKeepRuns            = pipelinesascode.GroupName + "/max-keep-runs"
+	CancelInProgress       = pipelinesascode.GroupName + "/cancel-in-progress"
+	LogURL                 = pipelinesascode.GroupName + "/log-url"
+	ExecutionOrder         = pipelinesascode.GroupName + "/execution-order"
+	SCMReportingPLRStarted = pipelinesascode.GroupName + "/scm-reporting-plr-started"
 	// PublicGithubAPIURL default is "https://api.github.com" but it can be overridden by X-GitHub-Enterprise-Host header.
 	PublicGithubAPIURL   = "https://api.github.com"
 	GithubApplicationID  = "github-application-id"

pkg/kubeinteraction/labels.go

@@ -55,7 +55,6 @@ func AddLabelsAndAnnotations(event *info.Event, pipelineRun *tektonv1.PipelineRu
 		keys.SourceBranch:  event.HeadBranch,
 		keys.Repository:    repo.GetName(),
 		keys.GitProvider:   providerConfig.Name,
-		keys.State:         StateStarted,
 		keys.ControllerInfo: fmt.Sprintf(`{"name":"%s","configmap":"%s","secret":"%s", "gRepo": "%s"}`,
 			paramsinfo.Controller.Name, paramsinfo.Controller.Configmap, paramsinfo.Controller.Secret, paramsinfo.Controller.GlobalRepository),
 	}

pkg/pipelineascode/pipelineascode.go

@@ -176,14 +176,10 @@ func (p *PacRun) startPR(ctx context.Context, match matcher.Match) (*tektonv1.Pi
 		p.logger.Errorf("Error adding labels/annotations to PipelineRun '%s' in namespace '%s': %v", match.PipelineRun.GetName(), match.Repo.GetNamespace(), err)
 	}
 
-	// if concurrency is defined then start the pipelineRun in pending state and
-	// state as queued
+	// if concurrency is defined then start the pipelineRun in pending state
 	if match.Repo.Spec.ConcurrencyLimit != nil && *match.Repo.Spec.ConcurrencyLimit != 0 {
 		// pending status
 		match.PipelineRun.Spec.Status = tektonv1.PipelineRunSpecStatusPending
-		// pac state as queued
-		match.PipelineRun.Labels[keys.State] = kubeinteraction.StateQueued
-		match.PipelineRun.Annotations[keys.State] = kubeinteraction.StateQueued
 	}
 
 	// Create the actual pipelineRun
@@ -240,23 +236,31 @@ func (p *PacRun) startPR(ctx context.Context, match matcher.Match) (*tektonv1.Pi
 		OriginalPipelineRunName: pr.GetAnnotations()[keys.OriginalPRName],
 	}
 
+	// Patch the pipelineRun with the appropriate annotations and labels.
+	// Set the state so the watcher will continue with reconciling the pipelineRun
+	// The watcher reconciles only pipelineRuns that has the state annotation.
 	patchAnnotations := map[string]string{}
 	patchLabels := map[string]string{}
 	whatPatching := ""
 	// if pipelineRun is in pending state then report status as queued
+	// The pipelineRun can be pending because of PAC's concurrency limit or because of an external mutatingwebhook
 	if pr.Spec.Status == tektonv1.PipelineRunSpecStatusPending {
 		status.Status = queuedStatus
 		if status.Text, err = mt.MakeTemplate(p.vcx.GetTemplate(provider.QueueingPipelineType)); err != nil {
 			return nil, fmt.Errorf("cannot create message template: %w", err)
 		}
-		// If the PipelineRun is in the "queued" state, add the appropriate label and annotation.
-		// These are later used by the watcher to determine whether the PipelineRun status
-		// should be reported back to the Git provider. We do add the `state` annotations and label when
-		// concurrency is enabled but this would happen when PipelineRun's status has been changed by
-		// the other controller and PaC is not aware of that change.
 		whatPatching = "annotations.state and labels.state"
 		patchAnnotations[keys.State] = kubeinteraction.StateQueued
 		patchLabels[keys.State] = kubeinteraction.StateQueued
+	} else {
+		// Mark that the start will be reported to the Git provider
+		patchAnnotations[keys.SCMReportingPLRStarted] = "true"
+		patchAnnotations[keys.State] = kubeinteraction.StateStarted
+		patchLabels[keys.State] = kubeinteraction.StateStarted
+		whatPatching = fmt.Sprintf(
+			"annotation.%s and annotations.state and labels.state",
+			keys.SCMReportingPLRStarted,
+		)
 	}
 
 	if err := p.vcx.CreateStatus(ctx, p.event, status); err != nil {
@@ -278,6 +282,20 @@ func (p *PacRun) startPR(ctx context.Context, match matcher.Match) (*tektonv1.Pi
 			// unneeded SIGSEGV's
 			return pr, fmt.Errorf("cannot patch pipelinerun %s: %w", pr.GetGenerateName(), err)
 		}
+		currentReason := ""
+		if len(pr.Status.GetConditions()) > 0 {
+			currentReason = pr.Status.GetConditions()[0].GetReason()
+		}
+
+		p.logger.Infof("PipelineRun %s/%s patched successfully - Spec.Status: %s, State annotation: '%s', SCMReportingPLRStarted annotation: '%s', Status reason: '%s', Git provider status: '%s', Patched: %s",
+			pr.GetNamespace(),
+			pr.GetName(),
+			pr.Spec.Status,
+			pr.GetAnnotations()[keys.State],
+			pr.GetAnnotations()[keys.SCMReportingPLRStarted],
+			currentReason,
+			status.Status,
+			whatPatching)
 	}
 
 	return pr, nil

pkg/pipelineascode/pipelineascode_test.go

@@ -718,10 +718,19 @@ func TestRun(t *testing.T) {
 						secretName, ok := pr.GetAnnotations()[keys.GitAuthSecret]
 						assert.Assert(t, ok, "Cannot find secret %s on annotations", secretName)
 					}
-					if tt.concurrencyLimit > 0 || pr.Spec.Status == pipelinev1.PipelineRunSpecStatusPending {
+					if pr.Spec.Status == pipelinev1.PipelineRunSpecStatusPending {
 						state, ok := pr.GetAnnotations()[keys.State]
 						assert.Assert(t, ok, "State hasn't been set on PR", state)
 						assert.Equal(t, state, kubeinteraction.StateQueued)
+
+						// When PipelineRun is queued, SCMReportingPLRStarted should not be set
+						_, scmStartedExists := pr.GetAnnotations()[keys.SCMReportingPLRStarted]
+						assert.Assert(t, !scmStartedExists, "SCMReportingPLRStarted should not be set for queued PipelineRuns")
+					} else {
+						// When PipelineRun is not queued, SCMReportingPLRStarted should be set to "true"
+						scmStarted, scmStartedExists := pr.GetAnnotations()[keys.SCMReportingPLRStarted]
+						assert.Assert(t, scmStartedExists, "SCMReportingPLRStarted should be set for non-queued PipelineRuns")
+						assert.Equal(t, scmStarted, "true", "SCMReportingPLRStarted should be 'true'")
 					}
 				}
 			}

pkg/reconciler/reconciler.go

@@ -54,6 +54,20 @@ var (
 func (r *Reconciler) ReconcileKind(ctx context.Context, pr *tektonv1.PipelineRun) pkgreconciler.Event {
 	ctx = info.StoreNS(ctx, system.Namespace())
 	logger := logging.FromContext(ctx).With("namespace", pr.GetNamespace())
+
+	logger.Debugf("reconciling pipelineRun %s/%s", pr.GetNamespace(), pr.GetName())
+
+	// make sure we have the latest pipelinerun to reconcile, since there is something updating at the same time
+	lpr, err := r.run.Clients.Tekton.TektonV1().PipelineRuns(pr.GetNamespace()).Get(ctx, pr.GetName(), metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("cannot get pipelineRun: %w", err)
+	}
+
+	if lpr.GetResourceVersion() != pr.GetResourceVersion() {
+		logger.Debugf("Skipping reconciliation, pipelineRun was updated (cached version %s vs fresh version %s)", pr.GetResourceVersion(), lpr.GetResourceVersion())
+		return nil
+	}
+
 	// if pipelineRun is in completed or failed state then return
 	state, exist := pr.GetAnnotations()[keys.State]
 	if exist && (state == kubeinteraction.StateCompleted || state == kubeinteraction.StateFailed) {
@@ -69,14 +83,20 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, pr *tektonv1.PipelineRun
 	// another controller outside PaC). To maintain consistency between the PipelineRun
 	// status and the Git provider status, we update both the PipelineRun resource and
 	// the corresponding status on the Git provider here.
-	if reason == string(tektonv1.PipelineRunReasonRunning) && state == kubeinteraction.StateQueued {
+	scmReportingPLRStarted, exist := pr.GetAnnotations()[keys.SCMReportingPLRStarted]
+	startReported := exist && scmReportingPLRStarted == "true"
+	logger.Debugf("pipelineRun %s/%s scmReportingPLRStarted=%v, exist=%v", pr.GetNamespace(), pr.GetName(), startReported, exist)
+
+	if reason == string(tektonv1.PipelineRunReasonRunning) && !startReported {
+		logger.Infof("pipelineRun %s/%s is running but not yet reported to provider, updating status", pr.GetNamespace(), pr.GetName())
 		repoName := pr.GetAnnotations()[keys.Repository]
 		repo, err := r.repoLister.Repositories(pr.Namespace).Get(repoName)
 		if err != nil {
 			return fmt.Errorf("failed to get repository CR: %w", err)
 		}
 		return r.updatePipelineRunToInProgress(ctx, logger, repo, pr)
 	}
+	logger.Debugf("pipelineRun %s/%s condition not met: reason='%s', startReported=%v", pr.GetNamespace(), pr.GetName(), reason, startReported)
 
 	// if its a GitHub App pipelineRun PR then process only if check run id is added otherwise wait
 	if _, ok := pr.Annotations[keys.InstallationID]; ok {
@@ -95,16 +115,6 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, pr *tektonv1.PipelineRun
 		return nil
 	}
 
-	// make sure we have the latest pipelinerun to reconcile, since there is something updating at the same time
-	lpr, err := r.run.Clients.Tekton.TektonV1().PipelineRuns(pr.GetNamespace()).Get(ctx, pr.GetName(), metav1.GetOptions{})
-	if err != nil {
-		return fmt.Errorf("cannot get pipelineRun: %w", err)
-	}
-
-	if lpr.GetResourceVersion() != pr.GetResourceVersion() {
-		return nil
-	}
-
 	// If we have a controllerInfo annotation, then we need to get the
 	// configmap configuration for it
 	//
@@ -343,16 +353,24 @@ func (r *Reconciler) updatePipelineRunToInProgress(ctx context.Context, logger *
 }
 
 func (r *Reconciler) updatePipelineRunState(ctx context.Context, logger *zap.SugaredLogger, pr *tektonv1.PipelineRun, state string) (*tektonv1.PipelineRun, error) {
+	currentState := pr.GetAnnotations()[keys.State]
+	logger.Infof("updating pipelineRun %v/%v state from %s to %s", pr.GetNamespace(), pr.GetName(), currentState, state)
+	annotations := map[string]string{
+		keys.State: state,
+	}
+	if state == kubeinteraction.StateStarted {
+		annotations[keys.SCMReportingPLRStarted] = "true"
+	}
+
 	mergePatch := map[string]any{
 		"metadata": map[string]any{
 			"labels": map[string]string{
 				keys.State: state,
 			},
-			"annotations": map[string]string{
-				keys.State: state,
-			},
+			"annotations": annotations,
 		},
 	}
+
 	// if state is started then remove pipelineRun pending status
 	if state == kubeinteraction.StateStarted {
 		mergePatch["spec"] = map[string]any{