Add cancel token for reconcile

api/pkg/apis/v1alpha1/utils/apiclient.go

@@ -50,6 +50,7 @@ type (
 	Dispatcher interface {
 		QueueJob(ctx context.Context, id string, namespace string, isDelete bool, isTarget bool, user string, password string) error
 		QueueDeploymentJob(ctx context.Context, namespace string, isDelete bool, deployment model.DeploymentSpec, user string, password string) error
+		CancelDeploymentJob(ctx context.Context, id string, jobId string, namespace string, user string, password string) error
 	}
 
 	ApiClient interface {
@@ -537,6 +538,28 @@ func (a *apiClient) QueueDeploymentJob(ctx context.Context, namespace string, is
 	return nil
 }
 
+func (a *apiClient) CancelDeploymentJob(ctx context.Context, id string, jobId string, namespace string, user string, password string) error {
+	// func (a *apiClient) CancelDeploymentJob(ctx context.Context, namespace string, deployment model.DeploymentSpec) error {
+	token, err := a.tokenProvider(ctx, a.baseUrl, a.client, user, password)
+	if err != nil {
+		return err
+	}
+
+	path := "solution/cancel"
+	query := url.Values{
+		"namespace": []string{namespace},
+		"instance":  []string{id},
+		"jobid":     []string{jobId},
+	}
+
+	log.DebugfCtx(ctx, "apiClient.CancelDeploymentJob: Deployment id: %s, namespace: %v", id, namespace)
+	_, err = a.callRestAPI(ctx, fmt.Sprintf("%s?%s", path, query.Encode()), "POST", nil, token)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 // Deprecated: Use QueueDeploymentJob instead
 func (a *apiClient) QueueJob(ctx context.Context, id string, namespace string, isDelete bool, isTarget bool, user string, password string) error {
 	token, err := a.tokenProvider(ctx, a.baseUrl, a.client, user, password)

api/pkg/apis/v1alpha1/vendors/job-vendor.go

@@ -97,11 +97,7 @@ func (e *JobVendor) Init(config vendors.VendorConfig, factories []managers.IMana
 		},
 	})
 
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return err
 }
 
 func (o *JobVendor) GetEndpoints() []v1alpha2.Endpoint {

api/pkg/apis/v1alpha1/vendors/solution-vendor.go

@@ -74,6 +74,12 @@ func (o *SolutionVendor) GetEndpoints() []v1alpha2.Endpoint {
 			Parameters: []string{"delete?"},
 			Handler:    o.onReconcile,
 		},
+		{
+			Methods: []string{fasthttp.MethodPost},
+			Route:   route + "/cancel",
+			Version: o.Version,
+			Handler: o.onCancel,
+		},
 		{
 			Methods: []string{fasthttp.MethodGet, fasthttp.MethodPost, fasthttp.MethodDelete},
 			Route:   route + "/queue",
@@ -82,6 +88,7 @@ func (o *SolutionVendor) GetEndpoints() []v1alpha2.Endpoint {
 		},
 	}
 }
+
 func (c *SolutionVendor) onQueue(request v1alpha2.COARequest) v1alpha2.COAResponse {
 	rContext, span := observability.StartSpan("Solution Vendor", request.Context, &map[string]string{
 		"method": "onQueue",
@@ -161,6 +168,10 @@ func (c *SolutionVendor) onQueue(request v1alpha2.COARequest) v1alpha2.COARespon
 				})
 			}
 			instance = deployment.Instance.ObjectMeta.Name
+
+			if delete == "true" {
+				c.SolutionManager.CancelPreviousJobs(ctx, namespace, instance, deployment.JobID)
+			}
 		}
 
 		if instance == "" {
@@ -258,7 +269,8 @@ func (c *SolutionVendor) onReconcile(request v1alpha2.COARequest) v1alpha2.COARe
 				targetName = v
 			}
 		}
