made pipeline status crons cascading, added metrics for cron process duration

Author: kartik-579

client/cron/CdApplicationStatusUpdateHandler.go

@@ -6,6 +6,7 @@ import (
 	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
 	"github.com/devtron-labs/devtron/api/bean"
 	client2 "github.com/devtron-labs/devtron/client/events"
+	"github.com/devtron-labs/devtron/internal/middleware"
 	"github.com/devtron-labs/devtron/internal/sql/repository"
 	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig"
 	"github.com/devtron-labs/devtron/pkg/app"
@@ -16,6 +17,7 @@ import (
 	"github.com/robfig/cron/v3"
 	"go.uber.org/zap"
 	"strconv"
+	"time"
 )
 
 type CdApplicationStatusUpdateHandler interface {
@@ -46,6 +48,19 @@ type CdApplicationStatusUpdateHandlerImpl struct {
 	installedAppVersionRepository        repository2.InstalledAppRepository
 }
 
+type CronLoggerImpl struct {
+	logger *zap.SugaredLogger
+}
+
+func (impl *CronLoggerImpl) Info(msg string, keysAndValues ...interface{}) {
+	impl.logger.Infow(msg, keysAndValues...)
+}
+
+func (impl *CronLoggerImpl) Error(err error, msg string, keysAndValues ...interface{}) {
+	keysAndValues = append([]interface{}{"err", err}, keysAndValues...)
+	impl.logger.Errorw(msg, keysAndValues...)
+}
+
 func NewCdApplicationStatusUpdateHandlerImpl(logger *zap.SugaredLogger, appService app.AppService,
 	workflowDagExecutor pipeline.WorkflowDagExecutor, installedAppService service.InstalledAppService,
 	CdHandler pipeline.CdHandler, AppStatusConfig *app.AppServiceConfig, pubsubClient *pubsub.PubSubClientServiceImpl,
@@ -54,8 +69,9 @@ func NewCdApplicationStatusUpdateHandlerImpl(logger *zap.SugaredLogger, appServi
 	cdWorkflowRepository pipelineConfig.CdWorkflowRepository,
 	pipelineRepository pipelineConfig.PipelineRepository, installedAppVersionHistoryRepository repository2.InstalledAppVersionHistoryRepository,
 	installedAppVersionRepository repository2.InstalledAppRepository) *CdApplicationStatusUpdateHandlerImpl {
+	cronLogger := &CronLoggerImpl{logger: logger}
 	cron := cron.New(
-		cron.WithChain())
+		cron.WithChain(cron.SkipIfStillRunning(cronLogger)))
 	cron.Start()
 	impl := &CdApplicationStatusUpdateHandlerImpl{
 		logger:                               logger,
@@ -141,6 +157,8 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) Subscribe() error {
 }
 
 func (impl *CdApplicationStatusUpdateHandlerImpl) HelmApplicationStatusUpdate() {
+	cronProcessStartTime := time.Now()
+	defer middleware.DeploymentStatusCronDuration.WithLabelValues(pipeline.DEVTRON_APP_HELM_PIPELINE_STATUS_UPDATE_CRON).Observe(time.Since(cronProcessStartTime).Seconds())
 	HelmPipelineStatusCheckEligibleTime, err := strconv.Atoi(impl.AppStatusConfig.HelmPipelineStatusCheckEligibleTime)
 	if err != nil {
 		impl.logger.Errorw("error in converting string to int", "err", err)
@@ -155,6 +173,11 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) HelmApplicationStatusUpdate()
 }
 
 func (impl *CdApplicationStatusUpdateHandlerImpl) ArgoApplicationStatusUpdate() {
+	cronProcessStartTime := time.Now()
+	defer middleware.DeploymentStatusCronDuration.WithLabelValues(pipeline.DEVTRON_APP_HELM_PIPELINE_STATUS_UPDATE_CRON).Observe(time.Since(cronProcessStartTime).Seconds())
+	//TODO: remove below cron with division of cron for argo pipelines of devtron-apps and helm-apps
+	defer middleware.DeploymentStatusCronDuration.WithLabelValues(pipeline.HELM_APP_ARGO_PIPELINE_STATUS_UPDATE_CRON).Observe(time.Since(cronProcessStartTime).Seconds())
+
 	getPipelineDeployedBeforeMinutes, err := strconv.Atoi(impl.AppStatusConfig.PipelineDegradedTime)
 	if err != nil {
 		impl.logger.Errorw("error in converting string to int", "err", err)

internal/middleware/instrument.go

@@ -113,6 +113,10 @@ var CiTriggerCounter = promauto.NewCounterVec(prometheus.CounterOpts{
 	Name: "ci_trigger_counter",
 }, []string{"appName", "pipelineName"})
 
+var DeploymentStatusCronDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Name: "deployment_status_cron_process_time",
+}, []string{"cronName"})
+
 // prometheusMiddleware implements mux.MiddlewareFunc.
 func PrometheusMiddleware(next http.Handler) http.Handler {
 	//	prometheus.MustRegister(requestCounter)

pkg/pipeline/CdHandler.go

@@ -56,6 +56,12 @@ import (
 	"time"
 )
 
+const (
+	DEVTRON_APP_HELM_PIPELINE_STATUS_UPDATE_CRON = "DTAppHelmPipelineStatusUpdateCron"
+	DEVTRON_APP_ARGO_PIPELINE_STATUS_UPDATE_CRON = "DTAppArgoPipelineStatusUpdateCron"
+	HELM_APP_ARGO_PIPELINE_STATUS_UPDATE_CRON    = "HelmAppArgoPipelineStatusUpdateCron"
+)
+
 type CdHandler interface {
 	UpdateWorkflow(workflowStatus v1alpha1.WorkflowStatus) (int, string, error)
 	GetCdBuildHistory(appId int, environmentId int, pipelineId int, offset int, size int) ([]pipelineConfig.CdWorkflowWithArtifact, error)
@@ -187,18 +193,19 @@ func (impl *CdHandlerImpl) CheckArgoAppStatusPeriodicallyAndUpdateInDb(getPipeli
 		impl.Logger.Errorw("error in getting pipelines having latest trigger stuck in non terminal statuses", "err", err)
 		return err
 	}
-	impl.Logger.Debugw("received stuck argo cd pipelines", "pipelines", pipelines)
+	impl.Logger.Debugw("received stuck argo cd pipelines", "pipelines", pipelines, "number of pipelines", len(pipelines))
+
+	for _, pipeline := range pipelines {
+		impl.CheckAndSendArgoPipelineStatusSyncEventIfNeeded(pipeline.Id, 1, false)
+	}
 
 	installedAppVersions, err := impl.installedAppRepository.GetArgoPipelinesHavingLatestTriggerStuckInNonTerminalStatusesForAppStore(getPipelineDeployedBeforeMinutes, getPipelineDeployedWithinHours)
 	if err != nil {
 		impl.Logger.Errorw("error in getting installedAppVersions having latest trigger stuck in non terminal statuses", "err", err)
 		return err
 	}
-	impl.Logger.Debugw("received stuck argo installed appStore app", "pipelines", installedAppVersions)
+	impl.Logger.Debugw("received stuck argo installed appStore app", "installedAppVersions", installedAppVersions, "number of triggers", len(installedAppVersions))
 
-	for _, pipeline := range pipelines {
-		impl.CheckAndSendArgoPipelineStatusSyncEventIfNeeded(pipeline.Id, 1, false)
-	}
 	for _, installedAppVersion := range installedAppVersions {
 		impl.CheckAndSendArgoPipelineStatusSyncEventIfNeeded(installedAppVersion.Id, 1, true)
 	}
@@ -458,7 +465,7 @@ func (impl *CdHandlerImpl) CheckHelmAppStatusPeriodicallyAndUpdateInDb(helmPipel
 		impl.Logger.Errorw("error in getting latest triggers of helm pipelines which are stuck in non terminal statuses", "err", err)
 		return err
 	}
-	impl.Logger.Infow("checking helm app status for non terminal deployment triggers", "wfrList", wfrList)
+	impl.Logger.Debugw("checking helm app status for non terminal deployment triggers", "wfrList", wfrList, "number of wfr", len(wfrList))
 	for _, wfr := range wfrList {
 		if time.Now().Sub(wfr.UpdatedOn) <= time.Duration(helmPipelineStatusCheckEligibleTime)*time.Second {
 			//if wfr is updated within configured time then do not include for this cron cycle