Skip to content

enhancement: made pipeline status crons cascading #3372

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
May 11, 2023
Merged

enhancement: made pipeline status crons cascading #3372

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
987b541
made pipeline status crons cascading, added metrics for cron process …
kartik-579 May 10, 2023
362e55b
updated defer calls
kartik-579 May 11, 2023
afc21a6
updated label for cron status metric
kartik-579 May 11, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 29 additions & 1 deletion client/cron/CdApplicationStatusUpdateHandler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -16,6 +17,7 @@ import (
"github.com/robfig/cron/v3"
"go.uber.org/zap"
"strconv"
"time"
)

type CdApplicationStatusUpdateHandler interface {
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -141,6 +157,10 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) Subscribe() error {
}

func (impl *CdApplicationStatusUpdateHandlerImpl) HelmApplicationStatusUpdate() {
cronProcessStartTime := time.Now()
defer func() {
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)
Expand All @@ -155,6 +175,14 @@ func (impl *CdApplicationStatusUpdateHandlerImpl) HelmApplicationStatusUpdate()
}

func (impl *CdApplicationStatusUpdateHandlerImpl) ArgoApplicationStatusUpdate() {
cronProcessStartTime := time.Now()
defer func() {
middleware.DeploymentStatusCronDuration.WithLabelValues(pipeline.DEVTRON_APP_ARGO_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 func() {
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)
Expand Down
4 changes: 4 additions & 0 deletions internal/middleware/instrument.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
19 changes: 13 additions & 6 deletions pkg/pipeline/CdHandler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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
Expand Down