diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index f7567ac01..65b2bdb4c 100644 --- a/internal/daemon/controller/controller.go +++ b/internal/daemon/controller/controller.go @@ -396,8 +396,8 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { jobRepoFn := func() (*job.Repository, error) { return job.NewRepository(ctx, dbase, dbase, c.kms) } - // TODO: Allow setting run jobs limit from config - schedulerOpts := []scheduler.Option{scheduler.WithRunJobsLimit(-1)} + + schedulerOpts := []scheduler.Option{} if c.conf.RawConfig.Controller.Scheduler.JobRunIntervalDuration > 0 { schedulerOpts = append(schedulerOpts, scheduler.WithRunJobsInterval(c.conf.RawConfig.Controller.Scheduler.JobRunIntervalDuration)) } diff --git a/internal/db/schema/migrations/oss/postgres/7/03_job.up.sql b/internal/db/schema/migrations/oss/postgres/7/03_job.up.sql index b572eff62..cf7cb29de 100644 --- a/internal/db/schema/migrations/oss/postgres/7/03_job.up.sql +++ b/internal/db/schema/migrations/oss/postgres/7/03_job.up.sql @@ -86,6 +86,7 @@ begin; create trigger immutable_columns before update on job_run for each row execute procedure immutable_columns('private_id', 'job_plugin_id', 'job_name', 'create_time'); + -- dropped in 93/02_drop_job_jobs_to_run.up.sql create view job_jobs_to_run as with running_jobs (job_plugin_id, job_name) as ( diff --git a/internal/db/schema/migrations/oss/postgres/93/02_drop_job_jobs_to_run.up.sql b/internal/db/schema/migrations/oss/postgres/93/02_drop_job_jobs_to_run.up.sql new file mode 100644 index 000000000..a1be8736d --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/93/02_drop_job_jobs_to_run.up.sql @@ -0,0 +1,10 @@ +-- Copyright (c) HashiCorp, Inc. +-- SPDX-License-Identifier: BUSL-1.1 + +-- This migration removes support for the job_jobs_to_run view as it is not used +-- anymore by the job repository. + +begin; + -- drops view from 7/03_job.up.sql + drop view job_jobs_to_run; +commit; \ No newline at end of file diff --git a/internal/scheduler/additional_verification_test.go b/internal/scheduler/additional_verification_test.go index bb5ec8c71..2d1bf498f 100644 --- a/internal/scheduler/additional_verification_test.go +++ b/internal/scheduler/additional_verification_test.go @@ -35,7 +35,7 @@ func TestSchedulerWorkflow(t *testing.T) { }) err := event.InitSysEventer(testLogger, testLock, "TestSchedulerWorkflow", event.WithEventerConfig(testConfig)) require.NoError(err) - sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second)) job1Ch := make(chan error) job1Ready := make(chan struct{}) @@ -118,7 +118,7 @@ func TestSchedulerCancelCtx(t *testing.T) { err := event.InitSysEventer(testLogger, testLock, "TestSchedulerCancelCtx", event.WithEventerConfig(testConfig)) require.NoError(err) - sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second)) fn, jobReady, jobDone := testJobFn() tj := testJob{name: "name", description: "desc", fn: fn, nextRunIn: time.Hour} @@ -168,7 +168,7 @@ func TestSchedulerInterruptedCancelCtx(t *testing.T) { err := event.InitSysEventer(testLogger, testLock, "TestSchedulerInterruptedCancelCtx", event.WithEventerConfig(testConfig)) require.NoError(err) - sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) fn, job1Ready, job1Done := testJobFn() tj1 := testJob{name: "name1", description: "desc", fn: fn, nextRunIn: time.Hour} @@ -270,7 +270,7 @@ func TestSchedulerJobProgress(t *testing.T) { err := event.InitSysEventer(testLogger, testLock, "TestSchedulerJobProgress", event.WithEventerConfig(testConfig)) require.NoError(err) - sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) jobReady := make(chan struct{}) done := make(chan struct{}) @@ -380,7 +380,7 @@ func TestSchedulerMonitorLoop(t *testing.T) { err := event.InitSysEventer(testLogger, testLock, "TestSchedulerMonitorLoop", event.WithEventerConfig(testConfig)) require.NoError(err) - sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithInterruptThreshold(time.Second), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithInterruptThreshold(time.Second), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second)) jobReady := make(chan struct{}) jobDone := make(chan struct{}) @@ -446,7 +446,7 @@ func TestSchedulerFinalStatusUpdate(t *testing.T) { err := event.InitSysEventer(testLogger, testLock, "TestSchedulerFinalStatusUpdate", event.WithEventerConfig(testConfig)) require.NoError(err) - sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Second)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Second)) jobReady := make(chan struct{}) jobErr := make(chan error) @@ -530,7 +530,7 @@ func TestSchedulerRunNow(t *testing.T) { require.NoError(err) // Create test scheduler that only runs jobs every hour - sched := TestScheduler(t, conn, wrapper, WithRunJobsLimit(10), WithRunJobsInterval(time.Hour)) + sched := TestScheduler(t, conn, wrapper, WithRunJobsInterval(time.Hour)) jobCh := make(chan struct{}) jobReady := make(chan struct{}) diff --git a/internal/scheduler/job/options.go b/internal/scheduler/job/options.go index 9dd852133..4d68ca4b6 100644 --- a/internal/scheduler/job/options.go +++ b/internal/scheduler/job/options.go @@ -8,8 +8,7 @@ import ( ) const ( - defaultRunJobsLimit = 1 - defaultPluginId = "pi_system" + defaultPluginId = "pi_system" ) // getOpts - iterate the inbound Options and return a struct @@ -27,16 +26,13 @@ type Option func(*options) // options = how options are represented type options struct { withNextRunIn time.Duration - withRunJobsLimit int withLimit int withName string withControllerId string } func getDefaultOptions() options { - return options{ - withRunJobsLimit: defaultRunJobsLimit, - } + return options{} // No default options. } // WithNextRunIn provides an option to provide the duration until the next run is scheduled. @@ -48,18 +44,6 @@ func WithNextRunIn(d time.Duration) Option { } } -// WithRunJobsLimit provides an option to provide the number of jobs to run. -// If WithRunJobsLimit == 0, then default run jobs limit is used. -// If WithRunJobsLimit < 0, then no limit is used. -func WithRunJobsLimit(l int) Option { - return func(o *options) { - o.withRunJobsLimit = l - if o.withRunJobsLimit == 0 { - o.withRunJobsLimit = defaultRunJobsLimit - } - } -} - // WithLimit provides an option to provide a limit for ListJobs. Intentionally // allowing negative integers. If WithLimit < 0, then unlimited results are // returned. If WithLimit == 0, then default limits are used for results. diff --git a/internal/scheduler/job/options_test.go b/internal/scheduler/job/options_test.go index ca0a53c88..115381b6e 100644 --- a/internal/scheduler/job/options_test.go +++ b/internal/scheduler/job/options_test.go @@ -21,21 +21,6 @@ func Test_GetOpts(t *testing.T) { testOpts.withNextRunIn = time.Hour assert.Equal(opts, testOpts) }) - t.Run("WithRunJobsLimit", func(t *testing.T) { - assert := assert.New(t) - opts := getOpts(WithRunJobsLimit(10)) - testOpts := getDefaultOptions() - assert.NotEqual(opts, testOpts) - testOpts.withRunJobsLimit = 10 - assert.Equal(opts, testOpts) - }) - t.Run("WithZeroRunJobsLimit", func(t *testing.T) { - assert := assert.New(t) - opts := getOpts(WithRunJobsLimit(0)) - testOpts := getDefaultOptions() - assert.Equal(opts, testOpts) - assert.Equal(defaultRunJobsLimit, opts.withRunJobsLimit) - }) t.Run("WithLimit", func(t *testing.T) { assert := assert.New(t) opts := getOpts(WithLimit(100)) diff --git a/internal/scheduler/job/query.go b/internal/scheduler/job/query.go index be3975d9f..686fb9b9c 100644 --- a/internal/scheduler/job/query.go +++ b/internal/scheduler/job/query.go @@ -7,13 +7,13 @@ const runJobsQuery = ` insert into job_run ( job_plugin_id, job_name, controller_id ) - select - job_plugin_id, job_name, ? - from job_jobs_to_run + select + j.plugin_id, j."name", ? + from job j + where next_scheduled_run <= current_timestamp order by next_scheduled_run asc - %s - on conflict - (job_plugin_id, job_name) + on conflict + (job_plugin_id, job_name) where status = 'running' do nothing returning *; diff --git a/internal/scheduler/job/repository_run.go b/internal/scheduler/job/repository_run.go index a13cfc9de..b5c0eec41 100644 --- a/internal/scheduler/job/repository_run.go +++ b/internal/scheduler/job/repository_run.go @@ -17,29 +17,17 @@ import ( // If there are not jobs to run, an empty slice will be returned with a nil error. // // • serverId is required and is the private_id of the server that will run the jobs. -// -// The only valid option is WithRunJobsLimit, if not provided RunJobs will run only 1 job. -func (r *Repository) RunJobs(ctx context.Context, serverId string, opt ...Option) ([]*Run, error) { +// No options are supported. +func (r *Repository) RunJobs(ctx context.Context, serverId string, _ ...Option) ([]*Run, error) { const op = "job.(Repository).RunJobs" if serverId == "" { return nil, errors.New(ctx, errors.InvalidParameter, op, "missing server id") } - opts := getOpts(opt...) - var limit string - switch { - case opts.withRunJobsLimit == 0: - // zero signals the defaults should be used. - limit = fmt.Sprintf("limit %d", defaultRunJobsLimit) - case opts.withRunJobsLimit > 0: - limit = fmt.Sprintf("limit %d", opts.withRunJobsLimit) - } - - query := fmt.Sprintf(runJobsQuery, limit) var runs []*Run _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { - rows, err := w.Query(ctx, query, []any{serverId}) + rows, err := w.Query(ctx, runJobsQuery, []any{serverId}) if err != nil { return errors.Wrap(ctx, err, op) } diff --git a/internal/scheduler/job/repository_run_test.go b/internal/scheduler/job/repository_run_test.go index 2b8e70f10..a5565520a 100644 --- a/internal/scheduler/job/repository_run_test.go +++ b/internal/scheduler/job/repository_run_test.go @@ -5,7 +5,6 @@ package job import ( "context" - "fmt" "sort" "testing" "time" @@ -120,73 +119,6 @@ func TestRepository_RunJobs(t *testing.T) { } } -func TestRepository_RunJobs_Limits(t *testing.T) { - t.Parallel() - ctx := context.Background() - conn, _ := db.TestSetup(t, "postgres") - rw := db.New(conn) - wrapper := db.TestWrapper(t) - kms := kms.TestKms(t, conn, wrapper) - iam.TestRepo(t, conn, wrapper) - - numJobs := 20 - server := testController(t, conn, wrapper) - - tests := []struct { - name string - opts []Option - wantLen int - }{ - { - name: "with-more-than-available", - opts: []Option{WithRunJobsLimit(numJobs * 2)}, - wantLen: numJobs, - }, - { - name: "with-no-option", - wantLen: defaultRunJobsLimit, - }, - { - name: "with-limit", - opts: []Option{WithRunJobsLimit(3)}, - wantLen: 3, - }, - { - name: "with-zero-limit", - opts: []Option{WithRunJobsLimit(0)}, - wantLen: defaultRunJobsLimit, - }, - { - name: "unlimited", - opts: []Option{WithRunJobsLimit(-1)}, - wantLen: numJobs, - }, - } - - for _, tt := range tests { - tt := tt - t.Run(tt.name, func(t *testing.T) { - assert, require := assert.New(t), require.New(t) - repo, err := NewRepository(ctx, rw, rw, kms) - assert.NoError(err) - require.NotNil(repo) - - for i := 0; i < numJobs; i++ { - testJob(t, conn, fmt.Sprintf("%v-%d", tt.name, i), "description", wrapper) - } - - got, err := repo.RunJobs(ctx, server.PrivateId, tt.opts...) - require.NoError(err) - assert.Len(got, tt.wantLen) - - // Clean up jobs for next run - rows, err := rw.Query(ctx, "delete from job", nil) - require.NoError(err) - _ = rows.Close() - }) - } -} - func TestRepository_RunJobsOrder(t *testing.T) { t.Parallel() ctx := context.Background() @@ -210,41 +142,18 @@ func TestRepository_RunJobsOrder(t *testing.T) { runs, err := repo.RunJobs(ctx, server.PrivateId) require.NoError(err) - require.Len(runs, 1) - run := runs[0] - assert.Equal(run.JobName, firstJob.Name) - assert.Equal(run.JobPluginId, firstJob.PluginId) + require.Len(runs, 3) - // End first job with time between last and middle - err = repo.CompleteRun(ctx, run.PrivateId, -6*time.Hour) - require.NoError(err) + // We should see the job runs ordered by scheduled time. + // firstJob > middleJob > lastJob + assert.Equal(firstJob.Name, runs[0].JobName) + assert.Equal(firstJob.PluginId, runs[0].JobPluginId) - runs, err = repo.RunJobs(ctx, server.PrivateId) - require.NoError(err) - require.Len(runs, 1) - run = runs[0] - assert.Equal(run.JobName, middleJob.Name) - assert.Equal(run.JobPluginId, middleJob.PluginId) + assert.Equal(middleJob.Name, runs[1].JobName) + assert.Equal(middleJob.PluginId, runs[1].JobPluginId) - // firstJob should be up again, as it is scheduled before lastJob - runs, err = repo.RunJobs(ctx, server.PrivateId) - require.NoError(err) - require.Len(runs, 1) - run = runs[0] - assert.Equal(run.JobName, firstJob.Name) - assert.Equal(run.JobPluginId, firstJob.PluginId) - - runs, err = repo.RunJobs(ctx, server.PrivateId) - require.NoError(err) - require.Len(runs, 1) - run = runs[0] - assert.Equal(run.JobName, lastJob.Name) - assert.Equal(run.JobPluginId, lastJob.PluginId) - - // All jobs are running no work should be returned - runs, err = repo.RunJobs(ctx, server.PrivateId) - require.NoError(err) - require.Len(runs, 0) + assert.Equal(lastJob.Name, runs[2].JobName) + assert.Equal(lastJob.PluginId, runs[2].JobPluginId) } func TestRepository_UpdateProgress(t *testing.T) { @@ -862,7 +771,6 @@ func TestRepository_InterruptServerRuns(t *testing.T) { runs: []args{ { ControllerId: server1.PrivateId, - opts: []Option{WithRunJobsLimit(3)}, expectedJobs: []*Job{job1, job2, job3}, }, }, @@ -877,7 +785,6 @@ func TestRepository_InterruptServerRuns(t *testing.T) { runs: []args{ { ControllerId: server2.PrivateId, - opts: []Option{WithRunJobsLimit(3)}, expectedJobs: []*Job{job1, job2, job3}, }, }, @@ -923,124 +830,6 @@ func TestRepository_InterruptServerRuns(t *testing.T) { }, }, }, - { - name: "multiple-servers-interrupt-all", - runs: []args{ - { - ControllerId: server1.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{job1}, - }, - { - ControllerId: server2.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{job2}, - }, - { - ControllerId: server3.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{job3}, - }, - }, - interrupts: []args{ - { - expectedJobs: []*Job{job1, job2, job3}, - }, - }, - }, - { - name: "multiple-servers-with-server-id", - runs: []args{ - { - ControllerId: server1.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{job1}, - }, - { - ControllerId: server2.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{job2}, - }, - { - ControllerId: server3.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{job3}, - }, - }, - interrupts: []args{ - { - opts: []Option{WithControllerId(server1.PrivateId)}, - expectedJobs: []*Job{job1}, - }, - { - opts: []Option{WithControllerId(server2.PrivateId)}, - expectedJobs: []*Job{job2}, - }, - { - opts: []Option{WithControllerId(server3.PrivateId)}, - expectedJobs: []*Job{job3}, - }, - }, - }, - { - name: "multiple-servers-distributed-runs", - runs: []args{ - { - ControllerId: server1.PrivateId, - opts: []Option{WithRunJobsLimit(2)}, - expectedJobs: []*Job{job1, job2}, - }, - { - ControllerId: server2.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{job3}, - }, - { - ControllerId: server3.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{}, - }, - }, - interrupts: []args{ - { - opts: []Option{WithControllerId(server1.PrivateId)}, - expectedJobs: []*Job{job1, job2}, - }, - { - opts: []Option{WithControllerId(server2.PrivateId)}, - expectedJobs: []*Job{job3}, - }, - { - opts: []Option{WithControllerId(server3.PrivateId)}, - expectedJobs: []*Job{}, - }, - }, - }, - { - name: "multiple-servers-distributed-runs-interrupt-all", - runs: []args{ - { - ControllerId: server1.PrivateId, - opts: []Option{WithRunJobsLimit(2)}, - expectedJobs: []*Job{job1, job2}, - }, - { - ControllerId: server2.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{job3}, - }, - { - ControllerId: server3.PrivateId, - opts: []Option{WithRunJobsLimit(1)}, - expectedJobs: []*Job{}, - }, - }, - interrupts: []args{ - { - expectedJobs: []*Job{job1, job2, job3}, - }, - }, - }, } for _, tt := range tests { tt := tt @@ -1049,7 +838,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) { require.NoError(err) for _, r := range tt.runs { - runs, err := repo.RunJobs(ctx, r.ControllerId, r.opts...) + runs, err := repo.RunJobs(ctx, r.ControllerId) require.NoError(err) assert.Len(runs, len(r.expectedJobs)) sort.Slice(runs, func(i, j int) bool { return runs[i].JobName < runs[j].JobName }) diff --git a/internal/scheduler/options.go b/internal/scheduler/options.go index c9a3c1d42..056ced186 100644 --- a/internal/scheduler/options.go +++ b/internal/scheduler/options.go @@ -6,7 +6,6 @@ package scheduler import "time" const ( - defaultRunJobsLimit = 1 defaultRunJobsInterval = time.Minute defaultMonitorInterval = 30 * time.Second defaultInterruptThreshold = 5 * time.Minute @@ -27,7 +26,6 @@ type Option func(*options) // options = how options are represented type options struct { withNextRunIn time.Duration - withRunJobsLimit int withRunJobInterval time.Duration withMonitorInterval time.Duration withInterruptThreshold time.Duration @@ -36,26 +34,12 @@ type options struct { func getDefaultOptions() options { return options{ - withRunJobsLimit: defaultRunJobsLimit, withRunJobInterval: defaultRunJobsInterval, withMonitorInterval: defaultMonitorInterval, withInterruptThreshold: defaultInterruptThreshold, } } -// WithRunJobsLimit provides an option to provide the number of jobs that will be requested -// by the scheduler when querying for jobs to run. -// If WithRunJobsLimit == 0, then default run jobs limit is used. -// If WithRunJobsLimit < 0, then no limit is used. -func WithRunJobsLimit(l int) Option { - return func(o *options) { - o.withRunJobsLimit = l - if o.withRunJobsLimit == 0 { - o.withRunJobsLimit = defaultRunJobsLimit - } - } -} - // WithRunJobsInterval provides an option to provide the interval at which the scheduler // will query the repository for jobs to run. // If WithRunJobsInterval == 0, then default interval is used. diff --git a/internal/scheduler/options_test.go b/internal/scheduler/options_test.go index d5e872a67..5b3fc4733 100644 --- a/internal/scheduler/options_test.go +++ b/internal/scheduler/options_test.go @@ -13,20 +13,6 @@ import ( // Test_GetOpts provides unit tests for GetOpts and all the options func Test_GetOpts(t *testing.T) { t.Parallel() - t.Run("WithRunJobsLimit", func(t *testing.T) { - assert := assert.New(t) - opts := getOpts(WithRunJobsLimit(10)) - testOpts := getDefaultOptions() - assert.NotEqual(opts, testOpts) - testOpts.withRunJobsLimit = 10 - assert.Equal(opts, testOpts) - }) - t.Run("WithZeroRunJobsLimit", func(t *testing.T) { - assert := assert.New(t) - opts := getOpts(WithRunJobsLimit(0)) - testOpts := getDefaultOptions() - assert.Equal(opts, testOpts) - }) t.Run("WithRunJobsInterval", func(t *testing.T) { assert := assert.New(t) opts := getOpts(WithRunJobsInterval(time.Hour)) diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 067854a90..66592fe7b 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -32,7 +32,6 @@ type Scheduler struct { runningJobs *sync.Map started ua.Bool - runJobsLimit int runJobsInterval time.Duration monitorInterval time.Duration interruptThreshold time.Duration @@ -45,7 +44,7 @@ type Scheduler struct { // // • jobRepoFn must be provided and is a function that returns the job repository // -// WithRunJobsLimit, WithRunJobsInterval, WithMonitorInterval and WithInterruptThreshold are +// WithRunJobsInterval, WithMonitorInterval and WithInterruptThreshold are // the only valid options. func New(ctx context.Context, serverId string, jobRepoFn jobRepoFactory, opt ...Option) (*Scheduler, error) { const op = "scheduler.New" @@ -62,7 +61,6 @@ func New(ctx context.Context, serverId string, jobRepoFn jobRepoFactory, opt ... jobRepoFn: jobRepoFn, registeredJobs: new(sync.Map), runningJobs: new(sync.Map), - runJobsLimit: opts.withRunJobsLimit, runJobsInterval: opts.withRunJobInterval, monitorInterval: opts.withMonitorInterval, interruptThreshold: opts.withInterruptThreshold, @@ -190,7 +188,7 @@ func (s *Scheduler) start(ctx context.Context) { event.WriteSysEvent(ctx, op, "scheduling loop running", "server id", s.serverId, "run interval", s.runJobsInterval.String(), - "run limit", s.runJobsLimit) + ) timer := time.NewTimer(0) var wg sync.WaitGroup for { @@ -218,7 +216,7 @@ func (s *Scheduler) schedule(ctx context.Context, wg *sync.WaitGroup) { return } - runs, err := repo.RunJobs(ctx, s.serverId, job.WithRunJobsLimit(s.runJobsLimit)) + runs, err := repo.RunJobs(ctx, s.serverId) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error getting jobs to run from repo")) return diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index c9b22bcbc..b96a55cdb 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -34,7 +34,6 @@ func TestScheduler_New(t *testing.T) { type args struct { serverId string jobRepo jobRepoFactory - runLimit int runInterval time.Duration monitorInterval time.Duration } @@ -70,7 +69,6 @@ func TestScheduler_New(t *testing.T) { }, want: args{ serverId: "test-server", - runLimit: defaultRunJobsLimit, runInterval: defaultRunJobsInterval, monitorInterval: defaultMonitorInterval, }, @@ -86,7 +84,6 @@ func TestScheduler_New(t *testing.T) { }, want: args{ serverId: "test-server", - runLimit: defaultRunJobsLimit, monitorInterval: defaultMonitorInterval, runInterval: time.Hour, }, @@ -97,12 +94,9 @@ func TestScheduler_New(t *testing.T) { serverId: "test-server", jobRepo: jobRepoFn, }, - opts: []Option{ - WithRunJobsLimit(-1), - }, + opts: []Option{}, want: args{ serverId: "test-server", - runLimit: -1, runInterval: defaultRunJobsInterval, monitorInterval: defaultMonitorInterval, }, @@ -113,12 +107,9 @@ func TestScheduler_New(t *testing.T) { serverId: "test-server", jobRepo: jobRepoFn, }, - opts: []Option{ - WithRunJobsLimit(20), - }, + opts: []Option{}, want: args{ serverId: "test-server", - runLimit: 20, runInterval: defaultRunJobsInterval, monitorInterval: defaultMonitorInterval, }, @@ -134,7 +125,6 @@ func TestScheduler_New(t *testing.T) { }, want: args{ serverId: "test-server", - runLimit: defaultRunJobsLimit, runInterval: defaultRunJobsInterval, monitorInterval: time.Hour, }, @@ -147,12 +137,10 @@ func TestScheduler_New(t *testing.T) { }, opts: []Option{ WithRunJobsInterval(time.Hour), - WithRunJobsLimit(20), WithMonitorInterval(2 * time.Hour), }, want: args{ serverId: "test-server", - runLimit: 20, runInterval: time.Hour, monitorInterval: 2 * time.Hour, }, @@ -174,7 +162,6 @@ func TestScheduler_New(t *testing.T) { require.NoError(err) assert.Equal(tt.want.serverId, got.serverId) - assert.Equal(tt.want.runLimit, got.runJobsLimit) assert.Equal(tt.want.runInterval, got.runJobsInterval) assert.Equal(tt.want.monitorInterval, got.monitorInterval) assert.NotNil(got.jobRepoFn)