Skip to content

Commit 11d98eb

Browse files
fsaminfsamin
authored
feat(api): add hatchery and worker details in job run (#5101)
Signed-off-by: francois samin <[email protected]>
1 parent 5386f91 commit 11d98eb

File tree

9 files changed

+88
-14
lines changed

9 files changed

+88
-14
lines changed

‎engine/api/services/dao.go‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,11 +4,9 @@ import (
44
"context"
55
"time"
66

7-
"github.com/ovh/cds/engine/api/worker"
8-
97
"github.com/go-gorp/gorp"
10-
118
"github.com/ovh/cds/engine/api/database/gorpmapping"
9+
"github.com/ovh/cds/engine/api/worker"
1210
"github.com/ovh/cds/sdk"
1311
"github.com/ovh/cds/sdk/log"
1412
)

‎engine/api/workflow/execute_node_job_run.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -129,8 +129,6 @@ func UpdateNodeJobRunStatus(ctx context.Context, db gorp.SqlExecutor, store cach
129129

130130
report := new(ProcessorReport)
131131

132-
log.Debug("UpdateNodeJobRunStatus> job.ID=%d status=%s", job.ID, status)
133-
134132
_, next := observability.Span(ctx, "workflow.LoadRunByID")
135133
nodeRun, errLoad := LoadNodeRunByID(db, job.WorkflowNodeRunID, LoadRunOptions{})
136134
next()
@@ -238,7 +236,7 @@ func PrepareSpawnInfos(infos []sdk.SpawnInfo) []sdk.SpawnInfo {
238236

239237
// TakeNodeJobRun Take an a job run for update
240238
func TakeNodeJobRun(ctx context.Context, db gorp.SqlExecutor, store cache.Store, proj sdk.Project, jobID int64,
241-
workerModel, workerName, workerID string, infos []sdk.SpawnInfo) (*sdk.WorkflowNodeJobRun, *ProcessorReport, error) {
239+
workerModel, workerName, workerID string, infos []sdk.SpawnInfo, hatcheryName string) (*sdk.WorkflowNodeJobRun, *ProcessorReport, error) {
242240
var end func()
243241
ctx, end = observability.Span(ctx, "workflow.TakeNodeJobRun")
244242
defer end()
@@ -267,6 +265,8 @@ func TakeNodeJobRun(ctx context.Context, db gorp.SqlExecutor, store cache.Store,
267265
return nil, report, err
268266
}
269267

268+
job.HatcheryName = hatcheryName
269+
job.WorkerName = workerName
270270
job.Model = workerModel
271271
job.Job.WorkerName = workerName
272272
job.Job.WorkerID = workerID

‎engine/api/workflow/execute_node_run.go‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,6 @@ func syncTakeJobInNodeRun(ctx context.Context, db gorp.SqlExecutor, n *sdk.Workf
4444
_, end := observability.Span(ctx, "workflow.syncTakeJobInNodeRun")
4545
defer end()
4646

47-
log.Debug("workflow.syncTakeJobInNodeRun> job parameters= %+v", j.Parameters)
48-
4947
report := new(ProcessorReport)
5048

5149
//If status is not waiting neither build: nothing to do
@@ -70,6 +68,8 @@ func syncTakeJobInNodeRun(ctx context.Context, db gorp.SqlExecutor, n *sdk.Workf
7068
rj.Model = j.Model
7169
rj.ModelType = j.ModelType
7270
rj.ContainsService = j.ContainsService
71+
rj.WorkerName = j.WorkerName
72+
rj.HatcheryName = j.HatcheryName
7373
rj.Job = j.Job
7474
rj.Header = j.Header
7575
rj.Parameters = j.Parameters
@@ -635,6 +635,9 @@ func syncStage(ctx context.Context, db gorp.SqlExecutor, store cache.Store, stag
635635
runJob.ModelType = runJobDB.ModelType
636636
runJob.ContainsService = runJobDB.ContainsService
637637
runJob.Job = runJobDB.Job
638+
runJob.Model = runJobDB.Model
639+
runJob.WorkerName = runJobDB.WorkerName
640+
runJob.HatcheryName = runJobDB.HatcheryName
638641
}
639642
}
640643
}

‎engine/api/workflow/gorp_model.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"time"
66

77
"github.com/ovh/cds/engine/api/database/gorpmapping"
8+
"github.com/ovh/cds/sdk/log"
89

910
"github.com/ovh/cds/sdk"
1011
)
@@ -79,6 +80,8 @@ type JobRun struct {
7980
ContainsService bool `db:"contains_service"`
8081
ModelType sql.NullString `db:"model_type"`
8182
Header sql.NullString `db:"header"`
83+
HatcheryName string `db:"hatchery_name"`
84+
WorkerName string `db:"worker_name"`
8285
}
8386

8487
// ToJobRun transform the JobRun with data of the provided sdk.WorkflowNodeJobRun
@@ -103,6 +106,9 @@ func (j *JobRun) ToJobRun(jr *sdk.WorkflowNodeJobRun) (err error) {
103106
j.ModelType = sql.NullString{Valid: true, String: string(jr.ModelType)}
104107
j.ContainsService = jr.ContainsService
105108
j.ExecGroups, err = gorpmapping.JSONToNullString(jr.ExecGroups)
109+
j.WorkerName = jr.WorkerName
110+
j.HatcheryName = jr.HatcheryName
111+
log.Debug("ToJobRun> %+v", j)
106112
if err != nil {
107113
return sdk.WrapError(err, "column exec_groups")
108114
}
@@ -131,6 +137,9 @@ func (j JobRun) WorkflowNodeRunJob() (sdk.WorkflowNodeJobRun, error) {
131137
Done: j.Done,
132138
BookedBy: j.BookedBy,
133139
ContainsService: j.ContainsService,
140+
HatcheryName: j.HatcheryName,
141+
WorkerName: j.WorkerName,
142+
Model: j.Model,
134143
}
135144
if err := gorpmapping.JSONNullString(j.Job, &jr.Job); err != nil {
136145
return jr, sdk.WrapError(err, "column job")

‎engine/api/workflow/run_workflow_test.go‎

Lines changed: 33 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -605,15 +605,16 @@ queueRun:
605605
}
606606

607607
//TakeNodeJobRun
608-
takenJob, _, _ := workflow.TakeNodeJobRun(context.TODO(), db, cache, *proj, j.ID, "model", "worker", "1", []sdk.SpawnInfo{
608+
takenJobID := j.ID
609+
takenJob, _, _ := workflow.TakeNodeJobRun(context.TODO(), db, cache, *proj, takenJobID, "model", "worker", "1", []sdk.SpawnInfo{
609610
{
610611
APITime: time.Now(),
611612
RemoteTime: time.Now(),
612613
Message: sdk.SpawnMsg{
613614
ID: sdk.MsgSpawnInfoJobTaken.ID,
614615
},
615616
},
616-
})
617+
}, "hatchery_name")
617618

618619
//Load workflow node run
619620
nodeRun, err := workflow.LoadNodeRunByID(db, takenJob.WorkflowNodeRunID, workflow.LoadRunOptions{})
@@ -647,6 +648,12 @@ queueRun:
647648
t.FailNow()
648649
}
649650

651+
j, err = workflow.LoadNodeJobRun(context.TODO(), db, cache, j.ID)
652+
require.NoError(t, err)
653+
assert.Equal(t, "hatchery_name", j.HatcheryName)
654+
assert.NotEmpty(t, j.WorkerName)
655+
assert.NotEmpty(t, j.Model)
656+
650657
//TestUpdateNodeJobRunStatus
651658
_, err = workflow.UpdateNodeJobRunStatus(context.TODO(), db, cache, *proj, j, sdk.StatusSuccess)
652659
assert.NoError(t, err)
@@ -655,6 +662,30 @@ queueRun:
655662
t.FailNow()
656663
}
657664

665+
workflowRun, err = workflow.LoadRunByID(db, wr.ID, workflow.LoadRunOptions{})
666+
require.NoError(t, err)
667+
var jobRunFound bool
668+
checkJobRun:
669+
for _, noderuns := range workflowRun.WorkflowNodeRuns {
670+
for _, noderun := range noderuns {
671+
for _, stage := range noderun.Stages {
672+
for _, jobrun := range stage.RunJobs {
673+
t.Logf("checking job %d with %d", jobrun.ID, takenJobID)
674+
if jobrun.ID == j.ID {
675+
assert.Equal(t, "hatchery_name", jobrun.HatcheryName)
676+
assert.NotEmpty(t, jobrun.WorkerName)
677+
assert.NotEmpty(t, jobrun.Model)
678+
jobRunFound = true
679+
break checkJobRun
680+
}
681+
}
682+
}
683+
}
684+
}
685+
if !jobRunFound {
686+
t.Fatalf("unable to retrieve job run in the workflow run")
687+
}
688+
658689
logs, err := workflow.LoadLogs(db, takenJob.ID)
659690
assert.NoError(t, err)
660691
if t.Failed() {

‎engine/api/workflow_queue.go‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"github.com/ovh/venom"
1212
"github.com/sguiheux/go-coverage"
1313

14+
"github.com/ovh/cds/engine/api/authentication"
1415
"github.com/ovh/cds/engine/api/cache"
1516
"github.com/ovh/cds/engine/api/event"
1617
"github.com/ovh/cds/engine/api/group"
@@ -39,6 +40,21 @@ func (api *API) postTakeWorkflowJobHandler() service.Handler {
3940
return sdk.WithStack(sdk.ErrForbidden)
4041
}
4142

43+
consumer := getAPIConsumer(ctx)
44+
// Locking for the parent consumer
45+
var hatcheryName string
46+
if consumer.ParentID != nil {
47+
parentConsumer, err := authentication.LoadConsumerByID(ctx, api.mustDB(), *consumer.ParentID)
48+
if err != nil {
49+
return err
50+
}
51+
s, err := services.LoadByConsumerID(ctx, api.mustDB(), parentConsumer.ID)
52+
if err != nil {
53+
return err
54+
}
55+
hatcheryName = s.Name
56+
}
57+
4258
wk, err := worker.LoadByID(ctx, api.mustDB(), getAPIConsumer(ctx).Worker.ID)
4359
if err != nil {
4460
return err
@@ -77,7 +93,7 @@ func (api *API) postTakeWorkflowJobHandler() service.Handler {
7793
}
7894

7995
pbji := &sdk.WorkflowNodeJobRunData{}
80-
report, err := takeJob(ctx, api.mustDB, api.Cache, p, id, workerModelName, pbji, wk)
96+
report, err := takeJob(ctx, api.mustDB, api.Cache, p, id, workerModelName, pbji, wk, hatcheryName)
8197
if err != nil {
8298
return sdk.WrapError(err, "cannot takeJob nodeJobRunID:%d", id)
8399
}
@@ -89,7 +105,7 @@ func (api *API) postTakeWorkflowJobHandler() service.Handler {
89105
}
90106
}
91107

92-
func takeJob(ctx context.Context, dbFunc func() *gorp.DbMap, store cache.Store, p *sdk.Project, id int64, workerModel string, wnjri *sdk.WorkflowNodeJobRunData, wk *sdk.Worker) (*workflow.ProcessorReport, error) {
108+
func takeJob(ctx context.Context, dbFunc func() *gorp.DbMap, store cache.Store, p *sdk.Project, id int64, workerModel string, wnjri *sdk.WorkflowNodeJobRunData, wk *sdk.Worker, hatcheryName string) (*workflow.ProcessorReport, error) {
93109
// Start a tx
94110
tx, errBegin := dbFunc().Begin()
95111
if errBegin != nil {
@@ -110,7 +126,7 @@ func takeJob(ctx context.Context, dbFunc func() *gorp.DbMap, store cache.Store,
110126
}
111127

112128
// Take node job run
113-
job, report, err := workflow.TakeNodeJobRun(ctx, tx, store, *p, id, workerModel, wk.Name, wk.ID, infos)
129+
job, report, err := workflow.TakeNodeJobRun(ctx, tx, store, *p, id, workerModel, wk.Name, wk.ID, infos, hatcheryName)
114130
if err != nil {
115131
return nil, sdk.WrapError(err, "cannot take job %d", id)
116132
}

‎engine/api/workflow_queue_test.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ type testRunWorkflowCtx struct {
5151
workerToken string
5252
hatchery *sdk.Service
5353
hatcheryToken string
54+
model *sdk.Model
5455
}
5556

5657
func testRunWorkflow(t *testing.T, api *API, router *Router) testRunWorkflowCtx {
@@ -329,6 +330,7 @@ func testRegisterWorker(t *testing.T, api *API, router *Router, ctx *testRunWork
329330
w, workerJWT := RegisterWorker(t, api, g.ID, model.Name)
330331
ctx.workerToken = workerJWT
331332
ctx.worker = w
333+
ctx.model = model
332334
}
333335

334336
func testRegisterHatchery(t *testing.T, api *API, router *Router, ctx *testRunWorkflowCtx) {
@@ -451,7 +453,10 @@ func Test_postTakeWorkflowJobHandler(t *testing.T) {
451453

452454
run, err := workflow.LoadNodeJobRun(context.TODO(), api.mustDB(), api.Cache, ctx.job.ID)
453455
require.NoError(t, err)
454-
require.Equal(t, "Building", run.Status)
456+
assert.Equal(t, "Building", run.Status)
457+
assert.Equal(t, ctx.model.Name, run.Model)
458+
assert.Equal(t, ctx.worker.Name, run.WorkerName)
459+
assert.NotEmpty(t, run.HatcheryName)
455460
}
456461

457462
func Test_postBookWorkflowJobHandler(t *testing.T) {

‎engine/sql/194_workflow_node_job_run_infos.sql‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
-- +migrate Up
2+
ALTER TABLE workflow_node_run_job ADD COLUMN hatchery_name TEXT;
3+
ALTER TABLE workflow_node_run_job ADD COLUMN worker_name TEXT;
4+
5+
-- +migrate Down
6+
ALTER TABLE workflow_node_run_job DROP COLUMN hatchery_name;
7+
ALTER TABLE workflow_node_run_job DROP COLUMN worker_name;

‎sdk/workflow_run.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -372,6 +372,8 @@ type WorkflowNodeJobRun struct {
372372
IntegrationPluginBinaries []GRPCPluginBinary `json:"integration_plugin_binaries,omitempty"`
373373
Header WorkflowRunHeaders `json:"header,omitempty"`
374374
ContainsService bool `json:"contains_service,omitempty"`
375+
HatcheryName string `json:"hatchery_name,omitempty"`
376+
WorkerName string `json:"worker_name,omitempty"`
375377
}
376378

377379
// WorkflowNodeJobRunSummary is a light representation of WorkflowNodeJobRun for CDS event
@@ -384,6 +386,9 @@ type WorkflowNodeJobRunSummary struct {
384386
Done int64 `json:"done,omitempty"`
385387
Job ExecutedJobSummary `json:"job_summary,omitempty"`
386388
SpawnInfos []SpawnInfo `json:"spawninfos"`
389+
HatcheryName string `json:"hatchery_name,omitempty"`
390+
WorkerName string `json:"worker_name,omitempty"`
391+
WorkerModelName string `json:"worker_model_name,omitempty"`
387392
}
388393

389394
// ToSummary transforms a WorkflowNodeJobRun into a WorkflowNodeJobRunSummary

0 commit comments

Comments
 (0)