Merge pull request #115 from MartinBasti/fix-out-of-order-deployments

Prevent out-of-order deployments

controllers/pipeline/pipeline_adapter.go

@@ -65,7 +65,7 @@ func NewAdapter(pipelineRun *tektonv1beta1.PipelineRun, component *applicationap
 // EnsureSnapshotExists is an operation that will ensure that a pipeline Snapshot associated
 // to the PipelineRun being processed exists. Otherwise, it will create a new pipeline Snapshot.
 func (a *Adapter) EnsureSnapshotExists() (reconciler.OperationResult, error) {
-	if !tekton.IsBuildPipelineRun(a.pipelineRun) || !tekton.HasPipelineRunSucceeded(a.pipelineRun) {
+	if !tekton.IsBuildPipelineRun(a.pipelineRun) || !helpers.HasPipelineRunSucceeded(a.pipelineRun) {
 		return reconciler.ContinueProcessing()
 	}
 
@@ -74,6 +74,19 @@ func (a *Adapter) EnsureSnapshotExists() (reconciler.OperationResult, error) {
 		return reconciler.ContinueProcessing()
 	}
 
+	isLatest, err := a.isLatestSucceededPipelineRun()
+	if err != nil {
+		return reconciler.RequeueWithError(err)
+	}
+	if !isLatest {
+		// not the last started pipeline that succeeded for current snapshot
+		// this prevents deploying older pipeline run over new deployment
+		a.logger.Info("The pipelineRun", a.pipelineRun.Name,
+			"is not the latest succeded pipelineRun for component",
+			a.component.Name, "will not create a new Snapshot.")
+		return reconciler.ContinueProcessing()
+	}
+
 	expectedSnapshot, err := a.prepareSnapshotForPipelineRun(a.pipelineRun, a.component, a.application)
 	if err != nil {
 		return reconciler.RequeueWithError(err)
@@ -109,7 +122,7 @@ func (a *Adapter) EnsureSnapshotExists() (reconciler.OperationResult, error) {
 // EnsureSnapshotPassedAllTests is an operation that will ensure that a pipeline Snapshot
 // to the PipelineRun being processed passed all tests for all defined non-optional IntegrationTestScenarios.
 func (a *Adapter) EnsureSnapshotPassedAllTests() (reconciler.OperationResult, error) {
-	if !tekton.IsIntegrationPipelineRun(a.pipelineRun) || !tekton.HasPipelineRunSucceeded(a.pipelineRun) {
+	if !tekton.IsIntegrationPipelineRun(a.pipelineRun) || !helpers.HasPipelineRunSucceeded(a.pipelineRun) {
 		return reconciler.ContinueProcessing()
 	}
 
@@ -457,3 +470,29 @@ func (a *Adapter) createCompositeSnapshotsIfConflictExists(application *applicat
 
 	return nil, nil
 }
+
+// isLatestSucceededPipelineRun return true if pipelineRun is the latest succeded pipelineRun
+// for the component. Pipeline start timestamp is used for comparison because we care about
+// time when pipeline was created.
+func (a *Adapter) isLatestSucceededPipelineRun() (bool, error) {
+
+	pipelineStartTime := a.pipelineRun.CreationTimestamp.Time
+
+	pipelineRuns, err := helpers.GetSucceededBuildPipelineRunsForComponent(a.client, a.context, a.component)
+	if err != nil {
+		return false, err
+	}
+	for _, run := range *pipelineRuns {
+		if a.pipelineRun.Name == run.Name {
+			// it's the same pipeline
+			continue
+		}
+		timestamp := run.CreationTimestamp.Time
+		if pipelineStartTime.Before(timestamp) {
+			// pipeline is not the latest
+			// 1 second is minimal granularity, if both pipelines started at the same second, we cannot decide
+			return false, nil
+		}
+	}
+	return true, nil
+}

controllers/pipeline/pipeline_adapter_test.go

@@ -14,6 +14,7 @@ limitations under the License.
 package pipeline
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -34,9 +35,11 @@ import (
 
 	ctrl "sigs.k8s.io/controller-runtime"
 
+	"github.com/go-logr/logr"
 	applicationapiv1alpha1 "github.com/redhat-appstudio/application-api/api/v1alpha1"
 	"github.com/redhat-appstudio/integration-service/status"
 	tektonv1beta1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
+	"github.com/tonglil/buflogr"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -568,4 +571,131 @@ var _ = Describe("Pipeline Adapter", Ordered, func() {
 			return result.RequeueRequest && err != nil && err.Error() == "ReportStatusError"
 		}, time.Second*10).Should(BeTrue())
 	})
+
+	When("multiple succesfull build pipeline runs exists for the same component", func() {
+
+		var (
+			testpipelineRunBuild2 *tektonv1beta1.PipelineRun
+		)
+
+		BeforeAll(func() {
+			testpipelineRunBuild2 = &tektonv1beta1.PipelineRun{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "pipelinerun-build-sample-2",
+					Namespace: "default",
+					Labels: map[string]string{
+						"pipelines.appstudio.openshift.io/type": "build",
+						"pipelines.openshift.io/used-by":        "build-cloud",
+						"pipelines.openshift.io/runtime":        "nodejs",
+						"pipelines.openshift.io/strategy":       "s2i",
+						"appstudio.openshift.io/component":      "component-sample",
+						"pipelinesascode.tekton.dev/event-type": "pull_request",
+					},
+					Annotations: map[string]string{
+						"appstudio.redhat.com/updateComponentOnSuccess": "false",
+						"pipelinesascode.tekton.dev/on-target-branch":   "[main,master]",
+						"foo": "bar",
+					},
+				},
+				Spec: tektonv1beta1.PipelineRunSpec{
+					PipelineRef: &tektonv1beta1.PipelineRef{
+						Name:   "build-pipeline-pass",
+						Bundle: "quay.io/kpavic/test-bundle:build-pipeline-pass",
+					},
+					Params: []tektonv1beta1.Param{
+						{
+							Name: "output-image",
+							Value: tektonv1beta1.ArrayOrString{
+								Type:      "string",
+								StringVal: "quay.io/redhat-appstudio/sample-image",
+							},
+						},
+					},
+				},
+			}
+			Expect(k8sClient.Create(ctx, testpipelineRunBuild2)).Should(Succeed())
+
+			testpipelineRunBuild2.Status = tektonv1beta1.PipelineRunStatus{
+				PipelineRunStatusFields: tektonv1beta1.PipelineRunStatusFields{
+					TaskRuns: map[string]*tektonv1beta1.PipelineRunTaskRunStatus{
+						"index1": {
+							PipelineTaskName: "build-container",
+							Status: &tektonv1beta1.TaskRunStatus{
+								TaskRunStatusFields: tektonv1beta1.TaskRunStatusFields{
+									TaskRunResults: []tektonv1beta1.TaskRunResult{
+										{
+											Name:  "IMAGE_DIGEST",
+											Value: *tektonv1beta1.NewArrayOrString("image_digest_value"),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				Status: v1beta1.Status{
+					Conditions: v1beta1.Conditions{
+						apis.Condition{
+							Reason: "Completed",
+							Status: "True",
+							Type:   apis.ConditionSucceeded,
+						},
+					},
+				},
+			}
+			Expect(k8sClient.Status().Update(ctx, testpipelineRunBuild2)).Should(Succeed())
+			Eventually(func() error {
+				err := k8sClient.Get(ctx, types.NamespacedName{
+					Name:      testpipelineRunBuild2.Name,
+					Namespace: "default",
+				}, testpipelineRunBuild2)
+				if err != nil {
+					return err
+				}
+				if !helpers.HasPipelineRunSucceeded(testpipelineRunBuild2) {
+					return fmt.Errorf("Pipeline is not marked as succeeded yet")
+				}
+				return err
+			}, time.Second*10).ShouldNot(HaveOccurred())
+		})
+
+		AfterAll(func() {
+			err := k8sClient.Delete(ctx, testpipelineRunBuild2)
+			Expect(err == nil || k8serrors.IsNotFound(err)).To(BeTrue())
+		})
+
+		It("isLatestSucceededPipelineRun reports second pipeline as the latest pipeline", func() {
+			// make sure the seocnd pipeline started as second
+			testpipelineRunBuild2.CreationTimestamp.Time = testpipelineRunBuild2.CreationTimestamp.Add(2 * time.Hour)
+			adapter = NewAdapter(testpipelineRunBuild2, hasComp, hasApp, ctrl.Log, k8sClient, ctx)
+			isLatest, err := adapter.isLatestSucceededPipelineRun()
+			Expect(err).To(BeNil())
+			Expect(isLatest).To(BeTrue())
+		})
+
+		It("isLatestSucceededPipelineRun doesn't report first pipeline as the latest pipeline", func() {
+			// make sure the first pipeline started as first
+			testpipelineRunBuild.CreationTimestamp.Time = testpipelineRunBuild.CreationTimestamp.Add(-2 * time.Hour)
+			adapter = NewAdapter(testpipelineRunBuild, hasComp, hasApp, ctrl.Log, k8sClient, ctx)
+			isLatest, err := adapter.isLatestSucceededPipelineRun()
+			Expect(err).To(BeNil())
+			Expect(isLatest).To(BeFalse())
+		})
+
+		It("ensure that EnsureSnapshotExists doesn't create snapshot for previous pipeline run", func() {
+			var buf bytes.Buffer
+			var log logr.Logger = buflogr.NewWithBuffer(&buf)
+
+			// make sure the first pipeline started as first
+			testpipelineRunBuild.CreationTimestamp.Time = testpipelineRunBuild.CreationTimestamp.Add(-2 * time.Hour)
+			adapter = NewAdapter(testpipelineRunBuild, hasComp, hasApp, log, k8sClient, ctx)
+			Eventually(func() bool {
+				result, err := adapter.EnsureSnapshotExists()
+				return !result.CancelRequest && err == nil
+			}, time.Second*10).Should(BeTrue())
+
+			expectedLogEntry := "INFO The pipelineRun pipelinerun-build-sample is not the latest succeded pipelineRun for component component-sample will not create a new Snapshot."
+			Expect(buf.String()).Should(ContainSubstring(expectedLogEntry))
+		})
+	})
 })

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/redhat-appstudio/operator-goodies v0.0.0-20221130140446-010c05bd7471
 	github.com/redhat-appstudio/release-service v0.0.0-20221116195308-308d82e909fa
 	github.com/tektoncd/pipeline v0.41.0
+	github.com/tonglil/buflogr v1.0.1
 	golang.org/x/oauth2 v0.1.0
 	k8s.io/api v0.25.3
 	k8s.io/apimachinery v0.25.3

go.sum

@@ -362,6 +362,8 @@ github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKs
 github.com/stvp/go-udp-testing v0.0.0-20201019212854-469649b16807/go.mod h1:7jxmlfBCDBXRzr0eAQJ48XC1hBu1np4CS5+cHEYfwpc=
 github.com/tektoncd/pipeline v0.41.0 h1:FksQuX83ZRasZygQPNmaR6hKBh6gy822XxRuoKBqPUE=
 github.com/tektoncd/pipeline v0.41.0/go.mod h1:YY4+PGfdsd6Qxn3PZXmCpKeS3heK8pIIcnUt37vRJ2Q=
+github.com/tonglil/buflogr v1.0.1 h1:WXFZLKxLfqcVSmckwiMCF8jJwjIgmStJmg63YKRF1p0=
+github.com/tonglil/buflogr v1.0.1/go.mod h1:yYWwvSpn/3uAaqjf6mJg/XMiAciaR0QcRJH2gJGDxNE=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=

helpers/integration.go

@@ -208,6 +208,47 @@ func GetAllPipelineRunsForSnapshotAndScenario(adapterClient client.Client, ctx c
 	return &integrationPipelineRuns.Items, nil
 }
 
+// GetAllPipelineRunsForComponent returns all PipelineRun for the
+// associated component. In the case the List operation fails,
+// an error will be returned.
+func GetAllBuildPipelineRunsForComponent(adapterClient client.Client, ctx context.Context, component *applicationapiv1alpha1.Component) (*[]tektonv1beta1.PipelineRun, error) {
+	buildPipelineRuns := &tektonv1beta1.PipelineRunList{}
+	opts := []client.ListOption{
+		client.InNamespace(component.Namespace),
+		client.MatchingLabels{
+			"pipelines.appstudio.openshift.io/type": "build",
+			"appstudio.openshift.io/component":      component.Name,
+		},
+	}
+
+	err := adapterClient.List(ctx, buildPipelineRuns, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return &buildPipelineRuns.Items, nil
+}
+
+// GetSucceededPipelineRunsForComponent returns all  succeeded PipelineRun for the
+// associated component. In the case the List operation fails,
+// an error will be returned.
+func GetSucceededBuildPipelineRunsForComponent(adapterClient client.Client, ctx context.Context, component *applicationapiv1alpha1.Component) (*[]tektonv1beta1.PipelineRun, error) {
+	var succeededPipelineRuns []tektonv1beta1.PipelineRun
+
+	buildPipelineRuns, err := GetAllBuildPipelineRunsForComponent(adapterClient, ctx, component)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, pipelineRun := range *buildPipelineRuns {
+		pipelineRun := pipelineRun // G601
+		if HasPipelineRunSucceeded(&pipelineRun) {
+			succeededPipelineRuns = append(succeededPipelineRuns, pipelineRun)
+		}
+
+	}
+	return &succeededPipelineRuns, nil
+}
+
 // GetLatestPipelineRunForSnapshotAndScenario returns the latest Integration PipelineRun for the
 // associated Snapshot and IntegrationTestScenario. In the case the List operation fails,
 // an error will be returned.
@@ -263,3 +304,13 @@ func GetTaskRunsFromPipelineRun(logger logr.Logger, pipelineRun *tektonv1beta1.P
 	sort.Sort(SortTaskRunsByStartTime(taskRuns))
 	return taskRuns
 }
+
+// HasPipelineRunSucceeded returns a boolean indicating whether the PipelineRun succeeded or not.
+// If the object passed to this function is not a PipelineRun, the function will return false.
+func HasPipelineRunSucceeded(object client.Object) bool {
+	if pr, ok := object.(*tektonv1beta1.PipelineRun); ok {
+		return pr.Status.GetCondition(apis.ConditionSucceeded).IsTrue()
+	}
+
+	return false
+}