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 {
@@ -533,6 +534,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,10 +97,6 @@ func (e *JobVendor) Init(config vendors.VendorConfig, factories []managers.IMana
 		},
 	})
 
-	if err != nil {
-		return err
-	}
-
 	return nil
 }
 

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

@@ -10,6 +10,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"time"
 
 	"github.com/eclipse-symphony/symphony/api/constants"
 	"github.com/eclipse-symphony/symphony/api/pkg/apis/v1alpha1/managers/solution"
@@ -30,6 +31,10 @@ type SolutionVendor struct {
 	SolutionManager *solution.SolutionManager
 }
 
+const (
+	defaultTimeout = 60 * time.Minute
+)
+
 func (o *SolutionVendor) GetInfo() vendors.VendorInfo {
 	return vendors.VendorInfo{
 		Version:  o.Vendor.Version,
@@ -51,6 +56,7 @@ func (e *SolutionVendor) Init(config vendors.VendorConfig, factories []managers.
 	if e.SolutionManager == nil {
 		return v1alpha2.NewCOAError(nil, "solution manager is not supplied", v1alpha2.MissingConfig)
 	}
+	e.SolutionManager.InitCancelMap()
 	return nil
 }
 
@@ -73,6 +79,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",
@@ -81,6 +93,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",
@@ -155,6 +168,16 @@ func (c *SolutionVendor) onQueue(request v1alpha2.COARequest) v1alpha2.COARespon
 				})
 			}
 			instance = deployment.Instance.ObjectMeta.Name
+
+			if delete == "true" {
+				// cancel the jobs in queue
+				sLog.InfofCtx(rContext, "V (Solution): onQueue, delete instance: %s, job id: %s", instance, deployment.JobID)
+				c.SolutionManager.CancelPreviousJobs(rContext, namespace, instance, deployment.JobID)
+			} else {
+				// track the job id for an ongoing job
+				sLog.InfofCtx(rContext, "V (Solution): onQueue, add tracking job id for instance: %s, job id: %s", instance, deployment.JobID)
+				c.SolutionManager.TrackJob(rContext, namespace, instance, deployment.JobID)
+			}
 		}
 
 		if instance == "" {
@@ -245,14 +268,29 @@ func (c *SolutionVendor) onReconcile(request v1alpha2.COARequest) v1alpha2.COARe
 				Body:  []byte(err.Error()),
 			})
 		}
-		delete := request.Parameters["delete"]
+		isRemove := request.Parameters["delete"]
 		targetName := ""
 		if request.Metadata != nil {
 			if v, ok := request.Metadata["active-target"]; ok {
 				targetName = v
 			}
 		}
-		summary, err := c.SolutionManager.Reconcile(ctx, deployment, delete == "true", namespace, targetName)
+
+		instance := deployment.Instance.ObjectMeta.Name
+		sLog.InfofCtx(ctx, "V (Solution): onReconcile create context with timeout, instance: %s, job id: %s, isRemove: %s", instance, deployment.JobID, isRemove)
+		cancelCtx, cancel := context.WithTimeout(ctx, defaultTimeout)
+		if isRemove != "true" {
+			c.SolutionManager.AddCancelFunc(ctx, namespace, instance, deployment.JobID, cancel)
+		}
+		defer func() {
+			log.InfofCtx(rContext, "V (Solution): onReconcile complete, namespace: %s, instance: %s, job id: %s, isRemove: %s", namespace, instance, deployment.JobID, isRemove)
+			cancel()
+			if isRemove != "true" {
+				c.SolutionManager.UntrackJob(rContext, namespace, instance, deployment.JobID)
+			}
+		}()
+
+		summary, err := c.SolutionManager.Reconcile(cancelCtx, deployment, isRemove == "true", namespace, targetName)
 		data, _ := json.Marshal(summary)
 		if err != nil {
 			sLog.ErrorfCtx(ctx, "V (Solution): onReconcile failed POST - reconcile %s", err.Error())
@@ -275,6 +313,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

@@ -172,7 +172,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

@@ -69,6 +69,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
@@ -209,6 +210,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 {