feat(api): cancel abandonned run results

contrib/grpcplugins/action/artifactory-promote/main.go

@@ -45,7 +45,7 @@ func (p *rtPromotePlugin) Run(ctx context.Context, q *actionplugin.ActionQuery)
 
 	if err := p.perform(ctx, artifacts, maturity, properties); err != nil {
 		res.Status = sdk.StatusFail
-		res.Status = err.Error()
+		res.Details = err.Error()
 		return res, err
 	}
 

contrib/grpcplugins/action/docker-push/main.go

@@ -58,9 +58,13 @@ func (actPlugin *dockerPushPlugin) Run(ctx context.Context, q *actionplugin.Acti
 	tags = strings.Replace(tags, ";", ",", -1) // If tags are separated by <semicolon>
 	tagSlice := strings.Split(tags, ",")
 
+	if !strings.ContainsRune(image, ':') { // Latest is the default tag
+		image = image + ":latest"
+	}
+
 	if err := actPlugin.perform(ctx, image, tagSlice, registry, auth); err != nil {
 		res.Status = sdk.StatusFail
-		res.Status = err.Error()
+		res.Details = err.Error()
 		return res, err
 	}
 

contrib/grpcplugins/action/helm-push/main.go

@@ -11,6 +11,7 @@ import (
 	"path"
 	"path/filepath"
 	"strconv"
+	"strings"
 	"time"
 
 	cm "github.com/chartmuseum/helm-push/pkg/chartmuseum"
@@ -68,11 +69,15 @@ func (p *helmPushPlugin) Run(ctx context.Context, q *actionplugin.ActionQuery) (
 	registryAccessToken := q.GetOptions()["registryAccessToken"]
 	registryAuthHeader := q.GetOptions()["registryAuthHeader"]
 
-	updateDependencies, err := strconv.ParseBool(updateDependenciesS)
-	if err != nil {
-		res.Status = sdk.StatusFail
-		res.Status = "invalid value for parameter <updateDependencies>"
-		return res, err
+	var updateDependencies = false
+	if strings.TrimSpace(updateDependenciesS) != "" {
+		var err error
+		updateDependencies, err = strconv.ParseBool(updateDependenciesS)
+		if err != nil {
+			res.Status = sdk.StatusFail
+			res.Details = "invalid value for parameter <updateDependencies>"
+			return res, err
+		}
 	}
 
 	registryOpts := chartMuseumOptions{
@@ -85,7 +90,7 @@ func (p *helmPushPlugin) Run(ctx context.Context, q *actionplugin.ActionQuery) (
 	result, d, err := p.perform(ctx, chartFolder, chartVersion, appVersion, updateDependencies, registryOpts)
 	if err != nil {
 		res.Status = sdk.StatusFail
-		res.Status = err.Error()
+		res.Details = err.Error()
 		return res, err
 	}
 

engine/api/api.go

@@ -906,9 +906,12 @@ func (a *API) Serve(ctx context.Context) error {
 	a.GoRoutines.RunWithRestart(ctx, "api.StopDeadJobs", func(ctx context.Context) {
 		a.StopDeadJobs(ctx)
 	})
-	a.GoRoutines.RunWithRestart(ctx, "api.xx", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "api.TriggerBlockedWorkflowRuns", func(ctx context.Context) {
 		a.TriggerBlockedWorkflowRuns(ctx)
 	})
+	a.GoRoutines.RunWithRestart(ctx, "api.CancelAbandonnedRunResults", func(ctx context.Context) {
+		a.CancelAbandonnedRunResults(ctx)
+	})
 
 	a.GoRoutines.RunWithRestart(ctx, "api.repositoryAnalysisPoller", func(ctx context.Context) {
 		a.repositoryAnalysisPoller(ctx, 5*time.Second)

engine/api/v2_workflow_run_craft.go

@@ -699,6 +699,10 @@ func buildRunContext(ctx context.Context, db *gorp.DbMap, store cache.Store, p s
 			semverCurrent = "0.1.0+" + strconv.FormatInt(wr.RunNumber, 10) + ".sha." + commit
 		}
 
+		// We replace the metadata "+" from semver because a lot of tools doesn't support it (docker, artifactory, ...)
+		semverNext = strings.ReplaceAll(semverNext, "+", "-")
+		semverCurrent = strings.ReplaceAll(semverCurrent, "+", "-")
+
 	case wr.RunEvent.ModelUpdateTrigger != nil:
 		ref = wr.RunEvent.ModelUpdateTrigger.Ref
 	case wr.RunEvent.WorkflowUpdateTrigger != nil:

engine/api/v2_workflow_run_job_routines.go

@@ -18,8 +18,62 @@ import (
 
 const jobLockKey = "jobs:lock"
 
+func (api *API) CancelAbandonnedRunResults(ctx context.Context) {
+	tick := time.NewTicker(5 * time.Minute)
+	defer tick.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() != nil {
+				log.Error(ctx, "%v", ctx.Err())
+			}
+			return
+		case <-tick.C:
+			ids, err := workflow_v2.LoadAbandonnedRunResultsID(ctx, api.mustDB())
+			if err != nil {
+				log.ErrorWithStackTrace(ctx, err)
+				continue
+			}
+			for _, id := range ids {
+				if err := api.cancelAbandonnedRunResult(ctx, api.mustDB(), id); err != nil {
+					log.ErrorWithStackTrace(ctx, err)
+				}
+			}
+		}
+	}
+}
+
+func (api *API) cancelAbandonnedRunResult(ctx context.Context, db *gorp.DbMap, id string) error {
+	tx, err := db.Begin()
+	if err != nil {
+		return sdk.WithStack(err)
+	}
+
+	defer tx.Rollback()
+
+	runResult, err := workflow_v2.LoadAndLockRunResultByID(ctx, tx, id)
+	if err != nil {
+		return err
+	}
+
+	if runResult == nil {
+		log.Debug(ctx, "RunResult %s skipped", id)
+		return nil
+	}
+
+	log.Debug(ctx, "cancelAbandonnedRunResult: %s", id)
+
+	runResult.Status = sdk.V2WorkflowRunResultStatusCanceled
+	if err := workflow_v2.UpdateRunResult(ctx, tx, runResult); err != nil {
+		return err
+	}
+
+	return sdk.WithStack(tx.Commit())
+}
+
 func (api *API) StopDeadJobs(ctx context.Context) {
 	tickStopDeadJobs := time.NewTicker(1 * time.Minute)
+	defer tickStopDeadJobs.Stop()
 	for {
 		select {
 		case <-ctx.Done():
@@ -44,6 +98,7 @@ func (api *API) StopDeadJobs(ctx context.Context) {
 
 func (api *API) ReEnqueueScheduledJobs(ctx context.Context) {
 	tickScheduledJob := time.NewTicker(1 * time.Minute)
+	defer tickScheduledJob.Stop()
 	for {
 		select {
 		case <-ctx.Done():

engine/api/workflow_v2/dao_run.go

@@ -127,6 +127,19 @@ func LoadRunResult(ctx context.Context, db gorp.SqlExecutor, runID string, id st
 	return &result.V2WorkflowRunResult, nil
 }
 
+func LoadAndLockRunResultByID(ctx context.Context, db gorp.SqlExecutor, id string) (*sdk.V2WorkflowRunResult, error) {
+	query := gorpmapping.NewQuery(`select * from v2_workflow_run_result where id = $1 for update skip locked`).Args(id)
+	var result dbV2WorkflowRunResult
+	found, err := gorpmapping.Get(ctx, db, query, &result)
+	if err != nil {
+		return nil, sdk.WrapError(err, "unable to load run result %v", id)
+	}
+	if !found {
+		return nil, nil
+	}
+	return &result.V2WorkflowRunResult, nil
+}
+
 func LoadRunResultsByRunJobID(ctx context.Context, db gorp.SqlExecutor, runJobID string) ([]sdk.V2WorkflowRunResult, error) {
 	query := gorpmapping.NewQuery(`select * from v2_workflow_run_result where workflow_run_job_id = $1`).Args(runJobID)
 	var result []dbV2WorkflowRunResult
@@ -171,6 +184,21 @@ func loadRunResultsByRunIDs(ctx context.Context, db gorp.SqlExecutor, runIDs ...
 	return mapRes, nil
 }
 
+func LoadAbandonnedRunResultsID(ctx context.Context, db gorp.SqlExecutor) ([]string, error) {
+	query := `
+  	select v2_workflow_run_result.id 
+	from v2_workflow_run_result 
+	join v2_workflow_run_job on v2_workflow_run_job.id = v2_workflow_run_result.workflow_run_job_id
+	where v2_workflow_run_job.status IN ('Fail', 'Stopped')
+	and v2_workflow_run_result.status = 'PENDING'
+	order by v2_workflow_run_result.issued_at ASC`
+	var results pq.StringArray
+	if _, err := db.Select(&results, query); err != nil {
+		return nil, sdk.WrapError(err, "unable to load abandonned run results")
+	}
+	return results, nil
+}
+
 func getRuns(ctx context.Context, db gorp.SqlExecutor, query gorpmapping.Query, opts ...gorpmapper.GetOptionFunc) ([]sdk.V2WorkflowRun, error) {
 	var dbWkfRuns []dbWorkflowRun
 	if err := gorpmapping.GetAll(ctx, db, query, &dbWkfRuns, opts...); err != nil {

sdk/v2_workflow_run.go

@@ -546,6 +546,7 @@ const (
 	V2WorkflowRunResultStatusCompleted = "COMPLETED"
 	V2WorkflowRunResultStatusPromoted  = "PROMOTED"
 	V2WorkflowRunResultStatusReleased  = "RELEASED"
+	V2WorkflowRunResultStatusCanceled  = "CANCELED"
 )
 
 type V2WorkflowRunResultArtifactManagerMetadata map[string]string