Merge ce3c29e into cb4ac63

pkg/server/handler/v1alpha1/webhook.go

@@ -97,6 +97,7 @@ func HandleWebhook(ctx context.Context, tenant, eventType, integration string) (
 
 func createWorkflowRun(tenant string, wft v1alpha1.WorkflowTrigger, data *scm.EventData) error {
 	ns := wft.Namespace
+	cancelPrevious := false
 	var err error
 	var project string
 	if wft.Labels != nil {
@@ -132,6 +133,7 @@ func createWorkflowRun(tenant string, wft v1alpha1.WorkflowTrigger, data *scm.Ev
 		for _, branch := range st.Push.Branches {
 			if branch == trimmedBranch {
 				trigger = true
+				cancelPrevious = true
 				break
 			}
 		}
@@ -144,19 +146,22 @@ func createWorkflowRun(tenant string, wft v1alpha1.WorkflowTrigger, data *scm.Ev
 		// Always trigger if Branches are not specified
 		if len(st.PullRequest.Branches) == 0 {
 			trigger = true
+			cancelPrevious = true
 			break
 		}
 
 		for _, branch := range st.PullRequest.Branches {
 			if branch == data.Branch {
 				trigger = true
+				cancelPrevious = true
 				break
 			}
 		}
 	case scm.PullRequestCommentEventType:
 		for _, comment := range st.PullRequestComment.Comments {
 			if comment == data.Comment {
 				trigger = true
+				cancelPrevious = true
 				break
 			}
 		}
@@ -208,6 +213,8 @@ func createWorkflowRun(tenant string, wft v1alpha1.WorkflowTrigger, data *scm.Ev
 		Spec: wft.Spec.WorkflowRunSpec,
 	}
 
+	cycloneClient := handler.K8sClient.CycloneV1alpha1()
+
 	wfr.Annotations, err = setSCMEventData(wfr.Annotations, data)
 	if err != nil {
 		return err
@@ -236,7 +243,7 @@ func createWorkflowRun(tenant string, wft v1alpha1.WorkflowTrigger, data *scm.Ev
 	}
 
 	accelerator.NewAccelerator(tenant, project, wfr).Accelerate()
-	_, err = handler.K8sClient.CycloneV1alpha1().WorkflowRuns(ns).Create(context.TODO(), wfr, metav1.CreateOptions{})
+	_, err = cycloneClient.WorkflowRuns(ns).Create(context.TODO(), wfr, metav1.CreateOptions{})
 	if err != nil {
 		return cerr.ConvertK8sError(err)
 	}
@@ -248,5 +255,30 @@ func createWorkflowRun(tenant string, wft v1alpha1.WorkflowTrigger, data *scm.Ev
 	if err != nil {
 		log.Warningf("Init pull request status for %s error: %v", wfr.Name, err)
 	}
+
+	if !cancelPrevious {
+		return nil
+	}
+
+	ctx := context.TODO()
+	wfrs, err := cycloneClient.WorkflowRuns(ns).List(ctx, metav1.ListOptions{
+		LabelSelector: fmt.Sprintf("%s=%s,%s=%s",
+			meta.LabelProjectName, project,
+			meta.LabelWorkflowName, wfName),
+	})
+	if err == nil {
+		for _, item := range wfrs.Items {
+			if item.Annotations[meta.AnnotationWorkflowRunTrigger] != string(data.Type) {
+				continue
+			}
+			_, err := stopWorkflowRun(ctx, &item)
+			if err != nil {
+				log.Warningf("Stop previous build: %v. project=%s, workflow=%s, trigger=%s, workflowRun=%s", err,
+					project, wfName, data.Type, item.Name)
+			}
+		}
+	} else {
+		log.Warningf("Fail to list previous WorkflowRuns: %v", err)
+	}
 	return nil
 }

pkg/server/handler/v1alpha1/workflowrun.go

@@ -279,20 +279,26 @@ func StopWorkflowRun(ctx context.Context, project, workflow, workflowrun, tenant
 		return nil, cerr.ConvertK8sError(err)
 	}
 
+	wfr, err = stopWorkflowRun(ctx, wfr)
+	if err != nil {
+		log.Errorf("Stop WorkflowRun %s error %s", workflowrun, err)
+		return nil, cerr.ConvertK8sError(err)
+	}
+	return wfr, nil
+}
+
+func stopWorkflowRun(ctx context.Context, wfr *v1alpha1.WorkflowRun) (*v1alpha1.WorkflowRun, error) {
 	// If wfr already in terminated state, skip it
 	if util.IsWorkflowRunTerminated(wfr) {
 		return wfr, nil
 	}
-
 	data, err := handler.BuildWfrStatusPatch(v1alpha1.StatusCancelled)
 	if err != nil {
-		log.Errorf("Stop WorkflowRun %s error %s", workflowrun, err)
 		return nil, err
 	}
 
-	wfr, err = handler.K8sClient.CycloneV1alpha1().WorkflowRuns(common.TenantNamespace(tenant)).Patch(context.TODO(), workflowrun, k8s_types.JSONPatchType, data, metav1.PatchOptions{})
-
-	return wfr, cerr.ConvertK8sError(err)
+	wfr, err = handler.K8sClient.CycloneV1alpha1().WorkflowRuns(wfr.Namespace).Patch(ctx, wfr.Name, k8s_types.JSONPatchType, data, metav1.PatchOptions{})
+	return wfr, err
 }
 
 // PauseWorkflowRun updates the workflowrun overall status to Waiting.