feat: wf pod restart

api/router/pubsub/WorkflowStatusUpdateHandler.go

@@ -78,11 +78,18 @@ func (impl *WorkflowStatusUpdateHandlerImpl) Subscribe() error {
 			return
 		}
 
+		err = impl.ciHandler.HandleReTriggerCI(wfStatus)
+		if err != nil {
+			impl.logger.Errorw("error in HandleReTriggerCI", "err", err)
+			//don't return as we have to update the workflow status
+		}
+
 		_, err = impl.ciHandler.UpdateWorkflow(wfStatus)
 		if err != nil {
 			impl.logger.Errorw("error on update workflow status", "err", err, "msg", string(msg.Data))
 			return
 		}
+
 	}
 	err := impl.pubsubClient.Subscribe(pubsub.WORKFLOW_STATUS_UPDATE_TOPIC, callback)
 
@@ -125,6 +132,13 @@ func (impl *WorkflowStatusUpdateHandlerImpl) SubscribeCD() error {
 			} else if wfrStatus == string(v1alpha1.NodeFailed) || wfrStatus == string(v1alpha1.NodeError) {
 				eventType = util.Fail
 			}
+
+			if wfrStatus == string(v1alpha1.NodeFailed) && wfStatus.Message == pipeline.POD_DELETED_MESSAGE {
+				err = impl.cdHandler.HandleCdStageReTrigger(wfr)
+				if err != nil {
+					impl.logger.Errorw("error in HandleCdStageReTrigger", "error", err)
+				}
+			}
 			if wfr.WorkflowType == bean.CD_WORKFLOW_TYPE_PRE {
 				event := impl.eventFactory.Build(eventType, &wfr.CdWorkflow.PipelineId, wfr.CdWorkflow.Pipeline.AppId, &wfr.CdWorkflow.Pipeline.EnvironmentId, util.CD)
 				impl.logger.Debugw("event pre stage", "event", event)

api/router/router.go

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

internal/sql/repository/pipelineConfig/CdWorfkflowRepository.go

@@ -19,6 +19,7 @@ package pipelineConfig
 
 import (
 	"context"
+	"fmt"
 	"github.com/argoproj/gitops-engine/pkg/health"
 	"github.com/devtron-labs/devtron/api/bean"
 	"github.com/devtron-labs/devtron/client/argocdServer/application"
@@ -49,6 +50,7 @@ type CdWorkflowRepository interface {
 	FindPreviousCdWfRunnerByStatus(pipelineId int, currentWFRunnerId int, status []string) ([]*CdWorkflowRunner, error)
 	FindConfigByPipelineId(pipelineId int) (*CdWorkflowConfig, error)
 	FindWorkflowRunnerById(wfrId int) (*CdWorkflowRunner, error)
+	FindRetriedWorkflowCountByReferenceId(wfrId int) (int, error)
 	FindLatestWfrByAppIdAndEnvironmentId(appId int, environmentId int) (*CdWorkflowRunner, error)
 	FindLatestCdWorkflowRunnerByEnvironmentIdAndRunnerType(appId int, environmentId int, runnerType bean.WorkflowType) (CdWorkflowRunner, error)
 
@@ -143,23 +145,24 @@ const WORKFLOW_EXECUTOR_TYPE_AWF = "AWF"
 const WORKFLOW_EXECUTOR_TYPE_SYSTEM = "SYSTEM"
 
 type CdWorkflowRunner struct {
-	tableName          struct{}             `sql:"cd_workflow_runner" pg:",discard_unknown_columns"`
-	Id                 int                  `sql:"id,pk"`
-	Name               string               `sql:"name"`
-	WorkflowType       bean.WorkflowType    `sql:"workflow_type"` //pre,post,deploy
-	ExecutorType       WorkflowExecutorType `sql:"executor_type"` //awf, system
-	Status             string               `sql:"status"`
-	PodStatus          string               `sql:"pod_status"`
-	Message            string               `sql:"message"`
-	StartedOn          time.Time            `sql:"started_on"`
-	FinishedOn         time.Time            `sql:"finished_on"`
-	Namespace          string               `sql:"namespace"`
-	LogLocation        string               `sql:"log_file_path"`
-	TriggeredBy        int32                `sql:"triggered_by"`
-	CdWorkflowId       int                  `sql:"cd_workflow_id"`
-	PodName            string               `sql:"pod_name"`
-	BlobStorageEnabled bool                 `sql:"blob_storage_enabled,notnull"`
-	CdWorkflow         *CdWorkflow
+	tableName             struct{}             `sql:"cd_workflow_runner" pg:",discard_unknown_columns"`
+	Id                    int                  `sql:"id,pk"`
+	Name                  string               `sql:"name"`
+	WorkflowType          bean.WorkflowType    `sql:"workflow_type"` //pre,post,deploy
+	ExecutorType          WorkflowExecutorType `sql:"executor_type"` //awf, system
+	Status                string               `sql:"status"`
+	PodStatus             string               `sql:"pod_status"`
+	Message               string               `sql:"message"`
+	StartedOn             time.Time            `sql:"started_on"`
+	FinishedOn            time.Time            `sql:"finished_on"`
+	Namespace             string               `sql:"namespace"`
+	LogLocation           string               `sql:"log_file_path"`
+	TriggeredBy           int32                `sql:"triggered_by"`
+	CdWorkflowId          int                  `sql:"cd_workflow_id"`
+	PodName               string               `sql:"pod_name"`
+	BlobStorageEnabled    bool                 `sql:"blob_storage_enabled,notnull"`
+	RefCdWorkflowRunnerId int                  `sql:"ref_cd_workflow_runner_id"`
+	CdWorkflow            *CdWorkflow
 	sql.AuditLog
 }
 
@@ -182,30 +185,31 @@ type CiPipelineMaterialResponse struct {
 }
 
 type CdWorkflowWithArtifact struct {
-	Id                 int                          `json:"id"`
-	CdWorkflowId       int                          `json:"cd_workflow_id"`
-	Name               string                       `json:"name"`
-	Status             string                       `json:"status"`
-	PodStatus          string                       `json:"pod_status"`
-	Message            string                       `json:"message"`
-	StartedOn          time.Time                    `json:"started_on"`
-	FinishedOn         time.Time                    `json:"finished_on"`
-	PipelineId         int                          `json:"pipeline_id"`
-	Namespace          string                       `json:"namespace"`
-	LogFilePath        string                       `json:"log_file_path"`
-	TriggeredBy        int32                        `json:"triggered_by"`
-	EmailId            string                       `json:"email_id"`
-	Image              string                       `json:"image"`
-	MaterialInfo       string                       `json:"material_info,omitempty"`
-	DataSource         string                       `json:"data_source,omitempty"`
-	CiArtifactId       int                          `json:"ci_artifact_id,omitempty"`
-	WorkflowType       string                       `json:"workflow_type,omitempty"`
-	ExecutorType       string                       `json:"executor_type,omitempty"`
-	BlobStorageEnabled bool                         `json:"blobStorageEnabled"`
-	GitTriggers        map[int]GitCommit            `json:"gitTriggers"`
-	CiMaterials        []CiPipelineMaterialResponse `json:"ciMaterials"`
-	ImageReleaseTags   []*repository2.ImageTag      `json:"imageReleaseTags"`
-	ImageComment       *repository2.ImageComment    `json:"imageComment"`
+	Id                    int                          `json:"id"`
+	CdWorkflowId          int                          `json:"cd_workflow_id"`
+	Name                  string                       `json:"name"`
+	Status                string                       `json:"status"`
+	PodStatus             string                       `json:"pod_status"`
+	Message               string                       `json:"message"`
+	StartedOn             time.Time                    `json:"started_on"`
+	FinishedOn            time.Time                    `json:"finished_on"`
+	PipelineId            int                          `json:"pipeline_id"`
+	Namespace             string                       `json:"namespace"`
+	LogFilePath           string                       `json:"log_file_path"`
+	TriggeredBy           int32                        `json:"triggered_by"`
+	EmailId               string                       `json:"email_id"`
+	Image                 string                       `json:"image"`
+	MaterialInfo          string                       `json:"material_info,omitempty"`
+	DataSource            string                       `json:"data_source,omitempty"`
+	CiArtifactId          int                          `json:"ci_artifact_id,omitempty"`
+	WorkflowType          string                       `json:"workflow_type,omitempty"`
+	ExecutorType          string                       `json:"executor_type,omitempty"`
+	BlobStorageEnabled    bool                         `json:"blobStorageEnabled"`
+	GitTriggers           map[int]GitCommit            `json:"gitTriggers"`
+	CiMaterials           []CiPipelineMaterialResponse `json:"ciMaterials"`
+	ImageReleaseTags      []*repository2.ImageTag      `json:"imageReleaseTags"`
+	ImageComment          *repository2.ImageComment    `json:"imageComment"`
+	RefCdWorkflowRunnerId int                          `json:"referenceCdWorkflowRunnerId"`
 }
 
 type TriggerWorkflowStatus struct {
@@ -486,6 +490,15 @@ func (impl *CdWorkflowRepositoryImpl) FindWorkflowRunnerById(wfrId int) (*CdWork
 	return wfr, err
 }
 
+func (impl *CdWorkflowRepositoryImpl) FindRetriedWorkflowCountByReferenceId(wfrId int) (int, error) {
+	retryCount := 0
+	query := fmt.Sprintf("select count(*) "+
+		"from cd_workflow_runner where ref_cd_workflow_runner_id = %v", wfrId)
+
+	_, err := impl.dbConnection.Query(&retryCount, query)
+	return retryCount, err
+}
+
 func (impl *CdWorkflowRepositoryImpl) FindByWorkflowIdAndRunnerType(ctx context.Context, wfId int, runnerType bean.WorkflowType) (CdWorkflowRunner, error) {
 	var wfr CdWorkflowRunner
 	_, span := otel.Tracer("orchestrator").Start(ctx, "cdWorkflowRepository.FindByWorkflowIdAndRunnerType")

internal/sql/repository/pipelineConfig/CiWorkflowRepository.go

@@ -18,6 +18,7 @@
 package pipelineConfig
 
 import (
+	"fmt"
 	"github.com/go-pg/pg"
 	"go.uber.org/zap"
 	"time"
@@ -33,6 +34,7 @@ type CiWorkflowRepository interface {
 	FindByStatusesIn(activeStatuses []string) ([]*CiWorkflow, error)
 	FindByPipelineId(pipelineId int, offset int, size int) ([]WorkflowWithArtifact, error)
 	FindById(id int) (*CiWorkflow, error)
+	FindRetriedWorkflowCountByReferenceId(id int) (int, error)
 	FindCiWorkflowGitTriggersById(id int) (workflow *CiWorkflow, err error)
 	FindByName(name string) (*CiWorkflow, error)
 
@@ -51,25 +53,26 @@ type CiWorkflowRepositoryImpl struct {
 }
 
 type CiWorkflow struct {
-	tableName          struct{}          `sql:"ci_workflow" pg:",discard_unknown_columns"`
-	Id                 int               `sql:"id,pk"`
-	Name               string            `sql:"name"`
-	Status             string            `sql:"status"`
-	PodStatus          string            `sql:"pod_status"`
-	Message            string            `sql:"message"`
-	StartedOn          time.Time         `sql:"started_on"`
-	FinishedOn         time.Time         `sql:"finished_on"`
-	CiPipelineId       int               `sql:"ci_pipeline_id"`
-	Namespace          string            `sql:"namespace"`
-	BlobStorageEnabled bool              `sql:"blob_storage_enabled,notnull"`
-	LogLocation        string            `sql:"log_file_path"`
-	GitTriggers        map[int]GitCommit `sql:"git_triggers"`
-	TriggeredBy        int32             `sql:"triggered_by"`
-	CiArtifactLocation string            `sql:"ci_artifact_location"`
-	PodName            string            `sql:"pod_name"`
-	CiBuildType        string            `sql:"ci_build_type"`
-	EnvironmentId      int               `sql:"environment_id"`
-	CiPipeline         *CiPipeline
+	tableName             struct{}          `sql:"ci_workflow" pg:",discard_unknown_columns"`
+	Id                    int               `sql:"id,pk"`
+	Name                  string            `sql:"name"`
+	Status                string            `sql:"status"`
+	PodStatus             string            `sql:"pod_status"`
+	Message               string            `sql:"message"`
+	StartedOn             time.Time         `sql:"started_on"`
+	FinishedOn            time.Time         `sql:"finished_on"`
+	CiPipelineId          int               `sql:"ci_pipeline_id"`
+	Namespace             string            `sql:"namespace"`
+	BlobStorageEnabled    bool              `sql:"blob_storage_enabled,notnull"`
+	LogLocation           string            `sql:"log_file_path"`
+	GitTriggers           map[int]GitCommit `sql:"git_triggers"`
+	TriggeredBy           int32             `sql:"triggered_by"`
+	CiArtifactLocation    string            `sql:"ci_artifact_location"`
+	PodName               string            `sql:"pod_name"`
+	CiBuildType           string            `sql:"ci_build_type"`
+	EnvironmentId         int               `sql:"environment_id"`
+	ReferenceCiWorkflowId int               `sql:"ref_ci_workflow_id"`
+	CiPipeline            *CiPipeline
 }
 
 type WorkflowWithArtifact struct {
@@ -95,6 +98,7 @@ type WorkflowWithArtifact struct {
 	IsArtifactUploaded bool              `json:"is_artifact_uploaded"`
 	EnvironmentId      int               `json:"environmentId"`
 	EnvironmentName    string            `json:"environmentName"`
+	RefCiWorkflowId    int               `json:"referenceCiWorkflowId"`
 }
 
 type GitCommit struct {
@@ -111,9 +115,9 @@ type GitCommit struct {
 }
 
 type WebhookData struct {
-	Id              int
-	EventActionType string
-	Data            map[string]string
+	Id              int               `json:"id"`
+	EventActionType string            `json:"eventActionType"`
+	Data            map[string]string `json:"data"`
 }
 
 type CiWorkflowConfig struct {
@@ -192,6 +196,15 @@ func (impl *CiWorkflowRepositoryImpl) FindById(id int) (*CiWorkflow, error) {
 	return workflow, err
 }
 
+func (impl *CiWorkflowRepositoryImpl) FindRetriedWorkflowCountByReferenceId(id int) (int, error) {
+	retryCount := 0
+	query := fmt.Sprintf("select count(*) "+
+		"from ci_workflow where ref_ci_workflow_id = %v", id)
+
+	_, err := impl.dbConnection.Query(&retryCount, query)
+	return retryCount, err
+}
+
 func (impl *CiWorkflowRepositoryImpl) FindCiWorkflowGitTriggersById(id int) (ciWorkflow *CiWorkflow, err error) {
 	workflow := &CiWorkflow{}
 	err = impl.dbConnection.Model(workflow).

pkg/bean/app.go

@@ -242,35 +242,16 @@ type GitCiTriggerRequest struct {
 	ExtraEnvironmentVariables map[string]string  `json:"extraEnvironmentVariables"` // extra env variables which will be used for CI
 }
 
-type GitCommit struct {
-	Commit                 string //git hash
-	Author                 string
-	Date                   time.Time
-	Message                string
-	Changes                []string
-	WebhookData            *WebhookData
-	GitRepoUrl             string
-	GitRepoName            string
-	CiConfigureSourceType  pipelineConfig.SourceType
-	CiConfigureSourceValue string
-}
-
-type WebhookData struct {
-	Id              int               `json:"id"`
-	EventActionType string            `json:"eventActionType"`
-	Data            map[string]string `json:"data"`
-}
-
 type SourceType string
 
 type CiPipelineMaterial struct {
-	Id            int       `json:"Id"`
-	GitMaterialId int       `json:"GitMaterialId"`
-	Type          string    `json:"Type"`
-	Value         string    `json:"Value"`
-	Active        bool      `json:"Active"`
-	GitCommit     GitCommit `json:"GitCommit"`
-	GitTag        string    `json:"GitTag"`
+	Id            int                      `json:"Id"`
+	GitMaterialId int                      `json:"GitMaterialId"`
+	Type          string                   `json:"Type"`
+	Value         string                   `json:"Value"`
+	Active        bool                     `json:"Active"`
+	GitCommit     pipelineConfig.GitCommit `json:"GitCommit"`
+	GitTag        string                   `json:"GitTag"`
 }
 
 type CiTriggerRequest struct {

pkg/bulkAction/BulkUpdateService.go

@@ -1448,7 +1448,7 @@ func (impl BulkUpdateServiceImpl) BulkBuildTrigger(request *BulkApplicationForEn
 			var ciMaterials []bean2.CiPipelineMaterial
 			ciMaterials = append(ciMaterials, bean2.CiPipelineMaterial{
 				Id:        materialId,
-				GitCommit: bean2.GitCommit{Commit: commitHash},
+				GitCommit: pipelineConfig.GitCommit{Commit: commitHash},
 			})
 			ciTriggerRequest := bean2.CiTriggerRequest{
 				PipelineId:         ciPipelineId,

pkg/git/GitWebhookService.go

@@ -51,7 +51,7 @@ func (impl *GitWebhookServiceImpl) HandleGitWebhook(gitWebhookRequest gitSensor.
 		Type:          string(gitWebhookRequest.Type),
 		Value:         gitWebhookRequest.Value,
 		Active:        gitWebhookRequest.Active,
-		GitCommit: bean.GitCommit{
+		GitCommit: pipelineConfig.GitCommit{
 			Commit:  gitWebhookRequest.GitCommit.Commit,
 			Author:  gitWebhookRequest.GitCommit.Author,
 			Date:    gitWebhookRequest.GitCommit.Date,
@@ -62,7 +62,7 @@ func (impl *GitWebhookServiceImpl) HandleGitWebhook(gitWebhookRequest gitSensor.
 
 	if string(gitWebhookRequest.Type) == string(pipelineConfig.SOURCE_TYPE_WEBHOOK) {
 		webhookData := gitWebhookRequest.GitCommit.WebhookData
-		ciPipelineMaterial.GitCommit.WebhookData = &bean.WebhookData{
+		ciPipelineMaterial.GitCommit.WebhookData = pipelineConfig.WebhookData{
 			Id:              webhookData.Id,
 			EventActionType: webhookData.EventActionType,
 			Data:            webhookData.Data,

pkg/pipeline/CdConfig.go

@@ -73,6 +73,7 @@ type CdConfig struct {
 	BaseLogLocationPath              string                              `env:"BASE_LOG_LOCATION_PATH" envDefault:"/home/devtron/"`
 	CdWorkflowExecutorType           pipelineConfig.WorkflowExecutorType `env:"CD_WORKFLOW_EXECUTOR_TYPE" envDefault:"AWF"`
 	InAppLoggingEnabled              bool                                `env:"IN_APP_LOGGING_ENABLED" envDefault:"false"`
+	MaxCdWorkflowRunnerRetries       int                                 `env:"MAX_CD_WORKFLOW_RUNNER_RETRIES" envDefault:"0"`
 }
 
 func GetCdConfig() (*CdConfig, error) {