Skip to content

feat: cancel stuck pending jobs #17803

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 24 commits into from
May 20, 2025
Merged

feat: cancel stuck pending jobs #17803

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
0f51f35
added queries for fetching NotStartedProvisionerJobs
ibetitsmike Apr 1, 2025
2f3d606
added detector handling of not started jobs
ibetitsmike May 9, 2025
4b252eb
filling out started_at when unhanging not started jobs
ibetitsmike May 9, 2025
ca49519
WIP
ibetitsmike May 13, 2025
af994c2
refactored to reaper & added tests
ibetitsmike May 13, 2025
3815727
Revert "filling out started_at when unhanging not started jobs"
ibetitsmike May 13, 2025
b65f620
created new ORM update to avoid forcing setting StartedAt on every Co…
ibetitsmike May 13, 2025
3c7c323
added missing dbauthz tests
ibetitsmike May 13, 2025
35df01f
added checks for StartedAt value in the updated jobs
ibetitsmike May 13, 2025
8aa1ee2
refactor from reaper to jobreaper
ibetitsmike May 14, 2025
4385933
WIP
ibetitsmike May 14, 2025
96fee51
WIP
ibetitsmike May 14, 2025
d8db119
WIP
ibetitsmike May 15, 2025
5120fb1
WIP
ibetitsmike May 15, 2025
8d4fa5a
fixed sql comments
ibetitsmike May 15, 2025
18b809c
taking a step back with RBAC
ibetitsmike May 16, 2025
0fe1404
WIP
ibetitsmike May 16, 2025
77be34e
WIP
ibetitsmike May 16, 2025
4351529
WIP
ibetitsmike May 16, 2025
c03bfa3
fixed InOrg check for provisionerjob resource
ibetitsmike May 19, 2025
a15bd1c
PR review; naming in the comments, added comments for SQL, less verbo…
ibetitsmike May 19, 2025
5b9348f
fixes to tests after lint remove rand
ibetitsmike May 19, 2025
91d2d32
readded rand to fix gen failing in CI
ibetitsmike May 19, 2025
767cb77
adjusted TODOs
ibetitsmike May 20, 2025
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
Prev Previous commit
Next Next commit
refactor from reaper to jobreaper
  • Loading branch information
