feat(ui,api): display spawn info on waiting job

engine/api/api_routes.go

@@ -261,6 +261,7 @@ func (api *API) InitRouter() {
 	r.Handle("/project/{key}/workflows/{permWorkflowName}/runs/{number}/nodes/{nodeRunID}/stop", r.POSTEXECUTE(api.stopWorkflowNodeRunHandler))
 	r.Handle("/project/{key}/workflows/{permWorkflowName}/runs/{number}/nodes/{nodeID}/history", r.GET(api.getWorkflowNodeRunHistoryHandler))
 	r.Handle("/project/{key}/workflows/{permWorkflowName}/runs/{number}/{nodeName}/commits", r.GET(api.getWorkflowCommitsHandler))
+	r.Handle("/project/{key}/workflows/{permWorkflowName}/runs/{number}/nodes/{nodeRunID}/job/{runJobId}/info", r.GET(api.getWorkflowNodeRunJobSpawnInfosHandler))
 	r.Handle("/project/{key}/workflows/{permWorkflowName}/runs/{number}/nodes/{nodeRunID}/job/{runJobId}/log/service", r.GET(api.getWorkflowNodeRunJobServiceLogsHandler))
 	r.Handle("/project/{key}/workflows/{permWorkflowName}/runs/{number}/nodes/{nodeRunID}/job/{runJobId}/step/{stepOrder}", r.GET(api.getWorkflowNodeRunJobStepHandler))
 	r.Handle("/project/{key}/workflows/{permWorkflowName}/artifact/{artifactId}", r.GET(api.getDownloadArtifactHandler))

engine/api/workflow/dao_node_job_run_info.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
+	"sort"
 	"time"
 
 	"github.com/go-gorp/gorp"
@@ -13,12 +14,12 @@ import (
 	"github.com/ovh/cds/sdk/log"
 )
 
-//loadNodeRunJobInfo load infos (workflow_node_run_job_infos) for a job (workflow_node_run_job)
-func loadNodeRunJobInfo(db gorp.SqlExecutor, jobID int64) ([]sdk.SpawnInfo, error) {
+// LoadNodeRunJobInfo load infos (workflow_node_run_job_infos) for a job (workflow_node_run_job)
+func LoadNodeRunJobInfo(db gorp.SqlExecutor, jobID int64) ([]sdk.SpawnInfo, error) {
 	res := []struct {
 		Bytes sql.NullString `db:"spawninfos"`
 	}{}
-	query := "SELECT spawninfos FROM workflow_node_run_job_info WHERE workflow_node_run_job_id = $1"
+	query := "SELECT workflow_node_run_job_id, spawninfos FROM workflow_node_run_job_info WHERE workflow_node_run_job_id = $1"
 	if _, err := db.Select(&res, query, jobID); err != nil {
 		if err == sql.ErrNoRows {
 			return nil, nil
@@ -27,12 +28,19 @@ func loadNodeRunJobInfo(db gorp.SqlExecutor, jobID int64) ([]sdk.SpawnInfo, erro
 	}
 
 	spawnInfos := []sdk.SpawnInfo{}
-	for _, r := range res {
-		v := []sdk.SpawnInfo{}
-		gorpmapping.JSONNullString(r.Bytes, &v)
-		spawnInfos = append(spawnInfos, v...)
+	for i := range res {
+		spInfos := []sdk.SpawnInfo{}
+		if err := gorpmapping.JSONNullString(res[i].Bytes, &spInfos); err != nil {
+			// should never append, but log error
+			log.Warning("wrong spawnInfos format: res: %v for id: %v err: %v", res[i].Bytes, jobID, err)
+			continue
+		}
+		spawnInfos = append(spawnInfos, spInfos...)
 	}
-
+	// sort here and not in sql, as it's could be a json array in sql value
+	sort.Slice(spawnInfos, func(i, j int) bool {
+		return spawnInfos[i].APITime.Before(spawnInfos[j].APITime)
+	})
 	return spawnInfos, nil
 }
 
@@ -52,6 +60,6 @@ func insertNodeRunJobInfo(db gorp.SqlExecutor, info *sdk.WorkflowNodeJobRunInfo)
 		return fmt.Errorf("insertNodeRunJobInfo> Unable to insert into workflow_node_run_job_info id = %d", info.WorkflowNodeJobRunID)
 	}
 
-	log.Debug("insertNodeRunJobInfo> on node run: %d (%d)", info.ID, info.WorkflowNodeJobRunID)
+	log.Debug("insertNodeRunJobInfo> on node run: %v (%d)", info.SpawnInfos, info.WorkflowNodeJobRunID)
 	return nil
 }

engine/api/workflow/dao_run.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"github.com/go-gorp/gorp"
-
 	"github.com/ovh/cds/engine/api/database/gorpmapping"
 	"github.com/ovh/cds/engine/api/observability"
 	"github.com/ovh/cds/sdk"

engine/api/workflow/execute_node_job_run.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/go-gorp/gorp"
 	"github.com/lib/pq"
-
 	"github.com/ovh/cds/engine/api/application"
 	"github.com/ovh/cds/engine/api/cache"
 	"github.com/ovh/cds/engine/api/environment"
@@ -745,7 +744,7 @@ func RestartWorkflowNodeJob(ctx context.Context, db gorp.SqlExecutor, wNodeJob s
 		return sdk.WrapError(errNR, "RestartWorkflowNodeJob> Cannot load node run")
 	}
 
-	//Synchronise struct but not in db
+	//Synchronize struct but not in db
 	sync, errS := SyncNodeRunRunJob(ctx, db, nodeRun, wNodeJob)
 	if errS != nil {
 		return sdk.WrapError(errS, "RestartWorkflowNodeJob> error on sync nodeJobRun")

engine/api/workflow/execute_node_run.go

@@ -426,6 +426,12 @@ func addJobsToQueue(ctx context.Context, db gorp.SqlExecutor, stage *sdk.Stage,
 				Message:    spawnInfos,
 				RemoteTime: time.Now(),
 			}}
+		} else {
+			wjob.SpawnInfos = []sdk.SpawnInfo{sdk.SpawnInfo{
+				APITime:    time.Now(),
+				Message:    sdk.SpawnMsg{ID: sdk.MsgSpawnInfoJobInQueue.ID},
+				RemoteTime: time.Now(),
+			}}
 		}
 
 		//Insert in database
@@ -436,6 +442,10 @@ func addJobsToQueue(ctx context.Context, db gorp.SqlExecutor, stage *sdk.Stage,
 		}
 		next()
 
+		if err := AddSpawnInfosNodeJobRun(db, wjob.ID, PrepareSpawnInfos(wjob.SpawnInfos)); err != nil {
+			return nil, sdk.WrapError(err, "Cannot save spawn info job %d", wjob.ID)
+		}
+
 		//Put the job run in database
 		stage.RunJobs = append(stage.RunJobs, wjob)
 
@@ -502,7 +512,7 @@ func syncStage(db gorp.SqlExecutor, store cache.Store, stage *sdk.Stage) (bool,
 			if runJobDB.Status == sdk.StatusBuilding.String() || runJobDB.Status == sdk.StatusWaiting.String() {
 				stageEnd = false
 			}
-			spawnInfos, err := loadNodeRunJobInfo(db, runJob.ID)
+			spawnInfos, err := LoadNodeRunJobInfo(db, runJob.ID)
 			if err != nil {
 				return false, sdk.WrapError(err, "unable to load spawn infos for runJob: %d", runJob.ID)
 			}
@@ -837,7 +847,7 @@ func SyncNodeRunRunJob(ctx context.Context, db gorp.SqlExecutor, nodeRun *sdk.Wo
 		for j := range s.RunJobs {
 			runJob := &s.RunJobs[j]
 			if runJob.ID == nodeJobRun.ID {
-				spawnInfos, err := loadNodeRunJobInfo(db, runJob.ID)
+				spawnInfos, err := LoadNodeRunJobInfo(db, runJob.ID)
 				if err != nil {
 					return false, sdk.WrapError(err, "unable to load spawn infos for runJobID: %d", runJob.ID)
 				}

engine/api/workflow_run.go

@@ -1344,6 +1344,28 @@ func (api *API) getWorkflowRunArtifactsHandler() service.Handler {
 	}
 }
 
+func (api *API) getWorkflowNodeRunJobSpawnInfosHandler() service.Handler {
+	return func(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
+		runJobID, errJ := requestVarInt(r, "runJobId")
+		if errJ != nil {
+			return sdk.WrapError(errJ, "getWorkflowNodeRunJobSpawnInfosHandler> runJobId: invalid number")
+		}
+		db := api.mustDB()
+
+		spawnInfos, err := workflow.LoadNodeRunJobInfo(db, runJobID)
+		if err != nil {
+			return sdk.WrapError(err, "cannot load spawn infos for node run job id %d", runJobID)
+		}
+
+		l := r.Header.Get("Accept-Language")
+		for ki, info := range spawnInfos {
+			m := sdk.NewMessage(sdk.Messages[info.Message.ID], info.Message.Args...)
+			spawnInfos[ki].UserMessage = m.String(l)
+		}
+		return service.WriteJSON(w, spawnInfos, http.StatusOK)
+	}
+}
+
 func (api *API) getWorkflowNodeRunJobServiceLogsHandler() service.Handler {
 	return func(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
 		runJobID, errJ := requestVarInt(r, "runJobId")

engine/hatchery/swarm/swarm_util_create.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"regexp"
 	"strings"
+	"time"
 
 	types "github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
@@ -119,12 +120,64 @@ checkImage:
 	}
 
 	if !imageFound {
+		infos := []sdk.SpawnInfo{
+			{
+				RemoteTime: time.Now(),
+				Message: sdk.SpawnMsg{
+					ID:   sdk.MsgSpawnInfoHatcheryStartDockerPull.ID,
+					Args: []interface{}{h.Service().Name, fmt.Sprintf("%d", h.ID()), cArgs.image},
+				},
+			},
+		}
+
+		ctxtJobSendSpawnInfo, cancelJobSendSpawnInfo := context.WithTimeout(ctx, 10*time.Second)
+		if err := h.CDSClient().QueueJobSendSpawnInfo(ctxtJobSendSpawnInfo, spawnArgs.IsWorkflowJob, spawnArgs.JobID, infos); err != nil {
+			next()
+			log.Warning("spawnWorkerForJob> cannot client.QueueJobSendSpawnInfo for job %d: %s", spawnArgs.JobID, err)
+		}
+		cancelJobSendSpawnInfo()
+
 		_, next := observability.Span(ctx, "swarm.dockerClient.pullImage", observability.Tag("image", cArgs.image))
 		if err := h.pullImage(dockerClient, cArgs.image, timeoutPullImage); err != nil {
 			next()
+
+			infosEnd := []sdk.SpawnInfo{
+				{
+					RemoteTime: time.Now(),
+					Message: sdk.SpawnMsg{
+						ID:   sdk.MsgSpawnInfoHatcheryEndDockerPullErr.ID,
+						Args: []interface{}{h.Service().Name, fmt.Sprintf("%d", h.ID()), cArgs.image, err},
+					},
+				},
+			}
+
+			ctxtJobSendSpawnInfoEndDockerPullErr, cancelJobSendSpawnInfoDockerPullErr := context.WithTimeout(ctx, 10*time.Second)
+			if err := h.CDSClient().QueueJobSendSpawnInfo(ctxtJobSendSpawnInfoEndDockerPullErr, spawnArgs.IsWorkflowJob, spawnArgs.JobID, infosEnd); err != nil {
+				next()
+				log.Warning("spawnWorkerForJob> cannot client.QueueJobSendSpawnInfo for job %d: %s", spawnArgs.JobID, err)
+			}
+			cancelJobSendSpawnInfoDockerPullErr()
+
 			return sdk.WrapError(err, "Unable to pull image %s on %s", cArgs.image, dockerClient.name)
 		}
 		next()
+
+		infosEnd := []sdk.SpawnInfo{
+			{
+				RemoteTime: time.Now(),
+				Message: sdk.SpawnMsg{
+					ID:   sdk.MsgSpawnInfoHatcheryEndDockerPull.ID,
+					Args: []interface{}{h.Service().Name, fmt.Sprintf("%d", h.ID()), cArgs.image},
+				},
+			},
+		}
+
+		ctxtJobSendSpawnInfoEndDockerPull, cancelJobSendSpawnInfoDockerPull := context.WithTimeout(ctx, 10*time.Second)
+		if err := h.CDSClient().QueueJobSendSpawnInfo(ctxtJobSendSpawnInfoEndDockerPull, spawnArgs.IsWorkflowJob, spawnArgs.JobID, infosEnd); err != nil {
+			next()
+			log.Warning("spawnWorkerForJob> cannot client.QueueJobSendSpawnInfo for job %d: %s", spawnArgs.JobID, err)
+		}
+		cancelJobSendSpawnInfoDockerPull()
 	}
 
 	_, next = observability.Span(ctx, "swarm.dockerClient.ContainerCreate", observability.Tag(observability.TagWorker, cArgs.name), observability.Tag("network", fmt.Sprintf("%v", networkingConfig)))

sdk/build.go

@@ -24,9 +24,10 @@ type PipelineBuildJob struct {
 
 // SpawnInfo contains an information about spawning
 type SpawnInfo struct {
-	APITime    time.Time `json:"api_time,omitempty" db:"-" mapstructure:"-"`
-	RemoteTime time.Time `json:"remote_time,omitempty" db:"-" mapstructure:"-"`
-	Message    SpawnMsg  `json:"message,omitempty" db:"-"`
+	WorkflowNodeJobRunID int64     `json:"WorkflowNodeJobRunID,omitempty" db:"-"`
+	APITime              time.Time `json:"api_time,omitempty" db:"-" mapstructure:"-"`
+	RemoteTime           time.Time `json:"remote_time,omitempty" db:"-" mapstructure:"-"`
+	Message              SpawnMsg  `json:"message,omitempty" db:"-"`
 	// UserMessage contains msg translated for end user
 	UserMessage string `json:"user_message,omitempty" db:"-"`
 }

sdk/messages.go

@@ -59,25 +59,29 @@ var (
 	MsgPipelineJobUpdated                  = &Message{"MsgPipelineJobUpdated", trad{FR: "Le job %s du stage %s a été mis à jour", EN: "Job %s in stage %s updated"}, nil}
 	MsgPipelineJobAdded                    = &Message{"MsgPipelineJobAdded", trad{FR: "Le job %s du stage %s a été ajouté", EN: "Job %s in stage %s added"}, nil}
 	MsgPipelineJobDeleted                  = &Message{"MsgPipelineJobDeleted", trad{FR: "Le job %s du stage %s a été supprimé", EN: "Job %s in stage %s deleted"}, nil}
+	MsgSpawnInfoDeprecatedModel            = &Message{"MsgSpawnInfoDeprecatedModel", trad{FR: "Attention vous utilisez un worker model (%s) déprécié", EN: "Pay attention you are using a deprecated worker model (%s)"}, nil}
 	MsgSpawnInfoHatcheryStarts             = &Message{"MsgSpawnInfoHatcheryStarts", trad{FR: "La Hatchery %s (%s) a démarré le lancement du worker avec le modèle %s", EN: "Hatchery %s (%s) starts spawn worker with model %s"}, nil}
+	MsgSpawnInfoHatcheryStartDockerPull    = &Message{"MsgSpawnInfoHatcheryStartDockerPull", trad{FR: "La Hatchery %s (%s) a démarré le docker pull de l'image %s...", EN: "Hatchery %s (%s) starts docker pull %s..."}, nil}
+	MsgSpawnInfoHatcheryEndDockerPull      = &Message{"MsgSpawnInfoHatcheryEndDockerPull", trad{FR: "La Hatchery %s (%s) a terminé le docker pull de l'image %s", EN: "Hatchery %s (%s) docker pull %s done"}, nil}
+	MsgSpawnInfoHatcheryEndDockerPullErr   = &Message{"MsgSpawnInfoHatcheryEndDockerPullErr", trad{FR: "⚠ La Hatchery %s (%s) a terminé le docker pull de l'image %s en erreur: %s", EN: "⚠ Hatchery %s (%s) - docker pull %s done with error: %v"}, nil}
 	MsgSpawnInfoHatcheryErrorSpawn         = &Message{"MsgSpawnInfoHatcheryErrorSpawn", trad{FR: "Une erreur est survenue lorsque la Hatchery %s (%s) a démarré un worker avec le modèle %s après %s, err:%s", EN: "Error while Hatchery %s (%s) spawn worker with model %s after %s, err:%s"}, nil}
 	MsgSpawnInfoHatcheryStartsSuccessfully = &Message{"MsgSpawnInfoHatcheryStartsSuccessfully", trad{FR: "La Hatchery %s (%s) a démarré le worker %s avec succès en %s", EN: "Hatchery %s (%s) spawn worker %s successfully in %s"}, nil}
 	MsgSpawnInfoWorkerEnd                  = &Message{"MsgSpawnInfoWorkerEnd", trad{FR: "Le worker %s a terminé et a passé %s à travailler sur les étapes", EN: "Worker %s finished working on this job and took %s to work on the steps"}, nil}
+	MsgSpawnInfoJobInQueue                 = &Message{"MsgSpawnInfoJobInQueue", trad{FR: "✓ Le job a été pris mis en file d'attente", EN: "✓ Job was queued"}, nil}
 	MsgSpawnInfoJobTaken                   = &Message{"MsgSpawnInfoJobTaken", trad{FR: "Le job %s a été pris par le worker %s", EN: "Job %s was taken by worker %s"}, nil}
 	MsgSpawnInfoJobTakenWorkerVersion      = &Message{"MsgSpawnInfoJobTakenWorkerVersion", trad{FR: "Worker %s version:%s os:%s arch:%s", EN: "Worker %s version:%s os:%s arch:%s"}, nil}
 	MsgSpawnInfoWorkerForJob               = &Message{"MsgSpawnInfoWorkerForJob", trad{FR: "Ce worker %s a été créé pour lancer ce job", EN: "This worker %s was created to take this action"}, nil}
-	MsgSpawnInfoWorkerForJobError          = &Message{"MsgSpawnInfoWorkerForJobError", trad{FR: "Ce worker %s a été créé pour lancer ce job, mais ne possède pas tous les pré-requis. Vérifiez que les prérequis suivants:%s", EN: "This worker %s was created to take this action, but does not have all prerequisites. Please verify the following prerequisites:%s"}, nil}
-	MsgSpawnInfoJobError                   = &Message{"MsgSpawnInfoJobError", trad{FR: "Impossible de lancer ce job : %s", EN: "Unable to run this job: %s"}, nil}
+	MsgSpawnInfoWorkerForJobError          = &Message{"MsgSpawnInfoWorkerForJobError", trad{FR: "⚠ Ce worker %s a été créé pour lancer ce job, mais ne possède pas tous les pré-requis. Vérifiez que les prérequis suivants:%s", EN: "⚠ This worker %s was created to take this action, but does not have all prerequisites. Please verify the following prerequisites:%s"}, nil}
+	MsgSpawnInfoJobError                   = &Message{"MsgSpawnInfoJobError", trad{FR: "⚠ Impossible de lancer ce job : %s", EN: "⚠ Unable to run this job: %s"}, nil}
 	MsgWorkflowStarting                    = &Message{"MsgWorkflowStarting", trad{FR: "Le workflow %s#%s a été démarré", EN: "Workflow %s#%s has been started"}, nil}
-	MsgWorkflowError                       = &Message{"MsgWorkflowError", trad{FR: "Une erreur est survenue: %v", EN: "An error has occured: %v"}, nil}
+	MsgWorkflowError                       = &Message{"MsgWorkflowError", trad{FR: "⚠ Une erreur est survenue: %v", EN: "⚠ An error has occured: %v"}, nil}
 	MsgWorkflowNodeStop                    = &Message{"MsgWorkflowNodeStop", trad{FR: "Le pipeline a été arrété par %s", EN: "The pipeline has been stopped by %s"}, nil}
 	MsgWorkflowNodeMutex                   = &Message{"MsgWorkflowNodeMutex", trad{FR: "Le pipeline %s est mis en attente tant qu'il est en cours sur un autre run", EN: "The pipeline %s is waiting while it's running on another run"}, nil}
 	MsgWorkflowNodeMutexRelease            = &Message{"MsgWorkflowNodeMutexRelease", trad{FR: "Lancement du pipeline %s", EN: "Triggering pipeline %s"}, nil}
 	MsgWorkflowImportedUpdated             = &Message{"MsgWorkflowImportedUpdated", trad{FR: "Le workflow %s a été mis à jour", EN: "Workflow %s has been updated"}, nil}
 	MsgWorkflowImportedInserted            = &Message{"MsgWorkflowImportedInserted", trad{FR: "Le workflow %s a été créé", EN: "Workflow %s has been created"}, nil}
 	MsgSpawnInfoHatcheryCannotStartJob     = &Message{"MsgSpawnInfoHatcheryCannotStart", trad{FR: "Aucune hatchery n'a pu démarrer de worker respectant vos pré-requis de job, merci de les vérifier.", EN: "No hatchery can spawn a worker corresponding your job's requirements. Please check your job's requirements."}, nil}
 	MsgWorkflowRunBranchDeleted            = &Message{"MsgWorkflowRunBranchDeleted", trad{FR: "La branche %s  a été supprimée", EN: "Branch %s has been deleted"}, nil}
-	MsgSpawnInfoDeprecatedModel            = &Message{"MsgSpawnInfoDeprecatedModel", trad{FR: "Attention vous utilisez un worker model (%s) déprécié", EN: "Pay attention you are using a deprecated worker model (%s)"}, nil}
 )
 
 // Messages contains all sdk Messages
@@ -124,7 +128,11 @@ var Messages = map[string]*Message{
 	MsgSpawnInfoHatcheryStarts.ID:             MsgSpawnInfoHatcheryStarts,
 	MsgSpawnInfoHatcheryErrorSpawn.ID:         MsgSpawnInfoHatcheryErrorSpawn,
 	MsgSpawnInfoHatcheryStartsSuccessfully.ID: MsgSpawnInfoHatcheryStartsSuccessfully,
+	MsgSpawnInfoHatcheryStartDockerPull.ID:    MsgSpawnInfoHatcheryStartDockerPull,
+	MsgSpawnInfoHatcheryEndDockerPull.ID:      MsgSpawnInfoHatcheryEndDockerPull,
+	MsgSpawnInfoHatcheryEndDockerPullErr.ID:   MsgSpawnInfoHatcheryEndDockerPullErr,
 	MsgSpawnInfoWorkerEnd.ID:                  MsgSpawnInfoWorkerEnd,
+	MsgSpawnInfoJobInQueue.ID:                 MsgSpawnInfoJobInQueue,
 	MsgSpawnInfoJobTaken.ID:                   MsgSpawnInfoJobTaken,
 	MsgSpawnInfoJobTakenWorkerVersion.ID:      MsgSpawnInfoJobTakenWorkerVersion,
 	MsgSpawnInfoWorkerForJob.ID:               MsgSpawnInfoWorkerForJob,

ui/src/app/views/workflow/run/node/pipeline/pipeline.html

@@ -52,15 +52,18 @@
         </div>
         <div class="row">
             <div class="column">
-                <div class="log animated fadeIn" *ngIf="selectedRunJob && mapStepStatus">
+                <div class="log animated fadeIn" *ngIf="selectedRunJob && mapStepStatus && mapJobStatus">
                     <ul>
                         <li>
-                            <app-workflow-rin-job-spawn-info
-                              [spawnInfos]="selectedRunJob.spawninfos"
-                              [variables]="selectedRunJob.parameters"
-                              [job]="selectedRunJob.job"
-                              [(displayServicesLogs)]="displayServiceLogs">
-                            </app-workflow-rin-job-spawn-info>
+                            <app-workflow-run-job-spawn-info
+                                [project]="project"
+                                [workflowName]="workflowName"
+                                [nodeRun]="nodeRun"
+                                [jobStatus]="mapJobStatus.get(selectedRunJob.job.pipeline_action_id).status"
+                                [nodeJobRun]="selectedRunJob"
+                                [variables]="selectedRunJob.parameters"
+                                [(displayServicesLogs)]="displayServiceLogs">
+                            </app-workflow-run-job-spawn-info>
                         </li>
                         <ng-container *ngIf="!displayServiceLogs">
                             <li *ngFor="let step of selectedRunJob.job.action.actions; let i = index">