-		summary, err := c.SolutionManager.Reconcile(ctx, deployment, delete == "true", namespace, targetName)
+
+		summary, err := c.SolutionManager.ReconcileWithCancelWrapper(ctx, deployment, delete == "true", namespace, targetName)
 		data, _ := json.Marshal(summary)
 		if err != nil {
 			sLog.ErrorfCtx(ctx, "V (Solution): onReconcile failed POST - reconcile %s", err.Error())
@@ -281,6 +293,26 @@ func (c *SolutionVendor) onReconcile(request v1alpha2.COARequest) v1alpha2.COARe
 	})
 }
 
+func (c *SolutionVendor) onCancel(request v1alpha2.COARequest) v1alpha2.COAResponse {
+	rContext, span := observability.StartSpan("Solution Vendor", request.Context, &map[string]string{
+		"method": "onCancel",
+	})
+	defer span.End()
+
+	namespace := request.Parameters["namespace"]
+	instance := request.Parameters["instance"]
+	jobId := request.Parameters["jobId"]
+
+	sLog.InfofCtx(rContext, "V (Solution): onCancel instance: %s job ID: %s", instance, jobId)
+	c.SolutionManager.CancelPreviousJobs(rContext, namespace, instance, jobId)
+
+	return observ_utils.CloseSpanWithCOAResponse(span, v1alpha2.COAResponse{
+		State:       v1alpha2.OK,
+		Body:        []byte("{\"result\":\"200 - OK\"}"),
+		ContentType: "application/json",
+	})
+}
+
 func (c *SolutionVendor) onApplyDeployment(request v1alpha2.COARequest) v1alpha2.COAResponse {
 	rContext, span := observability.StartSpan("Solution Vendor", request.Context, &map[string]string{
 		"method": "onApplyDeployment",

api/pkg/apis/v1alpha1/vendors/solution-vendor_test.go

@@ -189,7 +189,7 @@ func TestSolutionEndpoints(t *testing.T) {
 	vendor := createSolutionVendor()
 	vendor.Route = "solution"
 	endpoints := vendor.GetEndpoints()
-	assert.Equal(t, 3, len(endpoints))
+	assert.Equal(t, 4, len(endpoints))
 }
 
 func TestSolutionInfo(t *testing.T) {

coa/pkg/apis/v1alpha2/errors.go

@@ -122,6 +122,15 @@ func IsNotFound(err error) bool {
 	}
 	return coaE.State == NotFound
 }
+
+func IsCanceled(err error) bool {
+	coaE, ok := err.(COAError)
+	if !ok {
+		return false
+	}
+	return coaE.State == Canceled
+}
+
 func IsDelayed(err error) bool {
 	coaE, ok := err.(COAError)
 	if !ok {

coa/pkg/apis/v1alpha2/types.go

@@ -55,6 +55,7 @@ const (
 	ValidateFailed State = 8003
 	Updated        State = 8004
 	Deleted        State = 8005
+	Canceled       State = 8006
 	// Workflow status
 	Running        State = 9994
 	Paused         State = 9995
@@ -222,6 +223,8 @@ func (s State) String() string {
 		return "Updated"
 	case Deleted:
 		return "Deleted"
+	case Canceled:
+		return "Canceled"
 	case Running:
 		return "Running"
 	case Paused:

k8s/testing/mocks.go

@@ -307,6 +307,12 @@ func (c *MockApiClient) QueueDeploymentJob(ctx context.Context, namespace string
 	return args.Error(0)
 }
 
+// CancelDeploymentJob implements utils.ApiClient.
+func (c *MockApiClient) CancelDeploymentJob(ctx context.Context, id string, jobId string, namespace string, user string, password string) error {
+	args := c.Called(ctx, namespace, id, jobId, namespace)
+	return args.Error(0)
+}
+
 // QueueJob implements ApiClient.
 // Deprecated and not used.
 func (c *MockApiClient) QueueJob(ctx context.Context, id string, scope string, isDelete bool, isTarget bool, user string, password string) error {