@ibetitsmike
ibetitsmike committed May 14, 2025
commit 8aa1ee247b5eeeb0613af6d114bf9c37c304579f
4 changes: 2 additions & 2 deletions cli/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,12 +87,12 @@ import (
"github.com/coder/coder/v2/coderd/externalauth"
"github.com/coder/coder/v2/coderd/gitsshkey"
"github.com/coder/coder/v2/coderd/httpmw"
"github.com/coder/coder/v2/coderd/jobreaper"
"github.com/coder/coder/v2/coderd/notifications"
"github.com/coder/coder/v2/coderd/oauthpki"
"github.com/coder/coder/v2/coderd/prometheusmetrics"
"github.com/coder/coder/v2/coderd/prometheusmetrics/insights"
"github.com/coder/coder/v2/coderd/promoauth"
"github.com/coder/coder/v2/coderd/reaper"
"github.com/coder/coder/v2/coderd/schedule"
"github.com/coder/coder/v2/coderd/telemetry"
"github.com/coder/coder/v2/coderd/tracing"
Expand Down Expand Up @@ -1129,7 +1129,7 @@ func (r *RootCmd) Server(newAPI func(context.Context, *coderd.Options) (*coderd.

hangDetectorTicker := time.NewTicker(vals.JobHangDetectorInterval.Value())
defer hangDetectorTicker.Stop()
hangDetector := reaper.New(ctx, options.Database, options.Pubsub, logger, hangDetectorTicker.C)
hangDetector := jobreaper.New(ctx, options.Database, options.Pubsub, logger, hangDetectorTicker.C)
hangDetector.Start()
defer hangDetector.Close()

Expand Down
4 changes: 2 additions & 2 deletions coderd/coderdtest/coderdtest.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,11 +68,11 @@ import (
"github.com/coder/coder/v2/coderd/externalauth"
"github.com/coder/coder/v2/coderd/gitsshkey"
"github.com/coder/coder/v2/coderd/httpmw"
"github.com/coder/coder/v2/coderd/jobreaper"
"github.com/coder/coder/v2/coderd/notifications"
"github.com/coder/coder/v2/coderd/notifications/notificationstest"
"github.com/coder/coder/v2/coderd/rbac"
"github.com/coder/coder/v2/coderd/rbac/policy"
"github.com/coder/coder/v2/coderd/reaper"
"github.com/coder/coder/v2/coderd/runtimeconfig"
"github.com/coder/coder/v2/coderd/schedule"
"github.com/coder/coder/v2/coderd/telemetry"
Expand Down Expand Up @@ -367,7 +367,7 @@ func NewOptions(t testing.TB, options *Options) (func(http.Handler), context.Can

hangDetectorTicker := time.NewTicker(options.DeploymentValues.JobHangDetectorInterval.Value())
defer hangDetectorTicker.Stop()
hangDetector := reaper.New(ctx, options.Database, options.Pubsub, options.Logger.Named("reaper.detector"), hangDetectorTicker.C)
hangDetector := jobreaper.New(ctx, options.Database, options.Pubsub, options.Logger.Named("reaper.detector"), hangDetectorTicker.C)
hangDetector.Start()
t.Cleanup(hangDetector.Close)

Expand Down
2 changes: 1 addition & 1 deletion coderd/reaper/detector.go → coderd/jobreaper/detector.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
package reaper
package jobreaper

import (
"context"
Expand Down
56 changes: 28 additions & 28 deletions coderd/reaper/detector_test.go → coderd/jobreaper/detector_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
package reaper_test
package jobreaper_test

import (
"context"
Expand All @@ -20,9 +20,9 @@ import (
"github.com/coder/coder/v2/coderd/database/dbauthz"
"github.com/coder/coder/v2/coderd/database/dbgen"
"github.com/coder/coder/v2/coderd/database/dbtestutil"
"github.com/coder/coder/v2/coderd/jobreaper"
"github.com/coder/coder/v2/coderd/provisionerdserver"
"github.com/coder/coder/v2/coderd/rbac"
"github.com/coder/coder/v2/coderd/reaper"
"github.com/coder/coder/v2/provisionersdk"
"github.com/coder/coder/v2/testutil"
)
Expand All @@ -49,10 +49,10 @@ func jobLogMessages(jobType jobType, threshold float64) []string {
// reapParamsFromJob determines the type and threshold for a job being reaped
func reapParamsFromJob(job database.ProvisionerJob) (jobType, float64) {
jobType := hungJobType
threshold := reaper.HungJobDuration.Minutes()
threshold := jobreaper.HungJobDuration.Minutes()
if !job.StartedAt.Valid {
jobType = notStartedJobType
threshold = reaper.NotStartedTimeElapsed.Minutes()
threshold = jobreaper.NotStartedTimeElapsed.Minutes()
}
return jobType, threshold
}
Expand All @@ -69,10 +69,10 @@ func TestDetectorNoJobs(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- time.Now()

Expand All @@ -92,7 +92,7 @@ func TestDetectorNoHungJobs(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

// Insert some jobs that are running and haven't been updated in a while,
Expand All @@ -119,7 +119,7 @@ func TestDetectorNoHungJobs(t *testing.T) {
})
}

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

Expand All @@ -139,7 +139,7 @@ func TestDetectorHungWorkspaceBuild(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

var (
Expand Down Expand Up @@ -225,7 +225,7 @@ func TestDetectorHungWorkspaceBuild(t *testing.T) {
t.Log("previous job ID: ", previousWorkspaceBuildJob.ID)
t.Log("current job ID: ", currentWorkspaceBuildJob.ID)

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

Expand Down Expand Up @@ -261,7 +261,7 @@ func TestDetectorHungWorkspaceBuildNoOverrideState(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

var (
Expand Down Expand Up @@ -348,7 +348,7 @@ func TestDetectorHungWorkspaceBuildNoOverrideState(t *testing.T) {
t.Log("previous job ID: ", previousWorkspaceBuildJob.ID)
t.Log("current job ID: ", currentWorkspaceBuildJob.ID)

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

Expand Down Expand Up @@ -384,7 +384,7 @@ func TestDetectorHungWorkspaceBuildNoOverrideStateIfNoExistingBuild(t *testing.T
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

var (
Expand Down Expand Up @@ -441,7 +441,7 @@ func TestDetectorHungWorkspaceBuildNoOverrideStateIfNoExistingBuild(t *testing.T

t.Log("current job ID: ", currentWorkspaceBuildJob.ID)

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

Expand Down Expand Up @@ -477,7 +477,7 @@ func TestDetectorNotStartedWorkspaceBuildNoOverrideStateIfNoExistingBuild(t *tes
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

var (
Expand Down Expand Up @@ -533,7 +533,7 @@ func TestDetectorNotStartedWorkspaceBuildNoOverrideStateIfNoExistingBuild(t *tes

t.Log("current job ID: ", currentWorkspaceBuildJob.ID)

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

Expand Down Expand Up @@ -571,7 +571,7 @@ func TestDetectorHungOtherJobTypes(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

var (
Expand Down Expand Up @@ -633,7 +633,7 @@ func TestDetectorHungOtherJobTypes(t *testing.T) {
t.Log("template import job ID: ", templateImportJob.ID)
t.Log("template dry-run job ID: ", templateDryRunJob.ID)

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

Expand Down Expand Up @@ -675,7 +675,7 @@ func TestDetectorNotStartedOtherJobTypes(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

var (
Expand Down Expand Up @@ -736,7 +736,7 @@ func TestDetectorNotStartedOtherJobTypes(t *testing.T) {
t.Log("template import job ID: ", templateImportJob.ID)
t.Log("template dry-run job ID: ", templateDryRunJob.ID)

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

Expand Down Expand Up @@ -782,7 +782,7 @@ func TestDetectorHungCanceledJob(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

var (
Expand Down Expand Up @@ -822,7 +822,7 @@ func TestDetectorHungCanceledJob(t *testing.T) {

t.Log("template import job ID: ", templateImportJob.ID)

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

Expand Down Expand Up @@ -884,7 +884,7 @@ func TestDetectorPushesLogs(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
)

var (
Expand Down Expand Up @@ -937,7 +937,7 @@ func TestDetectorPushesLogs(t *testing.T) {
require.Len(t, logs, 10)
}

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()

// Create pubsub subscription to listen for new log events.
Expand Down Expand Up @@ -1004,15 +1004,15 @@ func TestDetectorMaxJobsPerRun(t *testing.T) {
db, pubsub = dbtestutil.NewDB(t)
log = testutil.Logger(t)
tickCh = make(chan time.Time)
statsCh = make(chan reaper.Stats)
statsCh = make(chan jobreaper.Stats)
org = dbgen.Organization(t, db, database.Organization{})
user = dbgen.User(t, db, database.User{})
file = dbgen.File(t, db, database.File{})
)

// Create MaxJobsPerRun + 1 hung jobs.
now := time.Now()
for i := 0; i < reaper.MaxJobsPerRun+1; i++ {
for i := 0; i < jobreaper.MaxJobsPerRun+1; i++ {
pj := dbgen.ProvisionerJob(t, db, pubsub, database.ProvisionerJob{
CreatedAt: now.Add(-time.Hour),
UpdatedAt: now.Add(-time.Hour),
Expand All @@ -1035,14 +1035,14 @@ func TestDetectorMaxJobsPerRun(t *testing.T) {
})
}

detector := reaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector := jobreaper.New(ctx, wrapDBAuthz(db, log), pubsub, log, tickCh).WithStatsChannel(statsCh)
detector.Start()
tickCh <- now

// Make sure that only MaxJobsPerRun jobs are terminated.
stats := <-statsCh
require.NoError(t, stats.Error)
require.Len(t, stats.TerminatedJobIDs, reaper.MaxJobsPerRun)
require.Len(t, stats.TerminatedJobIDs, jobreaper.MaxJobsPerRun)

// Run the detector again and make sure that only the remaining job is
// terminated.
Expand Down
4 changes: 2 additions & 2 deletions provisioner/terraform/serve.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ import (
"cdr.dev/slog"

"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/reaper"
"github.com/coder/coder/v2/coderd/jobreaper"
"github.com/coder/coder/v2/provisionersdk"
)

Expand Down Expand Up @@ -131,7 +131,7 @@ func Serve(ctx context.Context, options *ServeOptions) error {
options.Tracer = trace.NewNoopTracerProvider().Tracer("noop")
}
if options.ExitTimeout == 0 {
options.ExitTimeout = reaper.HungJobExitTimeout
options.ExitTimeout = jobreaper.HungJobExitTimeout
}
return provisionersdk.Serve(ctx, &server{
execMut: &sync.Mutex{},
Expand Down