chore: workflow refactoring

Wire.go

@@ -406,7 +406,7 @@ func InitializeApp() (*App, error) {
 		repository.NewNotificationSettingsRepositoryImpl,
 		wire.Bind(new(repository.NotificationSettingsRepository), new(*repository.NotificationSettingsRepositoryImpl)),
 		util.IntValidator,
-		pipeline.GetCiConfig,
+		pipeline.GetCiCdConfig,
 
 		pipeline.NewWorkflowServiceImpl,
 		wire.Bind(new(pipeline.WorkflowService), new(*pipeline.WorkflowServiceImpl)),
@@ -533,17 +533,13 @@ func InitializeApp() (*App, error) {
 		pipelineConfig.NewCdWorkflowRepositoryImpl,
 		wire.Bind(new(pipelineConfig.CdWorkflowRepository), new(*pipelineConfig.CdWorkflowRepositoryImpl)),
 
-		pipeline.NewCdWorkflowServiceImpl,
-		wire.Bind(new(pipeline.CdWorkflowService), new(*pipeline.CdWorkflowServiceImpl)),
-
 		pipeline.NewCdHandlerImpl,
 		wire.Bind(new(pipeline.CdHandler), new(*pipeline.CdHandlerImpl)),
 
 		pipeline.NewWorkflowDagExecutorImpl,
 		wire.Bind(new(pipeline.WorkflowDagExecutor), new(*pipeline.WorkflowDagExecutorImpl)),
 		appClone.NewAppCloneServiceImpl,
 		wire.Bind(new(appClone.AppCloneService), new(*appClone.AppCloneServiceImpl)),
-		pipeline.GetCdConfig,
 
 		router.NewDeploymentGroupRouterImpl,
 		wire.Bind(new(router.DeploymentGroupRouter), new(*router.DeploymentGroupRouterImpl)),
@@ -879,6 +875,8 @@ func InitializeApp() (*App, error) {
 		wire.Bind(new(appGroup2.AppGroupMappingRepository), new(*appGroup2.AppGroupMappingRepositoryImpl)),
 		pipeline.NewArgoWorkflowExecutorImpl,
 		wire.Bind(new(pipeline.ArgoWorkflowExecutor), new(*pipeline.ArgoWorkflowExecutorImpl)),
+		pipeline.NewSystemWorkflowExecutorImpl,
+		wire.Bind(new(pipeline.SystemWorkflowExecutor), new(*pipeline.SystemWorkflowExecutorImpl)),
 		repository5.NewManifestPushConfigRepository,
 		wire.Bind(new(repository5.ManifestPushConfigRepository), new(*repository5.ManifestPushConfigRepositoryImpl)),
 		app.NewGitOpsManifestPushServiceImpl,

api/restHandler/PubSubClientRestHandler.go

@@ -35,15 +35,15 @@ type PubSubClientRestHandler interface {
 type PubSubClientRestHandlerImpl struct {
 	pubsubClient *pubsub.PubSubClientServiceImpl
 	logger       *zap.SugaredLogger
-	cdConfig     *pipeline.CdConfig
+	cdConfig     *pipeline.CiCdConfig
 }
 
 type PublishRequest struct {
 	Topic   string          `json:"topic"`
 	Payload json.RawMessage `json:"payload"`
 }
 
-func NewPubSubClientRestHandlerImpl(pubsubClient *pubsub.PubSubClientServiceImpl, logger *zap.SugaredLogger, cdConfig *pipeline.CdConfig) *PubSubClientRestHandlerImpl {
+func NewPubSubClientRestHandlerImpl(pubsubClient *pubsub.PubSubClientServiceImpl, logger *zap.SugaredLogger, cdConfig *pipeline.CiCdConfig) *PubSubClientRestHandlerImpl {
 	return &PubSubClientRestHandlerImpl{
 		pubsubClient: pubsubClient,
 		logger:       logger,

api/router/pubsub/CiEventHandler.go

@@ -25,6 +25,7 @@ import (
 	"github.com/devtron-labs/devtron/internal/sql/repository"
 	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig"
 	"github.com/devtron-labs/devtron/pkg/pipeline"
+	bean2 "github.com/devtron-labs/devtron/pkg/pipeline/bean"
 	"github.com/devtron-labs/devtron/util"
 	"go.uber.org/zap"
 )
@@ -53,19 +54,19 @@ type CiEventHandlerImpl struct {
 }
 
 type CiCompleteEvent struct {
-	CiProjectDetails   []pipeline.CiProjectDetails `json:"ciProjectDetails"`
-	DockerImage        string                      `json:"dockerImage" validate:"required,image-validator"`
-	Digest             string                      `json:"digest"`
-	PipelineId         int                         `json:"pipelineId"`
-	WorkflowId         *int                        `json:"workflowId"`
-	TriggeredBy        int32                       `json:"triggeredBy"`
-	PipelineName       string                      `json:"pipelineName"`
-	DataSource         string                      `json:"dataSource"`
-	MaterialType       string                      `json:"materialType"`
-	Metrics            util.CIMetrics              `json:"metrics"`
-	AppName            string                      `json:"appName"`
-	IsArtifactUploaded bool                        `json:"isArtifactUploaded"`
-	FailureReason      string                      `json:"failureReason"`
+	CiProjectDetails   []bean2.CiProjectDetails `json:"ciProjectDetails"`
+	DockerImage        string                   `json:"dockerImage" validate:"required,image-validator"`
+	Digest             string                   `json:"digest"`
+	PipelineId         int                      `json:"pipelineId"`
+	WorkflowId         *int                     `json:"workflowId"`
+	TriggeredBy        int32                    `json:"triggeredBy"`
+	PipelineName       string                   `json:"pipelineName"`
+	DataSource         string                   `json:"dataSource"`
+	MaterialType       string                   `json:"materialType"`
+	Metrics            util.CIMetrics           `json:"metrics"`
+	AppName            string                   `json:"appName"`
+	IsArtifactUploaded bool                     `json:"isArtifactUploaded"`
+	FailureReason      string                   `json:"failureReason"`
 }
 
 func NewCiEventHandlerImpl(logger *zap.SugaredLogger, pubsubClient *pubsub.PubSubClientServiceImpl, webhookService pipeline.WebhookService, ciEventConfig *CiEventConfig) *CiEventHandlerImpl {

api/router/router.go

@@ -222,7 +222,6 @@ func NewMuxRouter(logger *zap.SugaredLogger, HelmRouter PipelineTriggerRouter, P
 		appGroupingRouter:                  appGroupingRouter,
 		rbacRoleRouter:                     rbacRoleRouter,
 		scopedVariableRouter:               scopedVariableRouter,
-
 	}
 	return r
 }

pkg/pipeline/ArgoWorkflowExecutor.go

@@ -3,30 +3,43 @@ package pipeline
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo-workflows/v3/pkg/client/clientset/versioned"
 	v1alpha12 "github.com/argoproj/argo-workflows/v3/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
+	"github.com/argoproj/argo-workflows/v3/workflow/util"
 	bean2 "github.com/devtron-labs/devtron/api/bean"
 	"github.com/devtron-labs/devtron/pkg/pipeline/bean"
 	"go.uber.org/zap"
 	v12 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/rest"
 	"net/url"
 )
 
 const (
-	STEP_NAME_REGEX     = "create-env-%s-gb-%d"
-	TEMPLATE_NAME_REGEX = "%s-gb-%d"
-	WORKFLOW_MINIO_CRED = "workflow-minio-cred"
-	CRED_ACCESS_KEY     = "accessKey"
-	CRED_SECRET_KEY     = "secretKey"
+	STEP_NAME_REGEX              = "create-env-%s-gb-%d"
+	TEMPLATE_NAME_REGEX          = "%s-gb-%d"
+	WORKFLOW_MINIO_CRED          = "workflow-minio-cred"
+	CRED_ACCESS_KEY              = "accessKey"
+	CRED_SECRET_KEY              = "secretKey"
+	S3_ENDPOINT_URL              = "s3.amazonaws.com"
+	DEVTRON_WORKFLOW_LABEL_KEY   = "devtron.ai/workflow-purpose"
+	DEVTRON_WORKFLOW_LABEL_VALUE = "cd"
+	WORKFLOW_GENERATE_NAME_REGEX = "%s-"
+	RESOURCE_CREATE_ACTION       = "create"
 )
 
+var ACCESS_KEY_SELECTOR = &v12.SecretKeySelector{Key: CRED_ACCESS_KEY, LocalObjectReference: v12.LocalObjectReference{Name: WORKFLOW_MINIO_CRED}}
+var SECRET_KEY_SELECTOR = &v12.SecretKeySelector{Key: CRED_SECRET_KEY, LocalObjectReference: v12.LocalObjectReference{Name: WORKFLOW_MINIO_CRED}}
+
 type WorkflowExecutor interface {
-	ExecuteWorkflow(workflowTemplate bean.WorkflowTemplate) error
+	ExecuteWorkflow(workflowTemplate bean.WorkflowTemplate) (*unstructured.UnstructuredList, error)
+	TerminateWorkflow(workflowName string, namespace string, clusterConfig *rest.Config) error
 }
 
 type ArgoWorkflowExecutor interface {
@@ -41,37 +54,51 @@ func NewArgoWorkflowExecutorImpl(logger *zap.SugaredLogger) *ArgoWorkflowExecuto
 	return &ArgoWorkflowExecutorImpl{logger: logger}
 }
 
-func (impl *ArgoWorkflowExecutorImpl) ExecuteWorkflow(workflowTemplate bean.WorkflowTemplate) error {
+func (impl *ArgoWorkflowExecutorImpl) TerminateWorkflow(workflowName string, namespace string, clusterConfig *rest.Config) error {
+	impl.logger.Debugw("terminating wf", "name", workflowName)
+	wfClient, err := impl.getClientInstance(namespace, clusterConfig)
+	if err != nil {
+		impl.logger.Errorw("cannot build wf client", "wfName", workflowName, "err", err)
+		return err
+	}
+	_, err = wfClient.Get(context.Background(), workflowName, v1.GetOptions{})
+	if err != nil {
+		impl.logger.Errorw("cannot find workflow", "name", workflowName, "err", err)
+		return errors.New("cannot find workflow " + workflowName)
+	}
+	err = util.TerminateWorkflow(context.Background(), wfClient, workflowName)
+	return err
+}
+
+func (impl *ArgoWorkflowExecutorImpl) ExecuteWorkflow(workflowTemplate bean.WorkflowTemplate) (*unstructured.UnstructuredList, error) {
 
-	entryPoint := CD_WORKFLOW_NAME
+	entryPoint := workflowTemplate.WorkflowType
 	// get cm and cs argo step templates
-	templates, err := impl.getArgoTemplates(workflowTemplate.ConfigMaps, workflowTemplate.Secrets)
+	templates, err := impl.getArgoTemplates(workflowTemplate.ConfigMaps, workflowTemplate.Secrets, workflowTemplate.WorkflowType == bean.CI_WORKFLOW_NAME)
 	if err != nil {
 		impl.logger.Errorw("error occurred while fetching argo templates and steps", "err", err)
-		return err
+		return nil, err
 	}
 	if len(templates) > 0 {
-		entryPoint = CD_WORKFLOW_WITH_STAGES
+		entryPoint = workflowTemplate.GetEntrypoint()
 	}
 
 	wfContainer := workflowTemplate.Containers[0]
-	cdTemplate := v1alpha1.Template{
-		Name:      CD_WORKFLOW_NAME,
+	ciCdTemplate := v1alpha1.Template{
+		Name:      workflowTemplate.WorkflowType,
 		Container: &wfContainer,
 		ActiveDeadlineSeconds: &intstr.IntOrString{
 			IntVal: int32(*workflowTemplate.ActiveDeadlineSeconds),
 		},
 	}
-	impl.updateBlobStorageConfig(workflowTemplate, &cdTemplate)
-	templates = append(templates, cdTemplate)
+	impl.updateBlobStorageConfig(workflowTemplate, &ciCdTemplate)
+	templates = append(templates, ciCdTemplate)
+
+	objectMeta := workflowTemplate.CreateObjectMetadata()
 
 	var (
-		cdWorkflow = v1alpha1.Workflow{
-			ObjectMeta: v1.ObjectMeta{
-				GenerateName: workflowTemplate.WorkflowNamePrefix + "-",
-				Annotations:  map[string]string{"workflows.argoproj.io/controller-instanceid": workflowTemplate.WfControllerInstanceID},
-				Labels:       map[string]string{"devtron.ai/workflow-purpose": "cd"},
-			},
+		ciCdWorkflow = v1alpha1.Workflow{
+			ObjectMeta: *objectMeta,
 			Spec: v1alpha1.WorkflowSpec{
 				ServiceAccountName: workflowTemplate.ServiceAccountName,
 				NodeSelector:       workflowTemplate.NodeSelector,
@@ -86,26 +113,36 @@ func (impl *ArgoWorkflowExecutorImpl) ExecuteWorkflow(workflowTemplate bean.Work
 		}
 	)
 
-	wfTemplate, err := json.Marshal(cdWorkflow)
+	wfTemplate, err := json.Marshal(ciCdWorkflow)
 	if err != nil {
 		impl.logger.Errorw("error occurred while marshalling json", "err", err)
-		return err
+		return nil, err
 	}
 	impl.logger.Debugw("workflow request to submit", "wf", string(wfTemplate))
 
 	wfClient, err := impl.getClientInstance(workflowTemplate.Namespace, workflowTemplate.ClusterConfig)
 	if err != nil {
 		impl.logger.Errorw("cannot build wf client", "err", err)
-		return err
+		return nil, err
 	}
 
-	createdWf, err := wfClient.Create(context.Background(), &cdWorkflow, v1.CreateOptions{})
+	createdWf, err := wfClient.Create(context.Background(), &ciCdWorkflow, v1.CreateOptions{})
 	if err != nil {
 		impl.logger.Errorw("error in wf trigger", "err", err)
-		return err
+		return nil, err
 	}
 	impl.logger.Debugw("workflow submitted: ", "name", createdWf.Name)
-	return nil
+	return impl.convertToUnstructured(createdWf), nil
+}
+
+func (impl *ArgoWorkflowExecutorImpl) convertToUnstructured(cdWorkflow interface{}) *unstructured.UnstructuredList {
+	unstructedObjMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&cdWorkflow)
+	if err != nil {
+		return nil
+	}
+	unstructuredObj := unstructured.Unstructured{Object: unstructedObjMap}
+	unstructuredList := &unstructured.UnstructuredList{Items: []unstructured.Unstructured{unstructuredObj}}
+	return unstructuredList
 }
 
 func (impl *ArgoWorkflowExecutorImpl) updateBlobStorageConfig(workflowTemplate bean.WorkflowTemplate, cdTemplate *v1alpha1.Template) {
@@ -182,7 +219,7 @@ func (impl *ArgoWorkflowExecutorImpl) updateBlobStorageConfig(workflowTemplate b
 	}
 }
 
-func (impl *ArgoWorkflowExecutorImpl) getArgoTemplates(configMaps []bean2.ConfigSecretMap, secrets []bean2.ConfigSecretMap) ([]v1alpha1.Template, error) {
+func (impl *ArgoWorkflowExecutorImpl) getArgoTemplates(configMaps []bean2.ConfigSecretMap, secrets []bean2.ConfigSecretMap, isCi bool) ([]v1alpha1.Template, error) {
 	var templates []v1alpha1.Template
 	var steps []v1alpha1.ParallelSteps
 	cmIndex := 0
@@ -211,18 +248,27 @@ func (impl *ArgoWorkflowExecutorImpl) getArgoTemplates(configMaps []bean2.Config
 		templates = append(templates, argoTemplate)
 		csIndex++
 	}
+	if len(templates) <= 0 {
+		return templates, nil
+	}
+	stepName := bean.CD_WORKFLOW_NAME
+	templateName := bean.CD_WORKFLOW_WITH_STAGES
+	if isCi {
+		stepName = bean.CI_WORKFLOW_NAME
+		templateName = bean.CI_WORKFLOW_WITH_STAGES
+	}
 
 	steps = append(steps, v1alpha1.ParallelSteps{
 		Steps: []v1alpha1.WorkflowStep{
 			{
 				Name:     "run-wf",
-				Template: CD_WORKFLOW_NAME,
+				Template: stepName,
 			},
 		},
 	})
 
 	templates = append(templates, v1alpha1.Template{
-		Name:  CD_WORKFLOW_WITH_STAGES,
+		Name:  templateName,
 		Steps: steps,
 	})
 
@@ -271,7 +317,7 @@ func (impl *ArgoWorkflowExecutorImpl) createStepAndTemplate(isSecret bool, cmSec
 	argoTemplate := v1alpha1.Template{
 		Name: templateName,
 		Resource: &v1alpha1.ResourceTemplate{
-			Action:            "create",
+			Action:            RESOURCE_CREATE_ACTION,
 			SetOwnerReference: true,
 			Manifest:          string(cmSecretJson),
 		},