Skip to content

Commit c65850b

Browse files
kylecarbskylecarbs
authored
fix: Simplify provisionerd job acquire (#158)
This uses a simple channel to detect whether a job is running or not, and moves all cancels to be in goroutines.
1 parent 7884b43 commit c65850b

File tree

11 files changed

+182
-127
lines changed

11 files changed

+182
-127
lines changed

‎coderd/coderd.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ type Options struct {
2323
func New(options *Options) http.Handler {
2424
api := &api{
2525
Database: options.Database,
26+
Logger: options.Logger,
2627
Pubsub: options.Pubsub,
2728
}
2829

@@ -110,5 +111,6 @@ func New(options *Options) http.Handler {
110111
// be added to this struct for code clarity.
111112
type api struct {
112113
Database database.Store
114+
Logger slog.Logger
113115
Pubsub database.Pubsub
114116
}

‎coderd/coderdtest/coderdtest.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -125,7 +125,7 @@ func New(t *testing.T) Server {
125125
}
126126

127127
handler := coderd.New(&coderd.Options{
128-
Logger: slogtest.Make(t, nil),
128+
Logger: slogtest.Make(t, nil).Leveled(slog.LevelDebug),
129129
Database: db,
130130
Pubsub: pubsub,
131131
})

‎coderd/provisionerdaemons.go

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ import (
1919
"storj.io/drpc/drpcmux"
2020
"storj.io/drpc/drpcserver"
2121

22+
"cdr.dev/slog"
23+
2224
"github.com/coder/coder/coderd/projectparameter"
2325
"github.com/coder/coder/database"
2426
"github.com/coder/coder/httpapi"
@@ -84,6 +86,7 @@ func (api *api) provisionerDaemonsServe(rw http.ResponseWriter, r *http.Request)
8486
Database: api.Database,
8587
Pubsub: api.Pubsub,
8688
Provisioners: daemon.Provisioners,
89+
Logger: api.Logger.Named(fmt.Sprintf("provisionerd-%s", daemon.Name)),
8790
})
8891
if err != nil {
8992
_ = conn.Close(websocket.StatusInternalError, fmt.Sprintf("drpc register provisioner daemon: %s", err))
@@ -109,6 +112,7 @@ type projectImportJob struct {
109112
// Implementation of the provisioner daemon protobuf server.
110113
type provisionerdServer struct {
111114
ID uuid.UUID
115+
Logger slog.Logger
112116
Provisioners []database.ProvisionerType
113117
Database database.Store
114118
Pubsub database.Pubsub
@@ -136,9 +140,11 @@ func (server *provisionerdServer) AcquireJob(ctx context.Context, _ *proto.Empty
136140
if err != nil {
137141
return nil, xerrors.Errorf("acquire job: %w", err)
138142
}
143+
server.Logger.Debug(ctx, "locked job from database", slog.F("id", job.ID))
144+
139145
// Marks the acquired job as failed with the error message provided.
140146
failJob := func(errorMessage string) error {
141-
err = server.Database.UpdateProvisionerJobByID(ctx, database.UpdateProvisionerJobByIDParams{
147+
err = server.Database.UpdateProvisionerJobWithCompleteByID(ctx, database.UpdateProvisionerJobWithCompleteByIDParams{
142148
ID: job.ID,
143149
CompletedAt: sql.NullTime{
144150
Time: database.Now(),
@@ -381,8 +387,12 @@ func (server *provisionerdServer) CancelJob(ctx context.Context, cancelJob *prot
381387
if err != nil {
382388
return nil, xerrors.Errorf("parse job id: %w", err)
383389
}
384-
err = server.Database.UpdateProvisionerJobByID(ctx, database.UpdateProvisionerJobByIDParams{
390+
err = server.Database.UpdateProvisionerJobWithCompleteByID(ctx, database.UpdateProvisionerJobWithCompleteByIDParams{
385391
ID: jobID,
392+
CompletedAt: sql.NullTime{
393+
Time: database.Now(),
394+
Valid: true,
395+
},
386396
CancelledAt: sql.NullTime{
387397
Time: database.Now(),
388398
Valid: true,
@@ -476,7 +486,7 @@ func (server *provisionerdServer) CompleteJob(ctx context.Context, completed *pr
476486

477487
// This must occur in a transaction in case of failure.
478488
err = server.Database.InTx(func(db database.Store) error {
479-
err = db.UpdateProvisionerJobByID(ctx, database.UpdateProvisionerJobByIDParams{
489+
err = db.UpdateProvisionerJobWithCompleteByID(ctx, database.UpdateProvisionerJobWithCompleteByIDParams{
480490
ID: jobID,
481491
UpdatedAt: database.Now(),
482492
CompletedAt: sql.NullTime{
@@ -495,6 +505,7 @@ func (server *provisionerdServer) CompleteJob(ctx context.Context, completed *pr
495505
return xerrors.Errorf("insert project parameter %q: %w", projectParameter.Name, err)
496506
}
497507
}
508+
server.Logger.Debug(ctx, "marked import job as completed", slog.F("job_id", jobID))
498509
return nil
499510
})
500511
if err != nil {
@@ -513,7 +524,7 @@ func (server *provisionerdServer) CompleteJob(ctx context.Context, completed *pr
513524
}
514525

515526
err = server.Database.InTx(func(db database.Store) error {
516-
err = db.UpdateProvisionerJobByID(ctx, database.UpdateProvisionerJobByIDParams{
527+
err = db.UpdateProvisionerJobWithCompleteByID(ctx, database.UpdateProvisionerJobWithCompleteByIDParams{
517528
ID: jobID,
518529
UpdatedAt: database.Now(),
519530
CompletedAt: sql.NullTime{

‎coderd/provisioners.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ type ProvisionerJobStatus string
1212

1313
// Completed returns whether the job is still processing.
1414
func (p ProvisionerJobStatus) Completed() bool {
15-
return p == ProvisionerJobStatusSucceeded || p == ProvisionerJobStatusFailed
15+
return p == ProvisionerJobStatusSucceeded || p == ProvisionerJobStatusFailed || p == ProvisionerJobStatusCancelled
1616
}
1717

1818
const (

‎coderd/workspacehistory.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,10 @@ func (api *api) postWorkspaceHistoryByUser(rw http.ResponseWriter, r *http.Reque
8282
Message: fmt.Sprintf("The provided project history %q has failed to import. You cannot create workspaces using it!", projectHistory.Name),
8383
})
8484
return
85+
case ProvisionerJobStatusCancelled:
86+
httpapi.Write(rw, http.StatusPreconditionFailed, httpapi.Response{
87+
Message: "The provided project history was canceled during import. You cannot create workspaces using it!",
88+
})
8589
}
8690

8791
project, err := api.Database.GetProjectByID(r.Context(), projectHistory.ProjectID)

‎coderd/workspacehistory_test.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ func TestWorkspaceHistory(t *testing.T) {
5656
require.Eventually(t, func() bool {
5757
hist, err := client.ProjectHistory(context.Background(), user.Organization, project.Name, projectHistory.Name)
5858
require.NoError(t, err)
59+
t.Logf("Import status: %s\n", hist.Import.Status)
5960
return hist.Import.Status.Completed()
6061
}, 15*time.Second, 50*time.Millisecond)
6162
return projectHistory

‎database/databasefake/databasefake.go

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -904,9 +904,24 @@ func (q *fakeQuerier) UpdateProvisionerJobByID(_ context.Context, arg database.U
904904
if arg.ID.String() != job.ID.String() {
905905
continue
906906
}
907+
job.UpdatedAt = arg.UpdatedAt
908+
q.provisionerJobs[index] = job
909+
return nil
910+
}
911+
return sql.ErrNoRows
912+
}
913+
914+
func (q *fakeQuerier) UpdateProvisionerJobWithCompleteByID(_ context.Context, arg database.UpdateProvisionerJobWithCompleteByIDParams) error {
915+
q.mutex.Lock()
916+
defer q.mutex.Unlock()
917+
918+
for index, job := range q.provisionerJobs {
919+
if arg.ID.String() != job.ID.String() {
920+
continue
921+
}
922+
job.UpdatedAt = arg.UpdatedAt
907923
job.CompletedAt = arg.CompletedAt
908924
job.CancelledAt = arg.CancelledAt
909-
job.UpdatedAt = arg.UpdatedAt
910925
job.Error = arg.Error
911926
q.provisionerJobs[index] = job
912927
return nil

‎database/querier.go

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎database/query.sql

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -604,12 +604,20 @@ WHERE
604604
id = $1;
605605

606606
-- name: UpdateProvisionerJobByID :exec
607+
UPDATE
608+
provisioner_job
609+
SET
610+
updated_at = $2
611+
WHERE
612+
id = $1;
613+
614+
-- name: UpdateProvisionerJobWithCompleteByID :exec
607615
UPDATE
608616
provisioner_job
609617
SET
610618
updated_at = $2,
611-
cancelled_at = $3,
612-
completed_at = $4,
619+
completed_at = $3,
620+
cancelled_at = $4,
613621
error = $5
614622
WHERE
615623
id = $1;

‎database/query.sql.go

Lines changed: 26 additions & 7 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)