fix: Pipeline status issues fixes

Wire.go

@@ -402,8 +402,8 @@ func InitializeApp() (*App, error) {
 		rbac.NewEnforcerUtilImpl,
 		wire.Bind(new(rbac.EnforcerUtil), new(*rbac.EnforcerUtilImpl)),
 
-		app.NewDeploymentFailureHandlerImpl,
-		wire.Bind(new(app.DeploymentFailureHandler), new(*app.DeploymentFailureHandlerImpl)),
+		app.NewDeploymentEventHandlerImpl,
+		wire.Bind(new(app.DeploymentEventHandler), new(*app.DeploymentEventHandlerImpl)),
 		chartConfig.NewPipelineConfigRepository,
 		wire.Bind(new(chartConfig.PipelineConfigRepository), new(*chartConfig.PipelineConfigRepositoryImpl)),
 

api/restHandler/AppListingRestHandler.go

@@ -28,6 +28,7 @@ import (
 	client "github.com/devtron-labs/devtron/api/helm-app"
 	"github.com/devtron-labs/devtron/api/restHandler/common"
 	"github.com/devtron-labs/devtron/client/argocdServer/application"
+	"github.com/devtron-labs/devtron/client/cron"
 	"github.com/devtron-labs/devtron/internal/constants"
 	"github.com/devtron-labs/devtron/internal/util"
 	"github.com/devtron-labs/devtron/pkg/app"
@@ -63,21 +64,22 @@ type AppListingRestHandler interface {
 }
 
 type AppListingRestHandlerImpl struct {
-	application            application.ServiceClient
-	appListingService      app.AppListingService
-	teamService            team.TeamService
-	enforcer               casbin.Enforcer
-	pipeline               pipeline.PipelineBuilder
-	logger                 *zap.SugaredLogger
-	enforcerUtil           rbac.EnforcerUtil
-	deploymentGroupService deploymentGroup.DeploymentGroupService
-	userService            user.UserService
-	helmAppClient          client.HelmAppClient
-	clusterService         cluster.ClusterService
-	helmAppService         client.HelmAppService
-	argoUserService        argo.ArgoUserService
-	k8sApplicationService  k8s.K8sApplicationService
-	installedAppService    service1.InstalledAppService
+	application                      application.ServiceClient
+	appListingService                app.AppListingService
+	teamService                      team.TeamService
+	enforcer                         casbin.Enforcer
+	pipeline                         pipeline.PipelineBuilder
+	logger                           *zap.SugaredLogger
+	enforcerUtil                     rbac.EnforcerUtil
+	deploymentGroupService           deploymentGroup.DeploymentGroupService
+	userService                      user.UserService
+	helmAppClient                    client.HelmAppClient
+	clusterService                   cluster.ClusterService
+	helmAppService                   client.HelmAppService
+	argoUserService                  argo.ArgoUserService
+	k8sApplicationService            k8s.K8sApplicationService
+	installedAppService              service1.InstalledAppService
+	cdApplicationStatusUpdateHandler cron.CdApplicationStatusUpdateHandler
 }
 
 type AppStatus struct {
@@ -96,23 +98,25 @@ func NewAppListingRestHandlerImpl(application application.ServiceClient,
 	logger *zap.SugaredLogger, enforcerUtil rbac.EnforcerUtil,
 	deploymentGroupService deploymentGroup.DeploymentGroupService, userService user.UserService,
 	helmAppClient client.HelmAppClient, clusterService cluster.ClusterService, helmAppService client.HelmAppService,
-	argoUserService argo.ArgoUserService, k8sApplicationService k8s.K8sApplicationService, installedAppService service1.InstalledAppService) *AppListingRestHandlerImpl {
+	argoUserService argo.ArgoUserService, k8sApplicationService k8s.K8sApplicationService, installedAppService service1.InstalledAppService,
+	cdApplicationStatusUpdateHandler cron.CdApplicationStatusUpdateHandler) *AppListingRestHandlerImpl {
 	appListingHandler := &AppListingRestHandlerImpl{
-		application:            application,
-		appListingService:      appListingService,
-		logger:                 logger,
-		teamService:            teamService,
-		pipeline:               pipeline,
-		enforcer:               enforcer,
-		enforcerUtil:           enforcerUtil,
-		deploymentGroupService: deploymentGroupService,
-		userService:            userService,
-		helmAppClient:          helmAppClient,
-		clusterService:         clusterService,
-		helmAppService:         helmAppService,
-		argoUserService:        argoUserService,
-		k8sApplicationService:  k8sApplicationService,
-		installedAppService:    installedAppService,
+		application:                      application,
+		appListingService:                appListingService,
+		logger:                           logger,
+		teamService:                      teamService,
+		pipeline:                         pipeline,
+		enforcer:                         enforcer,
+		enforcerUtil:                     enforcerUtil,
+		deploymentGroupService:           deploymentGroupService,
+		userService:                      userService,
+		helmAppClient:                    helmAppClient,
+		clusterService:                   clusterService,
+		helmAppService:                   helmAppService,
+		argoUserService:                  argoUserService,
+		k8sApplicationService:            k8sApplicationService,
+		installedAppService:              installedAppService,
+		cdApplicationStatusUpdateHandler: cdApplicationStatusUpdateHandler,
 	}
 	return appListingHandler
 }
@@ -722,6 +726,12 @@ func (handler AppListingRestHandlerImpl) fetchResourceTree(w http.ResponseWriter
 		}
 		appDetail.ResourceTree = util2.InterfaceToMapAdapter(resp)
 		handler.logger.Debugw("application environment status", "appId", appId, "envId", envId, "resp", resp)
+		if resp.Status == string(health.HealthStatusHealthy) {
+			err = handler.cdApplicationStatusUpdateHandler.SyncPipelineStatusForResourceTreeCall(acdAppName, appId, envId)
+			if err != nil {
+				handler.logger.Errorw("error in syncing pipeline status", "err", err)
+			}
+		}
 	} else if len(appDetail.AppName) > 0 && len(appDetail.EnvironmentName) > 0 && util.IsHelmApp(appDetail.DeploymentAppType) {
 		config, err := handler.helmAppService.GetClusterConf(appDetail.ClusterId)
 		if err != nil {

api/router/pubsub/ApplicationStatusUpdateHandler.go

@@ -19,6 +19,7 @@ package pubsub
 
 import (
 	"encoding/json"
+	"time"
 
 	v1alpha12 "github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
 	"github.com/devtron-labs/devtron/client/pubsub"
@@ -68,6 +69,7 @@ func NewApplicationStatusUpdateHandlerImpl(logger *zap.SugaredLogger, pubsubClie
 type ApplicationDetail struct {
 	Application    *v1alpha12.Application `json:"application"`
 	OldApplication *v1alpha12.Application `json:"oldApplication"`
+	StatusTime     time.Time              `json:"statusTime"`
 }
 
 func (impl *ApplicationStatusUpdateHandlerImpl) Subscribe() error {
@@ -87,8 +89,10 @@ func (impl *ApplicationStatusUpdateHandlerImpl) Subscribe() error {
 			return
 		}
 		//impl.logger.Infow("app update request", "application", newApp)
-
-		isHealthy, err := impl.appService.UpdateApplicationStatusAndCheckIsHealthy(newApp, oldApp)
+		if applicationDetail.StatusTime.IsZero() {
+			applicationDetail.StatusTime = time.Now()
+		}
+		isHealthy, err := impl.appService.UpdateApplicationStatusAndCheckIsHealthy(newApp, oldApp, applicationDetail.StatusTime)
 		if err != nil {
 			impl.logger.Errorw("error on application status update", "err", err, "msg", string(msg.Data))
 

client/argocdServer/application/Application.go

@@ -93,6 +93,7 @@ type ResourceTreeResponse struct {
 	*v1alpha1.ApplicationTree
 	NewGenerationReplicaSets []string                        `json:"newGenerationReplicaSets"`
 	Status                   string                          `json:"status"`
+	RevisionHash             string                          `json:"revisionHash"`
 	PodMetadata              []*PodMetadata                  `json:"podMetadata"`
 	Conditions               []v1alpha1.ApplicationCondition `json:"conditions"`
 }
@@ -389,10 +390,12 @@ func (c ServiceClientImpl) ResourceTree(ctxt context.Context, query *application
 	app, err := asc.Watch(ctxt, &appQuery)
 	var conditions = make([]v1alpha1.ApplicationCondition, 0)
 	status := "Unknown"
+	hash := ""
 	if app != nil {
 		appResp, err := app.Recv()
 		if err == nil {
 			status = string(appResp.Application.Status.Health.Status)
+			hash = appResp.Application.Status.Sync.Revision
 			conditions = appResp.Application.Status.Conditions
 			for _, condition := range conditions {
 				if condition.Type != v1alpha1.ApplicationConditionSharedResourceWarning {
@@ -404,7 +407,7 @@ func (c ServiceClientImpl) ResourceTree(ctxt context.Context, query *application
 			}
 		}
 	}
-	return &ResourceTreeResponse{resp, newReplicaSets, status, podMetadata, conditions}, err
+	return &ResourceTreeResponse{resp, newReplicaSets, status, hash, podMetadata, conditions}, err
 }
 
 func (c ServiceClientImpl) buildPodMetadata(resp *v1alpha1.ApplicationTree, responses []*Result) (podMetaData []*PodMetadata, newReplicaSets []string) {
@@ -426,15 +429,15 @@ func (c ServiceClientImpl) buildPodMetadata(resp *v1alpha1.ApplicationTree, resp
 				err := json.Unmarshal([]byte(manifestFromResponse), &manifest)
 				if err != nil {
 					c.logger.Error(err)
-				}else{
+				} else {
 					rolloutManifests = append(rolloutManifests, manifest)
 				}
 			} else if kind == "Deployment" {
 				manifest := make(map[string]interface{})
 				err := json.Unmarshal([]byte(manifestFromResponse), &manifest)
 				if err != nil {
 					c.logger.Error(err)
-				}else{
+				} else {
 					deploymentManifests = append(deploymentManifests, manifest)
 				}
 			} else if kind == "StatefulSet" {

client/cron/CdApplicationStatusUpdateHandler.go

@@ -1,11 +1,17 @@
 package cron
 
 import (
+	"encoding/json"
 	"fmt"
 	"github.com/caarlos0/env"
+	client2 "github.com/devtron-labs/devtron/client/events"
+	"github.com/devtron-labs/devtron/client/pubsub"
+	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig"
 	"github.com/devtron-labs/devtron/pkg/app"
 	"github.com/devtron-labs/devtron/pkg/appStore/deployment/service"
 	"github.com/devtron-labs/devtron/pkg/pipeline"
+	"github.com/devtron-labs/devtron/util"
+	"github.com/nats-io/nats.go"
 	"github.com/robfig/cron/v3"
 	"go.uber.org/zap"
 	"strconv"
@@ -14,16 +20,22 @@ import (
 type CdApplicationStatusUpdateHandler interface {
 	HelmApplicationStatusUpdate()
 	ArgoApplicationStatusUpdate()
+	ArgoPipelineTimelineUpdate()
+	Subscribe() error
+	SyncPipelineStatusForResourceTreeCall(acdAppName string, appId, envId int) error
 }
 
 type CdApplicationStatusUpdateHandlerImpl struct {
-	logger              *zap.SugaredLogger
-	cron                *cron.Cron
-	appService          app.AppService
-	workflowDagExecutor pipeline.WorkflowDagExecutor
-	installedAppService service.InstalledAppService
-	CdHandler           pipeline.CdHandler
-	AppStatusConfig     *AppStatusConfig
+	logger                           *zap.SugaredLogger
+	cron                             *cron.Cron
+	appService                       app.AppService
+	workflowDagExecutor              pipeline.WorkflowDagExecutor
+	installedAppService              service.InstalledAppService
+	CdHandler                        pipeline.CdHandler
+	AppStatusConfig                  *AppStatusConfig
+	pubsubClient                     *pubsub.PubSubClient
+	pipelineStatusTimelineRepository pipelineConfig.PipelineStatusTimelineRepository
+	eventClient                      client2.EventClient
 }
 
 type AppStatusConfig struct {
@@ -43,20 +55,34 @@ func GetAppStatusConfig() (*AppStatusConfig, error) {
 
 func NewCdApplicationStatusUpdateHandlerImpl(logger *zap.SugaredLogger, appService app.AppService,
 	workflowDagExecutor pipeline.WorkflowDagExecutor, installedAppService service.InstalledAppService,
-	CdHandler pipeline.CdHandler, AppStatusConfig *AppStatusConfig) *CdApplicationStatusUpdateHandlerImpl {
+	CdHandler pipeline.CdHandler, AppStatusConfig *AppStatusConfig, pubsubClient *pubsub.PubSubClient,
+	pipelineStatusTimelineRepository pipelineConfig.PipelineStatusTimelineRepository,
+	eventClient client2.EventClient) *CdApplicationStatusUpdateHandlerImpl {
 	cron := cron.New(
 		cron.WithChain())
 	cron.Start()
 	impl := &CdApplicationStatusUpdateHandlerImpl{
-		logger:              logger,
-		cron:                cron,
-		appService:          appService,
-		workflowDagExecutor: workflowDagExecutor,
-		installedAppService: installedAppService,
-		CdHandler:           CdHandler,
-		AppStatusConfig:     AppStatusConfig,
-	}
-	_, err := cron.AddFunc(AppStatusConfig.CdPipelineStatusCronTime, impl.HelmApplicationStatusUpdate)
+		logger:                           logger,
+		cron:                             cron,
+		appService:                       appService,
+		workflowDagExecutor:              workflowDagExecutor,
+		installedAppService:              installedAppService,
+		CdHandler:                        CdHandler,
+		AppStatusConfig:                  AppStatusConfig,
+		pubsubClient:                     pubsubClient,
+		pipelineStatusTimelineRepository: pipelineStatusTimelineRepository,
+		eventClient:                      eventClient,
+	}
+	err := util.AddStream(pubsubClient.JetStrCtxt, util.ORCHESTRATOR_STREAM)
+	if err != nil {
+		return nil
+	}
+	err = impl.Subscribe()
+	if err != nil {
+		logger.Errorw("error on subscribe", "err", err)
+		return nil
+	}
+	_, err = cron.AddFunc(AppStatusConfig.CdPipelineStatusCronTime, impl.HelmApplicationStatusUpdate)
 	if err != nil {
 		logger.Errorw("error in starting helm application status update cron job", "err", err)
 		return nil
@@ -66,9 +92,38 @@ func NewCdApplicationStatusUpdateHandlerImpl(logger *zap.SugaredLogger, appServi
 		logger.Errorw("error in starting argo application status update cron job", "err", err)
 		return nil
 	}
+	_, err = cron.AddFunc("@every 1m", impl.ArgoPipelineTimelineUpdate)
+	if err != nil {
+		logger.Errorw("error in starting argo application status update cron job", "err", err)
+		return nil
+	}
 	return impl
 }
 
+func (impl *CdApplicationStatusUpdateHandlerImpl) Subscribe() error {
+	_, err := impl.pubsubClient.JetStrCtxt.QueueSubscribe(util.ARGO_PIPELINE_STATUS_UPDATE_TOPIC, util.ARGO_PIPELINE_STATUS_UPDATE_GROUP, func(msg *nats.Msg) {
+		impl.logger.Debug("received argo pipeline status update request")
+		defer msg.Ack()
+		statusUpdateEvent := pipeline.ArgoPipelineStatusEvent{}
+		err := json.Unmarshal([]byte(string(msg.Data)), &statusUpdateEvent)
+		if err != nil {
+			impl.logger.Errorw("unmarshal error on argo pipeline status update event", "err", err)
+			return
+		}
+		impl.logger.Infow("ARGO_PIPELINE_STATUS_UPDATE_REQ", "stage", "subscribeDataUnmarshal", "data", statusUpdateEvent)
+		err = impl.CdHandler.UpdatePipelineTimelineAndStatusByLiveResourceTreeFetch(statusUpdateEvent.ArgoAppName, statusUpdateEvent.AppId, statusUpdateEvent.EnvId, statusUpdateEvent.IgnoreFailedWorkflowStatus)
+		if err != nil {
+			impl.logger.Errorw("error on argo pipeline status update", "err", err, "msg", string(msg.Data))
+			return
+		}
+	}, nats.Durable(util.ARGO_PIPELINE_STATUS_UPDATE_DURABLE), nats.DeliverLast(), nats.ManualAck(), nats.BindStream(util.ORCHESTRATOR_STREAM))
+	if err != nil {
+		impl.logger.Error("error in subscribing to argo application status update topic", "err", err)
+		return err
+	}
+	return nil
+}
+
 func (impl *CdApplicationStatusUpdateHandlerImpl) HelmApplicationStatusUpdate() {
 	degradedTime, err := strconv.Atoi(impl.AppStatusConfig.PipelineDegradedTime)
 	if err != nil {
@@ -97,3 +152,49 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) ArgoApplicationStatusUpdate()
 	}
 	return
 }
+
+func (impl *CdApplicationStatusUpdateHandlerImpl) ArgoPipelineTimelineUpdate() {
+	err := impl.CdHandler.CheckArgoPipelineTimelineStatusPeriodicallyAndUpdateInDb(30)
+	if err != nil {
+		impl.logger.Errorw("error argo app status update - cron job", "err", err)
+		return
+	}
+	return
+}
+
+func (impl *CdApplicationStatusUpdateHandlerImpl) SyncPipelineStatusForResourceTreeCall(acdAppName string, appId, envId int) error {
+	timeline, err := impl.pipelineStatusTimelineRepository.FetchLatestTimelineByAppIdAndEnvId(appId, envId)
+	if err != nil {
+		impl.logger.Errorw("error in getting timeline", "err", err)
+		return err
+	}
+
+	if !IsTerminalTimelineStatus(timeline.Status) {
+		//create new nats event
+		statusUpdateEvent := pipeline.ArgoPipelineStatusEvent{
+			ArgoAppName:                acdAppName,
+			AppId:                      appId,
+			EnvId:                      envId,
+			IgnoreFailedWorkflowStatus: true,
+		}
+		//write event
+		err := impl.eventClient.WriteNatsEvent(util.ARGO_PIPELINE_STATUS_UPDATE_TOPIC, statusUpdateEvent)
+		if err != nil {
+			impl.logger.Errorw("error in writing nats event", "topic", util.ARGO_PIPELINE_STATUS_UPDATE_TOPIC, "payload", statusUpdateEvent)
+			return err
+		}
+	}
+	return nil
+}
+
+func IsTerminalTimelineStatus(timeline pipelineConfig.TimelineStatus) bool {
+	switch timeline {
+	case
+		pipelineConfig.TIMELINE_STATUS_APP_HEALTHY,
+		pipelineConfig.TIMELINE_STATUS_APP_DEGRADED,
+		pipelineConfig.TIMELINE_STATUS_DEPLOYMENT_FAILED,
+		pipelineConfig.TIMELINE_STATUS_GIT_COMMIT_FAILED:
+		return true
+	}
+	return false
+}