Louis job scheduler (#1197)

* Adds job repo
* Adds scheduler
* Adds plugin table which is a place holder for future expansion of the scheduler. Currently all jobs that are created are linked to the pi_system plugin which are jobs that are held in a local registeredJobs map populated at controller startup.
This commit is contained in:
Louis Ruch authored and GitHub committed 2021-05-06 14:54:10 -07:00
1 parent d1d35e827c
commit feb5d7385d
40 files changed
+5445 -13

No files matched your search

+1
View File
@@ -133,6 +133,7 @@ protobuild:
@protoc-go-inject-tag -input=./internal/kms/store/oidc_key.pb.go
@protoc-go-inject-tag -input=./internal/target/store/target.pb.go
@protoc-go-inject-tag -input=./internal/auth/oidc/store/oidc.pb.go
@protoc-go-inject-tag -input=./internal/scheduler/job/store/job.pb.go
@rm -R ${TMP_DIR}
@@ -23,7 +23,7 @@ comment on domain wt_full_name is
'standard column for the full name of a person';
-- wt_url defines a type for URLs which must be longer that 3 chars and
-- less than 4k chars. It's defined to allow nulls, which can be overriden as
-- less than 4k chars. It's defined to allow nulls, which can be overridden as
-- needed when used in tables.
create domain wt_url as text
constraint wt_url_too_short
@@ -45,7 +45,7 @@ create domain wt_name as text
comment on domain wt_name is
'standard column for resource names';
-- wt_description defines a type for resource descriptionss that must be less
-- wt_description defines a type for resource descriptions that must be less
-- than 1024 chars. It's defined to allow nulls.
create domain wt_description as text
constraint wt_description_too_short
@@ -0,0 +1,23 @@
begin;
create function wt_add_seconds(sec integer, ts timestamp with time zone)
returns timestamp with time zone
as $$
select ts + sec * '1 second'::interval;
$$ language sql
stable
returns null on null input;
comment on function wt_add_seconds is
'wt_add_seconds returns ts + sec.';
create function wt_add_seconds_to_now(sec integer)
returns timestamp with time zone
as $$
select wt_add_seconds(sec, current_timestamp);
$$ language sql
stable
returns null on null input;
comment on function wt_add_seconds_to_now is
'wt_add_seconds_to_now returns current_timestamp + sec.';
commit;
@@ -0,0 +1,40 @@
begin;
create domain wt_plugin_id as text
not null
check(
length(trim(value)) > 10 or value = 'pi_system'
);
comment on domain wt_plugin_id is
'"pi_system", or random ID generated with github.com/hashicorp/vault/sdk/helper/base62';
create table plugin (
public_id wt_plugin_id primary key
);
comment on table plugin is
'plugin is a table where each row represents a unique plugin registered with Boundary.';
insert into plugin (public_id)
values
('pi_system');
create trigger immutable_columns before update on plugin
for each row execute procedure immutable_columns('public_id');
create or replace function
disallow_system_plugin_deletion()
returns trigger
as $$
begin
if old.public_id = 'pi_system' then
raise exception 'deletion of system plugin not allowed';
end if;
return old;
end;
$$ language plpgsql;
create trigger plugin_disallow_system_deletion before delete on plugin
for each row execute procedure disallow_system_plugin_deletion();
commit;
@@ -0,0 +1,111 @@
begin;
create table job (
plugin_id wt_plugin_id not null
constraint plugin_fk
references plugin(public_id)
on delete cascade
on update cascade,
name wt_name not null,
description wt_description not null,
next_scheduled_run timestamp with time zone not null,
primary key (plugin_id, name)
);
comment on table job is
'job is a table where each row represents a unique job that can only have one running instance at any specific time.';
create trigger immutable_columns before update on job
for each row execute procedure immutable_columns('plugin_id', 'name');
create table job_run_status_enm (
name text not null primary key
constraint only_predefined_job_status_allowed
check(name in ('running', 'completed', 'failed', 'interrupted'))
);
comment on table job_run_status_enm is
'job_run_status_enm is an enumeration table where each row contains a valid job run state.';
insert into job_run_status_enm (name)
values
('running'),
('completed'),
('failed'),
('interrupted');
create table job_run (
private_id wh_dim_id primary key
default wh_dim_id(),
job_plugin_id wt_plugin_id not null,
job_name wt_name not null,
server_id wt_private_id
constraint server_fkey
references server(private_id)
on delete set null
on update cascade,
create_time wt_timestamp,
update_time wt_timestamp,
end_time timestamp with time zone,
completed_count int not null
default 0
constraint completed_count_can_not_be_negative
check(completed_count >= 0),
total_count int not null
default 0
constraint total_count_can_not_be_negative
check(total_count >= 0),
status text not null
default 'running'
constraint job_run_status_enm_fkey
references job_run_status_enm (name)
on delete restrict
on update cascade,
constraint job_run_completed_count_less_than_equal_to_total_count
check(completed_count <= total_count),
constraint job_fkey
foreign key (job_plugin_id, job_name)
references job (plugin_id, name)
on delete cascade
on update cascade
);
comment on table job_run is
'job_run is a table where each row represents an instance of a job run that is either actively running or has already completed.';
create unique index job_run_status_constraint
on job_run (job_plugin_id, job_name)
where status = 'running';
create trigger update_time_column before update on job_run
for each row execute procedure update_time_column();
create trigger default_create_time_column before insert on job_run
for each row execute procedure default_create_time();
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');
create view job_jobs_to_run as
with
running_jobs (job_plugin_id, job_name) as (
select job_plugin_id, job_name
from job_run
where status = 'running'
),
final (job_plugin_id, job_name, next_scheduled_run) as (
select plugin_id, name, next_scheduled_run
from job j
where next_scheduled_run <= current_timestamp
and not exists (
select
from running_jobs
where job_plugin_id = j.plugin_id
and job_name = j.name
)
)
select job_plugin_id, job_name, next_scheduled_run from final;
commit;
+171 -3
View File
@@ -4,7 +4,7 @@ package schema
func init() {
migrationStates["postgres"] = migrationState{
binarySchemaVersion: 6001,
binarySchemaVersion: 7003,
upMigrations: map[int][]byte{
1: []byte(`
create domain wt_public_id as text
@@ -4892,7 +4892,7 @@ comment on domain wt_full_name is
'standard column for the full name of a person';
-- wt_url defines a type for URLs which must be longer that 3 chars and
-- less than 4k chars. It's defined to allow nulls, which can be overriden as
-- less than 4k chars. It's defined to allow nulls, which can be overridden as
-- needed when used in tables.
create domain wt_url as text
constraint wt_url_too_short
@@ -4914,7 +4914,7 @@ create domain wt_name as text
comment on domain wt_name is
'standard column for resource names';
-- wt_description defines a type for resource descriptionss that must be less
-- wt_description defines a type for resource descriptions that must be less
-- than 1024 chars. It's defined to allow nulls.
create domain wt_description as text
constraint wt_description_too_short
@@ -6178,6 +6178,174 @@ from
group by am.public_id, is_primary_auth_method; -- there can be only one public_id + is_primary_auth_method, so group by isn't a problem.
comment on view oidc_auth_method_with_value_obj is
'oidc auth method with its associated value objects (algs, auds, certs, scopes) as columns with | delimited values';
`),
7001: []byte(`
create function wt_add_seconds(sec integer, ts timestamp with time zone)
returns timestamp with time zone
as $$
select ts + sec * '1 second'::interval;
$$ language sql
stable
returns null on null input;
comment on function wt_add_seconds is
'wt_add_seconds returns ts + sec.';
create function wt_add_seconds_to_now(sec integer)
returns timestamp with time zone
as $$
select wt_add_seconds(sec, current_timestamp);
$$ language sql
stable
returns null on null input;
comment on function wt_add_seconds_to_now is
'wt_add_seconds_to_now returns current_timestamp + sec.';
`),
7002: []byte(`
create domain wt_plugin_id as text
not null
check(
length(trim(value)) > 10 or value = 'pi_system'
);
comment on domain wt_plugin_id is
'"pi_system", or random ID generated with github.com/hashicorp/vault/sdk/helper/base62';
create table plugin (
public_id wt_plugin_id primary key
);
comment on table plugin is
'plugin is a table where each row represents a unique plugin registered with Boundary.';
insert into plugin (public_id)
values
('pi_system');
create trigger immutable_columns before update on plugin
for each row execute procedure immutable_columns('public_id');
create or replace function
disallow_system_plugin_deletion()
returns trigger
as $$
begin
if old.public_id = 'pi_system' then
raise exception 'deletion of system plugin not allowed';
end if;
return old;
end;
$$ language plpgsql;
create trigger plugin_disallow_system_deletion before delete on plugin
for each row execute procedure disallow_system_plugin_deletion();
`),
7003: []byte(`
create table job (
plugin_id wt_plugin_id not null
constraint plugin_fk
references plugin(public_id)
on delete cascade
on update cascade,
name wt_name not null,
description wt_description not null,
next_scheduled_run timestamp with time zone not null,
primary key (plugin_id, name)
);
comment on table job is
'job is a table where each row represents a unique job that can only have one running instance at any specific time.';
create trigger immutable_columns before update on job
for each row execute procedure immutable_columns('plugin_id', 'name');
create table job_run_status_enm (
name text not null primary key
constraint only_predefined_job_status_allowed
check(name in ('running', 'completed', 'failed', 'interrupted'))
);
comment on table job_run_status_enm is
'job_run_status_enm is an enumeration table where each row contains a valid job run state.';
insert into job_run_status_enm (name)
values
('running'),
('completed'),
('failed'),
('interrupted');
create table job_run (
private_id wh_dim_id primary key
default wh_dim_id(),
job_plugin_id wt_plugin_id not null,
job_name wt_name not null,
server_id wt_private_id
constraint server_fkey
references server(private_id)
on delete set null
on update cascade,
create_time wt_timestamp,
update_time wt_timestamp,
end_time timestamp with time zone,
completed_count int not null
default 0
constraint completed_count_can_not_be_negative
check(completed_count >= 0),
total_count int not null
default 0
constraint total_count_can_not_be_negative
check(total_count >= 0),
status text not null
default 'running'
constraint job_run_status_enm_fkey
references job_run_status_enm (name)
on delete restrict
on update cascade,
constraint job_run_completed_count_less_than_equal_to_total_count
check(completed_count <= total_count),
constraint job_fkey
foreign key (job_plugin_id, job_name)
references job (plugin_id, name)
on delete cascade
on update cascade
);
comment on table job_run is
'job_run is a table where each row represents an instance of a job run that is either actively running or has already completed.';
create unique index job_run_status_constraint
on job_run (job_plugin_id, job_name)
where status = 'running';
create trigger update_time_column before update on job_run
for each row execute procedure update_time_column();
create trigger default_create_time_column before insert on job_run
for each row execute procedure default_create_time();
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');
create view job_jobs_to_run as
with
running_jobs (job_plugin_id, job_name) as (
select job_plugin_id, job_name
from job_run
where status = 'running'
),
final (job_plugin_id, job_name, next_scheduled_run) as (
select plugin_id, name, next_scheduled_run
from job j
where next_scheduled_run <= current_timestamp
and not exists (
select
from running_jobs
where job_plugin_id = j.plugin_id
and job_name = j.name
)
)
select job_plugin_id, job_name, next_scheduled_run from final;
`),
},
}
+28 -8
View File
@@ -89,16 +89,21 @@ func AssertPublicId(t *testing.T, prefix, actual string) {
assert.Equalf(t, prefix, parts[0], "PublicId want prefix: %q, got: %q in %q", prefix, parts[0], actual)
}
// TestVerifyOplog will verify that there is an oplog entry. An error is
// returned if the entry or it's metadata is not found. Returning an error
// TestVerifyOplog will verify that there is an oplog entry that matches the provided resourceId.
// By default it matches the provided resourceId against the `resource-public-id` tag. If the
// WithResourcePrivateId option is provided, the lookup will use the `resource-private-id` tag.
// An error is returned if the entry or it's metadata is not found. Returning an error
// allows clients to test if an entry was not written, which is a valid use case.
func TestVerifyOplog(t *testing.T, r Reader, resourcePublicId string, opt ...TestOption) error {
func TestVerifyOplog(t *testing.T, r Reader, resourceId string, opt ...TestOption) error {
t.Helper()
// sql where clauses
const (
whereBase = `
key = 'resource-public-id'
key = ?
and value = ?
`
whereOptype = `
and entry_id in (
select entry_id
@@ -116,9 +121,15 @@ and create_time > NOW()::timestamp - (interval '1 second' * ?)
withOperation := opts.withOperation
withCreateNotBefore := opts.withCreateNotBefore
whereKey := "resource-public-id"
if opts.withResourcePrivateId {
whereKey = "resource-private-id"
}
where := whereBase
whereArgs := []interface{}{
resourcePublicId,
whereKey,
resourceId,
}
if withOperation != oplog.OpType_OP_TYPE_UNSPECIFIED {
@@ -157,9 +168,10 @@ type TestOption func(*testOptions)
// options = how options are represented
type testOptions struct {
withCreateNotBefore *int
withOperation oplog.OpType
withTestDatabaseUrl string
withCreateNotBefore *int
withOperation oplog.OpType
withTestDatabaseUrl string
withResourcePrivateId bool
}
func getDefaultTestOptions() testOptions {
@@ -191,3 +203,11 @@ func WithTestDatabaseUrl(url string) TestOption {
o.withTestDatabaseUrl = url
}
}
// WithResourcePrivateId provides a way to specify that the resource lookup action
// uses `resource-private-id` tag instead of the default `resource-public-id` tag
func WithResourcePrivateId(enable bool) TestOption {
return func(o *testOptions) {
o.withResourcePrivateId = enable
}
}
+1
View File
@@ -36,6 +36,7 @@ const (
TokenMismatch Code = 112 // TokenMismatch represents that there was a token mismatch
TooShort Code = 113 // TooShort represents an error that means the provided input is not meeting minimum length requirements
AccountAlreadyAssociated Code = 114 // AccountAlreadyAssociated represents an attempt to associate an account failed since it was already associated.
InvalidJobRunState Code = 115 // InvalidJobRunState represents that a JobRun was in an invalid state
AuthAttemptExpired Code = 198 // AuthAttemptExpired represents an expired authentication attempt
AuthMethodInactive Code = 199 // AuthMethodInactive represents an error that means the auth method is not active.
+5
View File
@@ -102,6 +102,11 @@ func TestCode_Both_String_Info(t *testing.T) {
c: AccountAlreadyAssociated,
want: AccountAlreadyAssociated,
},
{
name: "InvalidJobRunState",
c: InvalidJobRunState,
want: InvalidJobRunState,
},
{
name: "InternalError",
c: Internal,
+4
View File
@@ -84,6 +84,10 @@ var errorCodeInfo = map[Code]Info{
Message: "account already associated with another user",
Kind: Parameter,
},
InvalidJobRunState: {
Message: "job run is already in a final run state",
Kind: Integrity,
},
PasswordTooShort: {
Message: "too short",
Kind: Password,
@@ -0,0 +1,68 @@
syntax = "proto3";
// Package store provides protobufs for storing types in the job package.
package controller.storage.job.store.v1;
option go_package = "github.com/hashicorp/boundary/internal/scheduler/job/store;store";
import "controller/storage/timestamp/v1/timestamp.proto";
import "controller/custom_options/v1/options.proto";
message Job {
// plugin_id is the primary key of the plugin that registered and owns the job
// @inject_tag: `gorm:"primary_key"`
string plugin_id = 1;
// The name of the job.
// @inject_tag: `gorm:"primary_key"`
string name = 2;
// The human friendly description of the job.
// @inject_tag: `gorm:"not_null"`
string description = 3;
// next_scheduled_run is the time that the next run should be created.
// @inject_tag: `gorm:"default:current_timestamp"`
timestamp.v1.Timestamp next_scheduled_run = 4;
}
message JobRun {
// private_id is used to access the job run but not intended to be available via the API
// @inject_tag: `gorm:"primary_key"`
string private_id = 1;
// The job_plugin_id is the id of the plugin that owns the job.
// @inject_tag: `gorm:"not_null"`
string job_plugin_id = 2;
// The job_name is the name of the job and must be set.
// @inject_tag: `gorm:"not_null"`
string job_name = 3;
// The server_id of the server running the job and must be set.
// @inject_tag: `gorm:"not_null"`
string server_id = 4;
// The create_time is set by the database.
// @inject_tag: `gorm:"default:current_timestamp"`
timestamp.v1.Timestamp create_time = 5;
// The update_time is set by the database.
// @inject_tag: `gorm:"default:current_timestamp"`
timestamp.v1.Timestamp update_time = 6;
// The end_time is set when the job run completes.
// @inject_tag: `gorm:"default:null"`
timestamp.v1.Timestamp end_time = 7;
// completed_count is set during an update to indicate the current progress.
// @inject_tag: `gorm:"default:0"`
uint32 completed_count = 8;
// total_count is set during an update to indicate the progress goal.
// @inject_tag: `gorm:"default:0"`
uint32 total_count = 9;
// status of the job run (running, completed, failed or interrupted).
// @inject_tag: `gorm:"not_null"`
string status = 10;
}
@@ -0,0 +1,345 @@
package scheduler
import (
"context"
"fmt"
"runtime"
"testing"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/scheduler/job"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestSchedulerWorkflow(t *testing.T) {
t.Parallel()
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
wrapper := db.TestWrapper(t)
iam.TestRepo(t, conn, wrapper)
server := testController(t, conn, wrapper)
sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithRunJobsInterval(time.Second))
job1Ch := make(chan error)
job1Ready := make(chan struct{})
fn1 := func(_ context.Context) error {
job1Ready <- struct{}{}
return <-job1Ch
}
tj1 := testJob{name: "name1", description: "desc", fn: fn1, nextRunIn: time.Hour}
err := sched.RegisterJob(context.Background(), tj1)
require.NoError(err)
job2Ch := make(chan error)
job2Ready := make(chan struct{})
fn2 := func(_ context.Context) error {
job2Ready <- struct{}{}
return <-job2Ch
}
tj2 := testJob{name: "name2", description: "desc", fn: fn2, nextRunIn: time.Hour}
err = sched.RegisterJob(context.Background(), tj2)
require.NoError(err)
err = sched.Start(context.Background())
require.NoError(err)
// Wait for scheduler to run both jobs
<-job1Ready
<-job2Ready
assert.Equal(mapLen(sched.runningJobs), 2)
// Fail first job, complete second job
job1Ch <- fmt.Errorf("failure")
job2Ch <- nil
// Scheduler should only try and run job1 again as job2 was successful
<-job1Ready
require.Equal(mapLen(sched.runningJobs), 1)
// Complete job 1
job1Ch <- nil
// Update job2 to run again
err = sched.UpdateJobNextRun(context.Background(), tj2.name, 0)
require.NoError(err)
<-job2Ready
require.Equal(mapLen(sched.runningJobs), 1)
// Complete job 2
job2Ch <- nil
}
func TestSchedulerCancelCtx(t *testing.T) {
t.Parallel()
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
wrapper := db.TestWrapper(t)
iam.TestRepo(t, conn, wrapper)
server := testController(t, conn, wrapper)
sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithRunJobsInterval(time.Second))
fn, jobReady, jobDone := testJobFn()
tj := testJob{name: "name", description: "desc", fn: fn, nextRunIn: time.Hour}
err := sched.RegisterJob(context.Background(), tj)
require.NoError(err)
baseCtx, baseCnl := context.WithCancel(context.Background())
err = sched.Start(baseCtx)
require.NoError(err)
// Wait for scheduler to run job
<-jobReady
assert.Equal(mapLen(sched.runningJobs), 1)
// Yield processor
runtime.Gosched()
// Verify job is not done
select {
case <-jobDone:
t.Fatal("expected job to be blocking on context")
default:
}
// Cancel the base context and all job context's should be cancelled and exit
baseCnl()
<-jobDone
}
func TestSchedulerInterruptedCancelCtx(t *testing.T) {
t.Parallel()
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
wrapper := db.TestWrapper(t)
rw := db.New(conn)
kmsCache := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
server := testController(t, conn, wrapper)
sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
fn, job1Ready, job1Done := testJobFn()
tj1 := testJob{name: "name1", description: "desc", fn: fn, nextRunIn: time.Hour}
err := sched.RegisterJob(context.Background(), tj1)
require.NoError(err)
fn, job2Ready, job2Done := testJobFn()
tj2 := testJob{name: "name2", description: "desc", fn: fn, nextRunIn: time.Hour}
err = sched.RegisterJob(context.Background(), tj2)
require.NoError(err)
baseCtx, baseCnl := context.WithCancel(context.Background())
defer baseCnl()
err = sched.Start(baseCtx)
require.NoError(err)
// Wait for scheduler to run both job
<-job1Ready
<-job2Ready
require.Equal(mapLen(sched.runningJobs), 2)
runJob, ok := sched.runningJobs.Load(tj1.name)
require.True(ok)
run1Id := runJob.(*runningJob).runId
runJob, ok = sched.runningJobs.Load(tj2.name)
require.True(ok)
run2Id := runJob.(*runningJob).runId
// Yield processor
runtime.Gosched()
// Verify job 1 is not done
select {
case <-job1Done:
t.Fatal("expected job 1 to be blocking on context")
default:
}
// Verify job 2 is not done
select {
case <-job2Done:
t.Fatal("expected job 2 to be blocking on context")
default:
}
// Interrupt job 1 run to cause monitor loop to trigger cancel
repo, err := job.NewRepository(rw, rw, kmsCache)
require.NoError(err)
run, err := repo.LookupRun(context.Background(), run1Id)
require.NoError(err)
run.Status = string(job.Interrupted)
rowsUpdated, err := rw.Update(context.Background(), run, []string{"Status"}, nil)
require.NoError(err)
assert.Equal(1, rowsUpdated)
// Once monitor cancels context the job should exit
<-job1Done
// Yield processor
runtime.Gosched()
// Verify job 2 is not done
select {
case <-job2Done:
t.Fatal("expected job 2 to be blocking on context")
default:
}
// Interrupt job 2 run to cause monitor loop to trigger cancel
repo, err = job.NewRepository(rw, rw, kmsCache)
require.NoError(err)
run, err = repo.LookupRun(context.Background(), run2Id)
require.NoError(err)
run.Status = string(job.Interrupted)
rowsUpdated, err = rw.Update(context.Background(), run, []string{"Status"}, nil)
require.NoError(err)
assert.Equal(1, rowsUpdated)
// Once monitor cancels context the job should exit
<-job2Done
}
func TestSchedulerJobProgress(t *testing.T) {
t.Parallel()
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
wrapper := db.TestWrapper(t)
rw := db.New(conn)
kmsCache := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
server := testController(t, conn, wrapper)
sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
jobReady := make(chan struct{})
fn := func(ctx context.Context) error {
jobReady <- struct{}{}
<-ctx.Done()
return nil
}
statusRequest := make(chan struct{})
jobStatus := make(chan JobStatus)
status := func() JobStatus {
statusRequest <- struct{}{}
return <-jobStatus
}
tj := testJob{name: "name", description: "desc", fn: fn, statusFn: status, nextRunIn: time.Hour}
err := sched.RegisterJob(context.Background(), tj)
require.NoError(err)
baseCtx, baseCnl := context.WithCancel(context.Background())
err = sched.Start(baseCtx)
require.NoError(err)
// Wait for scheduler to run job
<-jobReady
require.Equal(mapLen(sched.runningJobs), 1)
runJob, ok := sched.runningJobs.Load(tj.name)
require.True(ok)
runId := runJob.(*runningJob).runId
// Wait for scheduler to query for job status
<-statusRequest
// Send progress to monitor loop to persist
jobStatus <- JobStatus{Total: 10, Completed: 0}
// Wait for scheduler to query for job status before verifying previous results
<-statusRequest
repo, err := job.NewRepository(rw, rw, kmsCache)
require.NoError(err)
run, err := repo.LookupRun(context.Background(), runId)
require.NoError(err)
assert.Equal(string(job.Running), run.Status)
assert.Equal(uint32(10), run.TotalCount)
assert.Equal(uint32(0), run.CompletedCount)
// Send progress to monitor loop to persist
jobStatus <- JobStatus{Total: 20, Completed: 10}
// Wait for scheduler to query for job status before verifying previous results
<-statusRequest
run, err = repo.LookupRun(context.Background(), runId)
require.NoError(err)
assert.Equal(string(job.Running), run.Status)
assert.Equal(uint32(20), run.TotalCount)
assert.Equal(uint32(10), run.CompletedCount)
// Send progress to monitor loop to persist
jobStatus <- JobStatus{Total: 10, Completed: 20}
// Wait for scheduler to query for job status before verifying previous results
<-statusRequest
// Previous job status was invalid and should not have been persisted
run, err = repo.LookupRun(context.Background(), runId)
require.NoError(err)
assert.Equal(string(job.Running), run.Status)
assert.Equal(uint32(20), run.TotalCount)
assert.Equal(uint32(10), run.CompletedCount)
baseCnl()
// unblock goroutines waiting on channels
jobStatus <- JobStatus{}
}
func TestSchedulerMonitorLoop(t *testing.T) {
t.Parallel()
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
wrapper := db.TestWrapper(t)
rw := db.New(conn)
kmsCache := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
server := testController(t, conn, wrapper)
sched := testScheduler(t, conn, wrapper, server.PrivateId, WithRunJobsLimit(10), WithInterruptThreshold(time.Second), WithRunJobsInterval(time.Second), WithMonitorInterval(time.Second))
jobReady := make(chan struct{})
jobDone := make(chan struct{})
fn := func(ctx context.Context) error {
jobReady <- struct{}{}
<-ctx.Done()
jobDone <- struct{}{}
return nil
}
tj := testJob{name: "name", description: "desc", fn: fn, nextRunIn: time.Hour}
err := sched.RegisterJob(context.Background(), tj)
require.NoError(err)
baseCtx, baseCnl := context.WithCancel(context.Background())
defer baseCnl()
err = sched.Start(baseCtx)
require.NoError(err)
// Wait for scheduler to run job
<-jobReady
require.Equal(mapLen(sched.runningJobs), 1)
runJob, ok := sched.runningJobs.Load(tj.name)
require.True(ok)
runId := runJob.(*runningJob).runId
// Wait for scheduler to interrupt job
<-jobDone
repo, err := job.NewRepository(rw, rw, kmsCache)
require.NoError(err)
run, err := repo.LookupRun(context.Background(), runId)
require.NoError(err)
assert.Equal(string(job.Interrupted), run.Status)
}
+6
View File
@@ -0,0 +1,6 @@
// Package scheduler allows callers to register recurring jobs on the controller. The scheduler
// will periodically query the repository for registered jobs that should be run.
//
// Before a job can be invoked by the scheduler, it must be made known to the scheduler
// by being registered on the instance of the scheduler that is running.
package scheduler
+52
View File
@@ -0,0 +1,52 @@
package scheduler
import (
"context"
"time"
"github.com/hashicorp/boundary/internal/errors"
)
// Job defines an interface for jobs that can be invoked by the scheduler.
type Job interface {
// Status reports the job’s current status. The status is periodically persisted by
// the scheduler when a job is running, and will be used to verify a job is making progress.
Status() JobStatus
// Run performs the required work depending on the implementation.
// The context is used to notify the job that it should exit early.
Run(ctx context.Context) error
// NextRunIn returns the duration until the next job run should be scheduled. This
// method is invoked after a run has successfully completed and the next run time
// is being persisted by the scheduler.
NextRunIn() time.Duration
// Name is the human readable name of the job.
Name() string
// Description is the human readable description of the job.
Description() string
}
// JobStatus defines the struct that must be returned by the Job.Status() method.
type JobStatus struct {
// Completed and Total are used to indicate job progress,
// each job implementation will determine the definition of
// progress by calculating both Completed and Total.
Completed, Total int
}
func validateJob(j Job) error {
const op = "scheduler.validateJob"
if j == nil {
return errors.New(errors.InvalidParameter, op, "missing job")
}
if j.Name() == "" {
return errors.New(errors.InvalidParameter, op, "missing name")
}
if j.Description() == "" {
return errors.New(errors.InvalidParameter, op, "missing description")
}
return nil
}
@@ -0,0 +1,188 @@
package job
import (
"context"
"testing"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestJobWorkflow(t *testing.T) {
t.Parallel()
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kms := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
server := testController(t, conn, wrapper)
repo, err := NewRepository(rw, rw, kms)
require.NoError(err)
job, err := repo.CreateJob(context.Background(), "job1", "description")
require.NoError(err)
require.NotNil(job)
assert.Equal(defaultPluginId, job.PluginId)
runs, err := repo.RunJobs(context.Background(), server.PrivateId)
require.NoError(err)
assert.Len(runs, 1)
run := runs[0]
assert.NotEmpty(run.PrivateId)
assert.Equal(job.Name, run.JobName)
run, err = repo.UpdateProgress(context.Background(), run.PrivateId, 100, 110)
require.NoError(err)
assert.Equal(uint32(100), run.CompletedCount)
assert.Equal(uint32(110), run.TotalCount)
// The only available job is already running, a request for work should return nil
newRuns, err := repo.RunJobs(context.Background(), server.PrivateId)
require.NoError(err)
assert.Nil(newRuns)
run, err = repo.CompleteRun(context.Background(), run.PrivateId, time.Hour)
require.NoError(err)
assert.Equal(Completed.string(), run.Status)
job, err = repo.LookupJob(context.Background(), job.Name)
require.NoError(err)
assert.NotNil(job)
// The only available job has a next run in the future, a request for work should return nil
newRuns, err = repo.RunJobs(context.Background(), server.PrivateId)
require.NoError(err)
assert.Nil(newRuns)
// Update job next run to time in past
job, err = repo.UpdateJobNextRun(context.Background(), job.Name, 0)
require.NoError(err)
// Now that next scheduled time is in past, a request for work should return a Run
newRuns, err = repo.RunJobs(context.Background(), server.PrivateId)
require.NoError(err)
require.Len(newRuns, 1)
newRun := newRuns[0]
require.NotEmpty(newRun.PrivateId)
assert.Equal(job.Name, newRun.JobName)
assert.NotEqual(run.PrivateId, newRun.PrivateId)
// The only available job is already running, a request for work should return nil
newRuns, err = repo.RunJobs(context.Background(), server.PrivateId)
require.NoError(err)
assert.Nil(newRuns)
newRun, err = repo.FailRun(context.Background(), newRun.PrivateId)
require.NoError(err)
assert.Equal(Failed.string(), newRun.Status)
// Run failed so the job should be available for work immediately
newRuns, err = repo.RunJobs(context.Background(), server.PrivateId)
require.NoError(err)
assert.Len(newRuns, 1)
}
// TODO (LCR): This test can be dropped once plugin support is plumbed through job repo and scheduler
func TestPlugin(t *testing.T) {
t.Parallel()
t.Run("plugin-unique", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
rows, err := rw.Query(context.Background(), createJobQuery, []interface{}{
defaultPluginId,
"same-job-name",
"description",
0,
})
require.NoError(err)
_ = rows.Close()
// Creating a job with same name and pluginId should return a unique constraint error
rows, err = rw.Query(context.Background(), createJobQuery, []interface{}{
defaultPluginId,
"same-job-name",
"description",
0,
})
require.Error(err)
assert.Nil(rows)
assert.Equal("pq: duplicate key value violates unique constraint \"job_pkey\"", err.Error())
// Create test plugin id
testPluginId := "pi_test1234"
numRows, err := rw.Exec(context.Background(), "insert into plugin(public_id) values (?)", []interface{}{testPluginId})
require.NoError(err)
assert.Equal(1, numRows)
// Creating the a job with the same name and different pluginId should succeed
rows, err = rw.Query(context.Background(), createJobQuery, []interface{}{
testPluginId,
"same-job-name",
"description",
0,
})
assert.NoError(err)
_ = rows.Close()
// Creating a job with same name and pluginId should again return a unique constraint error
rows, err = rw.Query(context.Background(), createJobQuery, []interface{}{
testPluginId,
"same-job-name",
"description",
0,
})
require.Error(err)
assert.Nil(rows)
assert.Equal("pq: duplicate key value violates unique constraint \"job_pkey\"", err.Error())
})
t.Run("plugin-id-immutable", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
// Create test plugin id
testPluginId := "pi_test1234"
numRows, err := rw.Exec(context.Background(), "insert into plugin(public_id) values (?)", []interface{}{testPluginId})
assert.NoError(err)
assert.Equal(1, numRows)
newPluginId := "pi_newtest1234"
numUpdated, err := rw.Exec(context.Background(), "update plugin set public_id = ? where public_id = ?", []interface{}{newPluginId, testPluginId})
require.Error(err)
assert.Equal("db.Exec: immutable column: plugin.public_id: integrity violation: error #1003", err.Error())
assert.Equal(0, numUpdated)
})
t.Run("default-plugin-deletion-disallowed", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
numDeleted, err := rw.Exec(context.Background(), "delete from plugin where public_id = ?", []interface{}{defaultPluginId})
require.Error(err)
assert.Equal("db.Exec: deletion of system plugin not allowed: integrity violation: error #1104", err.Error())
assert.Equal(0, numDeleted)
// Create test plugin id
testPluginId := "pi_test1234"
numRows, err := rw.Exec(context.Background(), "insert into plugin(public_id) values (?)", []interface{}{testPluginId})
assert.NoError(err)
assert.Equal(1, numRows)
numDeleted, err = rw.Exec(context.Background(), "delete from plugin where public_id = ?", []interface{}{testPluginId})
assert.NoError(err)
assert.Equal(1, numDeleted)
})
}
+46
View File
@@ -0,0 +1,46 @@
// Package job provides a Job and a Run struct suitable for persisting to
// the repository.
//
// A Job represents work that should be run at a predetermined time and needs to be synchronized
// across servers to ensure that one and only one instance of a job is running at any given time.
// The job's name must be unique to the plugin that registered the job, the (plugin_id, name)
// uniqueness is enforced by the database.
//
// A Run represents a single execution of a job, only a single run can be in a
// running state for a specific job. The private_id (primary key) is generated by the
// repository during RunJobs.
//
// Repository
//
// A repository provides methods for creating, updating, retrieving, and
// deleting Jobs. It also provides methods to run jobs, update progress of runs
// and complete, fail or interrupt runs.
// A new repository should be created for each transaction. For example:
//
// var wrapper wrapping.Wrapper
// ... init wrapper ...
//
// // db implements both the reader and writer interfaces.
// db, _ := db.Open(db.Postgres, url)
//
// var repo *job.Repository
// repo, _ = job.NewRepository(db, db, wrapper)
//
// var j *job.Job
// j, _ = repo.CreateJob(context.Background(), j, "name", "description")
//
// var runs []*Run
// repo, _ = job.NewRepository(db, db, wrapper)
// runs, _ = repo.RunJobs(context.Background(), "serverId")
//
// ... run job ...
//
// var run *Run
// repo, _ = job.NewRepository(db, db, wrapper)
// run, _ = repo.UpdateProgress(ctx, run, []string{"TotalCount", "CompletedCount"})
//
// nextJobRun = time.Now().Add(time.Hour)
//
// repo, _ = job.NewRepository(db, db, wrapper)
// run, _ = repo.CompleteRun(ctx, run.PrivateId, job.Completed, nextJobRun)
package job
@@ -0,0 +1,90 @@
package job
import (
"context"
"testing"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/db/timestamp"
"github.com/hashicorp/boundary/internal/iam"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
)
func TestJobRun_ImmutableFields(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
wrapper := db.TestWrapper(t)
iam.TestRepo(t, conn, wrapper)
w := db.New(conn)
ts := timestamp.Timestamp{Timestamp: &timestamppb.Timestamp{Seconds: 0, Nanos: 0}}
job := testJob(t, conn, "testJob", "testDescription", wrapper)
server := testController(t, conn, wrapper)
oriRun, err := testRun(conn, job.PluginId, job.Name, server.PrivateId)
require.NoError(t, err)
tests := []struct {
name string
update *Run
fieldMask []string
}{
{
name: "private_id",
update: func() *Run {
j := oriRun.clone()
j.PrivateId = "new-run-id"
return j
}(),
fieldMask: []string{"PrivateId"},
},
{
name: "job pluginId",
update: func() *Run {
j := oriRun.clone()
j.JobPluginId = "new-job-plugin-id"
return j
}(),
fieldMask: []string{"JobPluginId"},
},
{
name: "job name",
update: func() *Run {
j := oriRun.clone()
j.JobName = "new-job-name"
return j
}(),
fieldMask: []string{"JobName"},
},
{
name: "create time",
update: func() *Run {
j := oriRun.clone()
j.CreateTime = &ts
return j
}(),
fieldMask: []string{"CreateTime"},
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
orig := oriRun.clone()
err := w.LookupById(context.Background(), orig)
require.NoError(err)
rowsUpdated, err := w.Update(context.Background(), tt.update, tt.fieldMask, nil, db.WithSkipVetForWrite(true))
require.Error(err)
assert.Equal(0, rowsUpdated)
after := oriRun.clone()
err = w.LookupById(context.Background(), after)
require.NoError(err)
assert.True(proto.Equal(orig, after))
})
}
}
+41
View File
@@ -0,0 +1,41 @@
package job
import (
"github.com/hashicorp/boundary/internal/scheduler/job/store"
"google.golang.org/protobuf/proto"
)
// Job represents work that should be run at a predetermined time and needs to be synchronized
// across servers to ensure that one and only one instance of a job is running at any given time.
type Job struct {
*store.Job
tableName string `gorm:"-"`
}
func (j *Job) clone() *Job {
cp := proto.Clone(j.Job)
return &Job{
Job: cp.(*store.Job),
tableName: j.tableName,
}
}
func allocJob() *Job {
return &Job{
Job: &store.Job{},
}
}
// TableName returns the table name for the job.
func (j *Job) TableName() string {
if j.tableName != "" {
return j.tableName
}
return "job"
}
// SetTableName sets the table name. If the caller attempts to
// set the name to "" the name will be reset to the default name "job".
func (j *Job) SetTableName(n string) {
j.tableName = n
}
+48
View File
@@ -0,0 +1,48 @@
package job
import (
"testing"
"github.com/hashicorp/boundary/internal/scheduler/job/store"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestJob_SetTableName(t *testing.T) {
t.Parallel()
defaultTableName := "job"
tests := []struct {
name string
initialName string
setNameTo string
want string
}{
{
name: "new-name",
initialName: "",
setNameTo: "new-name",
want: "new-name",
},
{
name: "reset to default",
initialName: "initial",
setNameTo: "",
want: defaultTableName,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
def := &Job{
Job: &store.Job{},
}
require.Equal(defaultTableName, def.TableName())
s := &Job{
Job: &store.Job{},
tableName: tt.initialName,
}
s.SetTableName(tt.setNameTo)
assert.Equal(tt.want, s.TableName())
})
}
}
+80
View File
@@ -0,0 +1,80 @@
package job
import (
"time"
)
const (
defaultRunJobsLimit = 1
defaultPluginId = "pi_system"
)
// getOpts - iterate the inbound Options and return a struct
func getOpts(opt ...Option) options {
opts := getDefaultOptions()
for _, o := range opt {
o(&opts)
}
return opts
}
// Option - how Options are passed as arguments
type Option func(*options)
// options = how options are represented
type options struct {
withNextRunIn time.Duration
withRunJobsLimit uint
withLimit int
withName string
withServerId string
}
func getDefaultOptions() options {
return options{
withRunJobsLimit: defaultRunJobsLimit,
}
}
// WithNextRunIn provides an option to provide the duration until the next run is scheduled.
// If this option is not provided the NextScheduledRun of the job will default to the
// current database time, and be available to run immediately.
func WithNextRunIn(d time.Duration) Option {
return func(o *options) {
o.withNextRunIn = d
}
}
// WithRunJobsLimit provides an option to provide the number of jobs to run.
// If WithRunJobsLimit == 0, then default run jobs limit is used.
func WithRunJobsLimit(l uint) 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.
func WithLimit(l int) Option {
return func(o *options) {
o.withLimit = l
}
}
// WithName provides an option to provide the name to match when calling ListJobs
func WithName(n string) Option {
return func(o *options) {
o.withName = n
}
}
// WithServerId provides an option to provide the server id to match when calling InterruptRuns
func WithServerId(id string) Option {
return func(o *options) {
o.withServerId = id
}
}
+60
View File
@@ -0,0 +1,60 @@
package job
import (
"testing"
"time"
"github.com/stretchr/testify/assert"
)
// Test_GetOpts provides unit tests for GetOpts and all the options
func Test_GetOpts(t *testing.T) {
t.Parallel()
t.Run("WithNextRunIn", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithNextRunIn(time.Hour))
testOpts := getDefaultOptions()
assert.NotEqual(opts, testOpts)
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(uint(defaultRunJobsLimit), opts.withRunJobsLimit)
})
t.Run("WithLimit", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithLimit(100))
testOpts := getDefaultOptions()
assert.NotEqual(opts, testOpts)
testOpts.withLimit = 100
assert.Equal(opts, testOpts)
})
t.Run("WithName", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithName("name"))
testOpts := getDefaultOptions()
assert.NotEqual(opts, testOpts)
testOpts.withName = "name"
assert.Equal(opts, testOpts)
})
t.Run("WithServerId", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithServerId("server_id"))
testOpts := getDefaultOptions()
assert.NotEqual(opts, testOpts)
testOpts.withServerId = "server_id"
assert.Equal(opts, testOpts)
})
}
+88
View File
@@ -0,0 +1,88 @@
package job
const runJobsQuery = `
insert into job_run (
job_plugin_id, job_name, server_id
)
select
job_plugin_id, job_name, ?
from job_jobs_to_run
order by next_scheduled_run asc
limit ?
returning *;
`
const createJobQuery = `
insert into job (
plugin_id, -- $1
name, -- $2
description, -- $3
next_scheduled_run -- $4
) values (
$1, -- plugin_id
$2, -- name
$3, -- description
wt_add_seconds_to_now($4) -- next_scheduled_run
)
returning *;
`
const setNextScheduledRunQuery = `
update
job
set
next_scheduled_run = wt_add_seconds_to_now(?)
where
plugin_id = ?
and name = ?
returning *;
`
const updateProgressQuery = `
update
job_run
set
completed_count = ?,
total_count = ?
where
private_id = ?
and status = 'running'
returning *;
`
const completeRunQuery = `
update
job_run
set
status = 'completed',
end_time = current_timestamp
where
private_id = ?
and status = 'running'
returning *;
`
const failRunQuery = `
update
job_run
set
status = 'failed',
end_time = current_timestamp
where
private_id = ?
and status = 'running'
returning *;
`
const interruptRunsQuery = `
update
job_run
set
status = 'interrupted',
end_time = current_timestamp
where
update_time <= wt_add_seconds_to_now(?)
and status = 'running'
%s
returning *;
`
+48
View File
@@ -0,0 +1,48 @@
package job
import (
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/kms"
)
// A Repository stores and retrieves the persistent types in the job
// package. It is not safe to use a repository concurrently.
type Repository struct {
reader db.Reader
writer db.Writer
kms *kms.Kms
// defaultLimit provides a default for limiting the number of results
// returned from the repo during a ListRequest
defaultLimit int
}
// NewRepository creates a new Repository. The returned repository should
// only be used for one transaction and it is not safe for concurrent go
// routines to access it. WithLimit option is used as a repo wide default
// limit applied to all ListX methods.
func NewRepository(r db.Reader, w db.Writer, kms *kms.Kms, opt ...Option) (*Repository, error) {
const op = "job.NewRepository"
switch {
case r == nil:
return nil, errors.New(errors.InvalidParameter, op, "missing db reader")
case w == nil:
return nil, errors.New(errors.InvalidParameter, op, "missing db writer")
case kms == nil:
return nil, errors.New(errors.InvalidParameter, op, "missing kms")
}
opts := getOpts(opt...)
if opts.withLimit == 0 {
// zero signals the boundary defaults should be used.
opts.withLimit = db.DefaultLimit
}
return &Repository{
reader: r,
writer: w,
kms: kms,
defaultLimit: opts.withLimit,
}, nil
}
+194
View File
@@ -0,0 +1,194 @@
package job
import (
"context"
"fmt"
"strings"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/errors"
)
// CreateJob inserts a job into the repository and returns a new *Job.
//
// • name must be provided and is the name of the job.
//
// • description must be provided and is the user-friendly description of the job.
//
// WithNextRunIn is the only valid options.
func (r *Repository) CreateJob(ctx context.Context, name, description string, opt ...Option) (*Job, error) {
const op = "job.(Repository).CreateJob"
if name == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing name")
}
if description == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing description")
}
opts := getOpts(opt...)
j := allocJob()
_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
func(r db.Reader, w db.Writer) error {
rows, err := r.Query(ctx, createJobQuery, []interface{}{
defaultPluginId,
name,
description,
int(opts.withNextRunIn.Round(time.Second).Seconds()),
})
if err != nil {
return errors.Wrap(err, op)
}
defer rows.Close()
var rowCnt int
for rows.Next() {
if rowCnt > 0 {
return errors.New(errors.MultipleRecords, op, "more than 1 job would have been created")
}
rowCnt++
err = r.ScanRows(rows, j)
if err != nil {
_ = rows.Close()
return errors.Wrap(err, op, errors.WithMsg("unable to scan rows for job"))
}
}
if rowCnt == 0 {
return errors.New(errors.NotSpecificIntegrity, op, "failed to create new job")
}
return nil
},
)
if err != nil {
if errors.IsUniqueError(err) {
return nil, errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("name %s already exists", name)))
}
return nil, errors.Wrap(err, op)
}
return j, nil
}
// UpdateJobNextRun updates the Job repository entry for the job name
// setting the job's NextScheduledRun to the current database time incremented by
// the nextRunIn parameter.
//
// All options are ignored.
func (r *Repository) UpdateJobNextRun(ctx context.Context, name string, nextRunIn time.Duration, _ ...Option) (*Job, error) {
const op = "job.(Repository).UpdateJobNextRun"
if name == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing name")
}
j := allocJob()
_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
func(r db.Reader, w db.Writer) error {
rows, err := w.Query(ctx, setNextScheduledRunQuery, []interface{}{int(nextRunIn.Round(time.Second).Seconds()), defaultPluginId, name})
if err != nil {
return errors.Wrap(err, op, errors.WithMsg(
fmt.Sprintf("failed to set next scheduled run time for job %v", name)))
}
defer rows.Close()
var rowCnt int
for rows.Next() {
if rowCnt > 0 {
return errors.New(errors.MultipleRecords, op, "more than 1 job would have been updated")
}
rowCnt++
err = r.ScanRows(rows, j)
if err != nil {
_ = rows.Close()
return errors.Wrap(err, op, errors.WithMsg("unable to scan rows"))
}
}
if rowCnt == 0 {
return errors.New(errors.RecordNotFound, op, fmt.Sprintf("job %q does not exist", name))
}
return nil
},
)
if err != nil {
return nil, errors.Wrap(err, op)
}
return j, nil
}
// LookupJob will look up a job in the repository using the job name. If the job is not
// found, it will return nil, nil.
//
// All options are ignored.
func (r *Repository) LookupJob(ctx context.Context, name string, _ ...Option) (*Job, error) {
const op = "job.(Repository).LookupJob"
if name == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing name")
}
j := allocJob()
if err := r.reader.LookupWhere(ctx, j, "name = ?", []interface{}{name}); err != nil {
if errors.IsNotFoundError(err) {
return nil, nil
}
return nil, errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("failed for %v", name)))
}
return j, nil
}
// ListJobs returns a slice of Jobs.
//
// WithName and WithLimit are the only valid options.
func (r *Repository) ListJobs(ctx context.Context, opt ...Option) ([]*Job, error) {
const op = "job.(Repository).ListJobs"
opts := getOpts(opt...)
limit := r.defaultLimit
if opts.withLimit != 0 {
// non-zero signals an override of the default limit for the repo.
limit = opts.withLimit
}
var args []interface{}
var where []string
if opts.withName != "" {
where, args = append(where, "name = ?"), append(args, opts.withName)
}
var jobs []*Job
err := r.reader.SearchWhere(ctx, &jobs, strings.Join(where, " and "), args, db.WithLimit(limit))
if err != nil {
return nil, errors.Wrap(err, op)
}
return jobs, nil
}
// deleteJob deletes the job for the provided job name from the repository
// returning a count of the number of records deleted.
//
// All options are ignored.
func (r *Repository) deleteJob(ctx context.Context, name string, _ ...Option) (int, error) {
const op = "job.(Repository).deleteJob"
if name == "" {
return db.NoRowsAffected, errors.New(errors.InvalidParameter, op, "missing name")
}
j := allocJob()
var rowsDeleted int
_, err := r.writer.DoTx(
ctx, db.StdRetryCnt, db.ExpBackoff{},
func(_ db.Reader, w db.Writer) (err error) {
rowsDeleted, err = w.Delete(ctx, j, db.WithWhere("name = ?", []interface{}{name}))
if err != nil {
return errors.Wrap(err, op)
}
if rowsDeleted > 1 {
return errors.New(errors.MultipleRecords, op, "more than 1 resource would have been deleted")
}
return nil
},
)
if err != nil {
return db.NoRowsAffected, errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("delete failed for %v", name)))
}
return rowsDeleted, nil
}
@@ -0,0 +1,425 @@
package job
import (
"context"
"fmt"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/scheduler/job/store"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/testing/protocmp"
)
func TestRepository_CreateJob(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kms := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
type args struct {
name, description string
}
tests := []struct {
name string
in args
want *Job
wantErr bool
wantErrCode errors.Code
wantErrMsg string
}{
{
name: "missing-name",
wantErr: true,
in: args{
description: "description",
},
wantErrCode: errors.InvalidParameter,
wantErrMsg: "job.(Repository).CreateJob: missing name: parameter violation: error #100",
},
{
name: "missing-description",
in: args{
name: "name",
},
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "job.(Repository).CreateJob: missing description: parameter violation: error #100",
},
{
name: "valid",
in: args{
name: "name",
description: "description",
},
want: &Job{
Job: &store.Job{
Name: "name",
Description: "description",
},
},
},
}
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(rw, rw, kms)
require.NoError(err)
require.NotNil(repo)
got, err := repo.CreateJob(context.Background(), tt.in.name, tt.in.description)
if tt.wantErr {
require.Error(err)
assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err)
assert.Equal(tt.wantErrMsg, err.Error())
return
}
require.NoError(err)
require.NotNil(got)
assert.NotSame(tt.in, got)
assert.Equal(tt.want.Name, got.Name)
assert.Equal(tt.want.Description, got.Description)
assert.Equal(defaultPluginId, got.PluginId)
assert.NotEmpty(got.NextScheduledRun)
})
}
t.Run("invalid-duplicate-names", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
repo, err := NewRepository(rw, rw, kms)
require.NoError(err)
require.NotNil(repo)
got, err := repo.CreateJob(context.Background(), "test-dup-name", "description")
require.NoError(err)
require.NotNil(got)
assert.Equal("test-dup-name", got.Name)
assert.Equal("description", got.Description)
got2, err := repo.CreateJob(context.Background(), "test-dup-name", "description")
require.Error(err)
assert.Truef(errors.Match(errors.T(errors.NotUnique), err), "want err code: %v got err: %v", errors.NotUnique, err)
assert.Nil(got2)
})
}
func TestRepository_LookupJob(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kms := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
job := testJob(t, conn, "name", "description", wrapper)
tests := []struct {
name string
in string
want *Job
wantErr bool
wantErrCode errors.Code
wantErrMsg string
}{
{
name: "with-no-name",
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "job.(Repository).LookupJob: missing name: parameter violation: error #100",
},
{
name: "with-non-existing-name",
in: "fake-name",
},
{
name: "with-existing-name",
in: job.Name,
want: job,
},
}
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(rw, rw, kms)
assert.NoError(err)
require.NotNil(repo)
got, err := repo.LookupJob(context.Background(), tt.in)
if tt.wantErr {
require.Error(err)
assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err)
assert.Equal(tt.wantErrMsg, err.Error())
assert.Nil(got)
return
}
require.NoError(err)
if tt.want == nil {
assert.Nil(got)
return
}
assert.NotEmpty(got.NextScheduledRun)
assert.Equal(tt.want.Name, got.Name)
assert.Equal(tt.want.Description, got.Description)
})
}
}
func TestRepository_deleteJob(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kms := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
job := testJob(t, conn, "name", "description", wrapper)
tests := []struct {
name string
in string
want int
wantErr bool
wantErrCode errors.Code
wantErrMsg string
}{
{
name: "With no name",
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "job.(Repository).deleteJob: missing name: parameter violation: error #100",
},
{
name: "With non existing name",
in: "fake-name",
want: 0,
},
{
name: "With existing name",
in: job.Name,
want: 1,
},
}
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(rw, rw, kms)
assert.NoError(err)
require.NotNil(repo)
got, err := repo.deleteJob(context.Background(), tt.in)
if tt.wantErr {
require.Error(err)
assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err)
assert.Equal(tt.wantErrMsg, err.Error())
assert.Zero(got)
return
}
require.NoError(err)
assert.EqualValues(tt.want, got)
})
}
}
func TestRepository_ListJobs(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kms := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
job1 := testJob(t, conn, "sameName", "description", wrapper)
job2 := testJob(t, conn, "differentName", "description", wrapper)
tests := []struct {
name string
opts []Option
want []*Job
}{
{
name: "no-options",
want: []*Job{job1, job2},
},
{
name: "with-same-name",
opts: []Option{
WithName("sameName"),
},
want: []*Job{job1},
},
{
name: "with-different-name",
opts: []Option{
WithName("differentName"),
},
want: []*Job{job2},
},
{
name: "with-fake-name",
opts: []Option{
WithName("fake-name"),
},
want: []*Job{},
},
}
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(rw, rw, kms)
assert.NoError(err)
require.NotNil(repo)
got, err := repo.ListJobs(context.Background(), tt.opts...)
require.NoError(err)
opts := []cmp.Option{
cmpopts.SortSlices(func(x, y *Job) bool { return x.Name < y.Name }),
protocmp.Transform(),
}
assert.Empty(cmp.Diff(tt.want, got, opts...))
})
}
}
func TestRepository_ListJobs_Limits(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kms := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
count := 10
jobs := make([]*Job, count)
for i := range jobs {
jobs[i] = testJob(t, conn, fmt.Sprintf("name-%d", i), "description", wrapper)
}
tests := []struct {
name string
repoOpts []Option
listOpts []Option
wantLen int
}{
{
name: "With no limits",
wantLen: count,
},
{
name: "With repo limit",
repoOpts: []Option{WithLimit(3)},
wantLen: 3,
},
{
name: "With negative repo limit",
repoOpts: []Option{WithLimit(-1)},
wantLen: count,
},
{
name: "With List limit",
listOpts: []Option{WithLimit(3)},
wantLen: 3,
},
{
name: "With negative List limit",
listOpts: []Option{WithLimit(-1)},
wantLen: count,
},
{
name: "With repo smaller than list limit",
repoOpts: []Option{WithLimit(2)},
listOpts: []Option{WithLimit(6)},
wantLen: 6,
},
{
name: "With repo larger than list limit",
repoOpts: []Option{WithLimit(6)},
listOpts: []Option{WithLimit(2)},
wantLen: 2,
},
}
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(rw, rw, kms, tt.repoOpts...)
assert.NoError(err)
require.NotNil(repo)
got, err := repo.ListJobs(context.Background(), tt.listOpts...)
require.NoError(err)
assert.Len(got, tt.wantLen)
})
}
}
func TestRepository_UpdateJobNextRun(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kmsCache := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
t.Run("valid", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
repo, err := NewRepository(rw, rw, kmsCache)
require.NoError(err)
require.NotNil(repo)
job, err := repo.CreateJob(context.Background(), "name", "description")
require.NoError(err)
got, err := repo.UpdateJobNextRun(context.Background(), job.Name, time.Hour)
require.NoError(err)
require.NotNil(got)
previousRunAt := job.NextScheduledRun.Timestamp.GetSeconds()
nextRunAt := got.NextScheduledRun.Timestamp.GetSeconds()
assert.True(nextRunAt >= previousRunAt+int64(time.Hour.Seconds()),
fmt.Sprintf("expected next run (%d) to be greater than or equal to the previous run (%d)",
nextRunAt, previousRunAt))
// update NextScheduledRun to pass equality check
job.NextScheduledRun = got.NextScheduledRun
assert.Equal(job, got)
})
t.Run("no-name", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
repo, err := NewRepository(rw, rw, kmsCache)
require.NoError(err)
require.NotNil(repo)
got, err := repo.UpdateJobNextRun(context.Background(), "", time.Hour)
require.Error(err)
require.Nil(got)
assert.Truef(errors.Match(errors.T(errors.InvalidParameter), err), "Unexpected error %s", err)
assert.Equal("job.(Repository).UpdateJobNextRun: missing name: parameter violation: error #100", err.Error())
})
t.Run("job-not-found", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
repo, err := NewRepository(rw, rw, kmsCache)
require.NoError(err)
require.NotNil(repo)
got, err := repo.UpdateJobNextRun(context.Background(), "fake-name", time.Hour)
require.Error(err)
require.Nil(got)
assert.Truef(errors.Match(errors.T(errors.RecordNotFound), err), "Unexpected error %s", err)
assert.Equal("job.(Repository).UpdateJobNextRun: db.DoTx: job.(Repository).UpdateJobNextRun: job \"fake-name\" does not exist: search issue: error #1100", err.Error())
})
}
+343
View File
@@ -0,0 +1,343 @@
package job
import (
"context"
"fmt"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/errors"
)
// RunJobs queries the job repository for jobs that need to be run. It creates new entries
// for each job that needs to be run in the job_run repository, returning a slice of *Run.
// 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) {
const op = "job.(Repository).RunJobs"
if serverId == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing server id")
}
opts := getOpts(opt...)
var runs []*Run
_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
func(r db.Reader, w db.Writer) error {
rows, err := r.Query(ctx, runJobsQuery, []interface{}{serverId, opts.withRunJobsLimit})
if err != nil {
return errors.Wrap(err, op)
}
defer rows.Close()
for rows.Next() {
run := allocRun()
err = r.ScanRows(rows, run)
if err != nil {
return errors.Wrap(err, op, errors.WithMsg("unable to scan rows for job run"))
}
runs = append(runs, run)
}
return nil
},
)
if err != nil {
return nil, errors.Wrap(err, op)
}
return runs, nil
}
// UpdateProgress updates the repository entry's completed and total counts for the provided runId.
//
// Once a run has been persisted with a final run status (completed, failed or interrupted),
// any future UpdateProgress attempts will return an error with Code errors.InvalidJobRunState.
// All options are ignored.
func (r *Repository) UpdateProgress(ctx context.Context, runId string, completed, total int, _ ...Option) (*Run, error) {
const op = "job.(Repository).UpdateProgress"
if runId == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing run id")
}
run := allocRun()
run.PrivateId = runId
_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
func(r db.Reader, w db.Writer) error {
rows, err := r.Query(ctx, updateProgressQuery, []interface{}{completed, total, runId})
if err != nil {
return errors.Wrap(err, op)
}
defer rows.Close()
var rowCnt int
for rows.Next() {
if rowCnt > 0 {
return errors.New(errors.MultipleRecords, op, "more than 1 job run would have been updated")
}
rowCnt++
err = r.ScanRows(rows, run)
if err != nil {
return errors.Wrap(err, op, errors.WithMsg("unable to scan rows for job run"))
}
}
if rowCnt == 0 {
// Failed to update run, either it does not exist or was in an invalid state
if err = r.LookupById(ctx, run); err != nil {
if errors.IsNotFoundError(err) {
return errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("job run %q does not exist", runId)))
}
return errors.Wrap(err, op)
}
return errors.New(errors.InvalidJobRunState, op, fmt.Sprintf("job run was in a final run state: %v", run.Status))
}
return nil
},
)
if err != nil {
return nil, errors.Wrap(err, op)
}
return run, nil
}
// CompleteRun updates the Run repository entry for the provided runId.
// It sets the status to 'completed' and updates the run's EndTime to the current database time.
// CompleteRun also updates the Job repository entry that is associated with this run,
// setting the job's NextScheduledRun to the current database time incremented by the nextRunIn
// parameter.
//
// Once a run has been persisted with a final run status (completed, failed
// or interrupted), any future calls to CompleteRun will return an error with Code
// errors.InvalidJobRunState.
// All options are ignored.
func (r *Repository) CompleteRun(ctx context.Context, runId string, nextRunIn time.Duration, _ ...Option) (*Run, error) {
const op = "job.(Repository).CompleteRun"
if runId == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing run id")
}
run := allocRun()
run.PrivateId = runId
_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
func(r db.Reader, w db.Writer) error {
rows, err := r.Query(ctx, completeRunQuery, []interface{}{runId})
if err != nil {
return errors.Wrap(err, op)
}
defer rows.Close()
var rowCnt int
for rows.Next() {
if rowCnt > 0 {
return errors.New(errors.MultipleRecords, op, "more than 1 job run would have been updated")
}
rowCnt++
err = r.ScanRows(rows, run)
if err != nil {
return errors.Wrap(err, op, errors.WithMsg("unable to scan rows for job run"))
}
}
if rowCnt == 0 {
// Failed to update run, either it does not exist or was in an invalid state
if err = r.LookupById(ctx, run); err != nil {
if errors.IsNotFoundError(err) {
return errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("job run %q does not exist", runId)))
}
return errors.Wrap(err, op)
}
return errors.New(errors.InvalidJobRunState, op, fmt.Sprintf("job run was in a final run state: %v", run.Status))
}
rows1, err := r.Query(ctx, setNextScheduledRunQuery, []interface{}{int(nextRunIn.Round(time.Second).Seconds()), run.JobPluginId, run.JobName})
if err != nil {
return errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("failed to set next scheduled run time for job: %s", run.JobName)))
}
defer rows1.Close()
job := allocJob()
rowCnt = 0
for rows1.Next() {
if rowCnt > 0 {
return errors.New(errors.MultipleRecords, op, "more than 1 job would have been updated")
}
rowCnt++
err = r.ScanRows(rows1, job)
if err != nil {
return errors.Wrap(err, op, errors.WithMsg("unable to scan rows for job"))
}
}
return nil
},
)
if err != nil {
return nil, errors.Wrap(err, op)
}
return run, nil
}
// FailRun updates the Run repository entry for the provided runId.
// It sets the status to 'failed' and updates the run's EndTime to the current database time.
//
// Once a run has been persisted with a final run status (completed, failed
// or interrupted), any future calls to FailRun will return an error with Code
// errors.InvalidJobRunState.
// All options are ignored.
func (r *Repository) FailRun(ctx context.Context, runId string, _ ...Option) (*Run, error) {
const op = "job.(Repository).FailRun"
if runId == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing run id")
}
run := allocRun()
run.PrivateId = runId
_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
func(r db.Reader, w db.Writer) error {
rows, err := r.Query(ctx, failRunQuery, []interface{}{runId})
if err != nil {
return errors.Wrap(err, op)
}
defer rows.Close()
var rowCnt int
for rows.Next() {
if rowCnt > 0 {
return errors.New(errors.MultipleRecords, op, "more than 1 job run would have been updated")
}
rowCnt++
err = r.ScanRows(rows, run)
if err != nil {
return errors.Wrap(err, op, errors.WithMsg("unable to scan rows for job run"))
}
}
if rowCnt == 0 {
// Failed to update run, either it does not exist or was in an invalid state
if err = r.LookupById(ctx, run); err != nil {
if errors.IsNotFoundError(err) {
return errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("job run %q does not exist", runId)))
}
return errors.Wrap(err, op)
}
return errors.New(errors.InvalidJobRunState, op, fmt.Sprintf("job run was in a final run state: %v", run.Status))
}
return nil
},
)
if err != nil {
return nil, errors.Wrap(err, op)
}
return run, nil
}
// InterruptRuns updates the Run repository entries for all job runs that have not been
// updated for the provided interruptThreshold. It sets the status to 'interrupted' and
// updates the run's EndTime to the current database time.
//
// Once a run has been persisted with a final run status (completed, failed
// or interrupted), any future calls to InterruptRuns will return an error with Code
// errors.InvalidJobRunState.
// WithServerId is the only valid option
func (r *Repository) InterruptRuns(ctx context.Context, interruptThreshold time.Duration, opt ...Option) ([]*Run, error) {
const op = "job.(Repository).InterruptRuns"
opts := getOpts(opt...)
// interruptThreshold is seconds in past so * -1
args := []interface{}{-1 * int(interruptThreshold.Round(time.Second).Seconds())}
var whereServerId string
if opts.withServerId != "" {
whereServerId = "and server_id = ?"
args = append(args, opts.withServerId)
}
query := fmt.Sprintf(interruptRunsQuery, whereServerId)
var runs []*Run
_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},
func(r db.Reader, w db.Writer) error {
rows, err := r.Query(ctx, query, args)
if err != nil {
return errors.Wrap(err, op)
}
defer rows.Close()
for rows.Next() {
run := allocRun()
err = r.ScanRows(rows, run)
if err != nil {
return errors.Wrap(err, op, errors.WithMsg("unable to scan rows for job run"))
}
runs = append(runs, run)
}
return nil
},
)
if err != nil {
return nil, errors.Wrap(err, op)
}
return runs, nil
}
// LookupRun will look up a run in the repository using the runId. If the run is not
// found, it will return nil, nil.
//
// All options are ignored.
func (r *Repository) LookupRun(ctx context.Context, runId string, _ ...Option) (*Run, error) {
const op = "job.(Repository).LookupRun"
if runId == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing run id")
}
run := allocRun()
run.PrivateId = runId
if err := r.reader.LookupById(ctx, run); err != nil {
if errors.IsNotFoundError(err) {
return nil, nil
}
return nil, errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("failed for %s", runId)))
}
return run, nil
}
// deleteRun deletes the job for the provided runId from the repository
// returning a count of the number of records deleted.
//
// All options are ignored.
func (r *Repository) deleteRun(ctx context.Context, runId string, _ ...Option) (int, error) {
const op = "job.(Repository).deleteRun"
if runId == "" {
return db.NoRowsAffected, errors.New(errors.InvalidParameter, op, "missing run id")
}
run := allocRun()
run.PrivateId = runId
var rowsDeleted int
_, err := r.writer.DoTx(
ctx, db.StdRetryCnt, db.ExpBackoff{},
func(_ db.Reader, w db.Writer) (err error) {
rowsDeleted, err = w.Delete(ctx, run)
if err != nil {
return errors.Wrap(err, op)
}
if rowsDeleted > 1 {
return errors.New(errors.MultipleRecords, op, "more than 1 resource would have been deleted")
}
return nil
},
)
if err != nil {
return db.NoRowsAffected, errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("delete failed for %s", runId)))
}
return rowsDeleted, nil
}
File diff suppressed because it is too large. Load diff
+131
View File
@@ -0,0 +1,131 @@
package job
import (
"testing"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/kms"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestRepository_New(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kmsCache := kms.TestKms(t, conn, wrapper)
type args struct {
r db.Reader
w db.Writer
kms *kms.Kms
}
tests := []struct {
name string
args args
opts []Option
want *Repository
wantErr bool
wantErrCode errors.Code
wantErrMsg string
}{
{
name: "valid",
args: args{
r: rw,
w: rw,
kms: kmsCache,
},
want: &Repository{
reader: rw,
writer: rw,
kms: kmsCache,
defaultLimit: db.DefaultLimit,
},
},
{
name: "valid-with-limit",
args: args{
r: rw,
w: rw,
kms: kmsCache,
},
opts: []Option{
WithLimit(10),
},
want: &Repository{
reader: rw,
writer: rw,
kms: kmsCache,
defaultLimit: 10,
},
},
{
name: "nil-reader",
args: args{
r: nil,
w: rw,
kms: kmsCache,
},
want: nil,
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "job.NewRepository: missing db reader: parameter violation: error #100",
},
{
name: "nil-writer",
args: args{
r: rw,
w: nil,
kms: kmsCache,
},
want: nil,
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "job.NewRepository: missing db writer: parameter violation: error #100",
},
{
name: "nil-kms",
args: args{
r: rw,
w: rw,
kms: nil,
},
want: nil,
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "job.NewRepository: missing kms: parameter violation: error #100",
},
{
name: "all-nils",
args: args{
r: nil,
w: nil,
kms: nil,
},
want: nil,
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "job.NewRepository: missing db reader: parameter violation: error #100",
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
got, err := NewRepository(tt.args.r, tt.args.w, tt.args.kms, tt.opts...)
if tt.wantErr {
require.Error(err)
assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err)
assert.Equal(tt.wantErrMsg, err.Error())
return
}
assert.NoError(err)
assert.NotNil(got)
assert.Equal(tt.want, got)
})
}
}
+40
View File
@@ -0,0 +1,40 @@
package job
import (
"github.com/hashicorp/boundary/internal/scheduler/job/store"
"google.golang.org/protobuf/proto"
)
// Run represents an instance of a job that is either actively running or has already completed.
type Run struct {
*store.JobRun
tableName string `gorm:"-"`
}
func (j *Run) clone() *Run {
cp := proto.Clone(j.JobRun)
return &Run{
JobRun: cp.(*store.JobRun),
tableName: j.tableName,
}
}
func allocRun() *Run {
return &Run{
JobRun: &store.JobRun{},
}
}
// TableName returns the table name for the job run.
func (j *Run) TableName() string {
if j.tableName != "" {
return j.tableName
}
return "job_run"
}
// SetTableName sets the table name. If the caller attempts to
// set the name to "" the name will be reset to the default name "job_run".
func (j *Run) SetTableName(n string) {
j.tableName = n
}
+48
View File
@@ -0,0 +1,48 @@
package job
import (
"testing"
"github.com/hashicorp/boundary/internal/scheduler/job/store"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestRun_SetTableName(t *testing.T) {
t.Parallel()
defaultTableName := "job_run"
tests := []struct {
name string
initialName string
setNameTo string
want string
}{
{
name: "new-name",
initialName: "",
setNameTo: "new-name",
want: "new-name",
},
{
name: "reset to default",
initialName: "initial",
setNameTo: "",
want: defaultTableName,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
def := &Run{
JobRun: &store.JobRun{},
}
require.Equal(defaultTableName, def.TableName())
s := &Run{
JobRun: &store.JobRun{},
tableName: tt.initialName,
}
s.SetTableName(tt.setNameTo)
assert.Equal(tt.want, s.TableName())
})
}
}
+22
View File
@@ -0,0 +1,22 @@
package job
type Status string
const (
// Running represents that the job run is actively running on a server
Running Status = "running"
// Completed represents that the job run has successfully finished
Completed Status = "completed"
// Failed represent that the job run had an error during execution
Failed Status = "failed"
// Interrupted represents that the job run was interrupted by a server
// other than the server running the job.
Interrupted Status = "interrupted"
)
func (s Status) string() string {
return string(s)
}
+385
View File
@@ -0,0 +1,385 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.26.0
// protoc v3.15.8
// source: controller/storage/job/store/v1/job.proto
// Package store provides protobufs for storing types in the job package.
package store
import (
timestamp "github.com/hashicorp/boundary/internal/db/timestamp"
_ "github.com/hashicorp/boundary/internal/gen/controller/protooptions"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
type Job struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
// plugin_id is the primary key of the plugin that registered and owns the job
// @inject_tag: `gorm:"primary_key"`
PluginId string `protobuf:"bytes,1,opt,name=plugin_id,json=pluginId,proto3" json:"plugin_id,omitempty" gorm:"primary_key"`
// The name of the job.
// @inject_tag: `gorm:"primary_key"`
Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty" gorm:"primary_key"`
// The human friendly description of the job.
// @inject_tag: `gorm:"not_null"`
Description string `protobuf:"bytes,3,opt,name=description,proto3" json:"description,omitempty" gorm:"not_null"`
// next_scheduled_run is the time that the next run should be created.
// @inject_tag: `gorm:"default:current_timestamp"`
NextScheduledRun *timestamp.Timestamp `protobuf:"bytes,4,opt,name=next_scheduled_run,json=nextScheduledRun,proto3" json:"next_scheduled_run,omitempty" gorm:"default:current_timestamp"`
}
func (x *Job) Reset() {
*x = Job{}
if protoimpl.UnsafeEnabled {
mi := &file_controller_storage_job_store_v1_job_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *Job) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Job) ProtoMessage() {}
func (x *Job) ProtoReflect() protoreflect.Message {
mi := &file_controller_storage_job_store_v1_job_proto_msgTypes[0]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Job.ProtoReflect.Descriptor instead.
func (*Job) Descriptor() ([]byte, []int) {
return file_controller_storage_job_store_v1_job_proto_rawDescGZIP(), []int{0}
}
func (x *Job) GetPluginId() string {
if x != nil {
return x.PluginId
}
return ""
}
func (x *Job) GetName() string {
if x != nil {
return x.Name
}
return ""
}
func (x *Job) GetDescription() string {
if x != nil {
return x.Description
}
return ""
}
func (x *Job) GetNextScheduledRun() *timestamp.Timestamp {
if x != nil {
return x.NextScheduledRun
}
return nil
}
type JobRun struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
// private_id is used to access the job run but not intended to be available via the API
// @inject_tag: `gorm:"primary_key"`
PrivateId string `protobuf:"bytes,1,opt,name=private_id,json=privateId,proto3" json:"private_id,omitempty" gorm:"primary_key"`
// The job_plugin_id is the id of the plugin that owns the job.
// @inject_tag: `gorm:"not_null"`
JobPluginId string `protobuf:"bytes,2,opt,name=job_plugin_id,json=jobPluginId,proto3" json:"job_plugin_id,omitempty" gorm:"not_null"`
// The job_name is the name of the job and must be set.
// @inject_tag: `gorm:"not_null"`
JobName string `protobuf:"bytes,3,opt,name=job_name,json=jobName,proto3" json:"job_name,omitempty" gorm:"not_null"`
// The server_id of the server running the job and must be set.
// @inject_tag: `gorm:"not_null"`
ServerId string `protobuf:"bytes,4,opt,name=server_id,json=serverId,proto3" json:"server_id,omitempty" gorm:"not_null"`
// The create_time is set by the database.
// @inject_tag: `gorm:"default:current_timestamp"`
CreateTime *timestamp.Timestamp `protobuf:"bytes,5,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:current_timestamp"`
// The update_time is set by the database.
// @inject_tag: `gorm:"default:current_timestamp"`
UpdateTime *timestamp.Timestamp `protobuf:"bytes,6,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty" gorm:"default:current_timestamp"`
// The end_time is set when the job run completes.
// @inject_tag: `gorm:"default:null"`
EndTime *timestamp.Timestamp `protobuf:"bytes,7,opt,name=end_time,json=endTime,proto3" json:"end_time,omitempty" gorm:"default:null"`
// completed_count is set during an update to indicate the current progress.
// @inject_tag: `gorm:"default:0"`
CompletedCount uint32 `protobuf:"varint,8,opt,name=completed_count,json=completedCount,proto3" json:"completed_count,omitempty" gorm:"default:0"`
// total_count is set during an update to indicate the progress goal.
// @inject_tag: `gorm:"default:0"`
TotalCount uint32 `protobuf:"varint,9,opt,name=total_count,json=totalCount,proto3" json:"total_count,omitempty" gorm:"default:0"`
// status of the job run (running, completed, failed or interrupted).
// @inject_tag: `gorm:"not_null"`
Status string `protobuf:"bytes,10,opt,name=status,proto3" json:"status,omitempty" gorm:"not_null"`
}
func (x *JobRun) Reset() {
*x = JobRun{}
if protoimpl.UnsafeEnabled {
mi := &file_controller_storage_job_store_v1_job_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *JobRun) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*JobRun) ProtoMessage() {}
func (x *JobRun) ProtoReflect() protoreflect.Message {
mi := &file_controller_storage_job_store_v1_job_proto_msgTypes[1]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use JobRun.ProtoReflect.Descriptor instead.
func (*JobRun) Descriptor() ([]byte, []int) {
return file_controller_storage_job_store_v1_job_proto_rawDescGZIP(), []int{1}
}
func (x *JobRun) GetPrivateId() string {
if x != nil {
return x.PrivateId
}
return ""
}
func (x *JobRun) GetJobPluginId() string {
if x != nil {
return x.JobPluginId
}
return ""
}
func (x *JobRun) GetJobName() string {
if x != nil {
return x.JobName
}
return ""
}
func (x *JobRun) GetServerId() string {
if x != nil {
return x.ServerId
}
return ""
}
func (x *JobRun) GetCreateTime() *timestamp.Timestamp {
if x != nil {
return x.CreateTime
}
return nil
}
func (x *JobRun) GetUpdateTime() *timestamp.Timestamp {
if x != nil {
return x.UpdateTime
}
return nil
}
func (x *JobRun) GetEndTime() *timestamp.Timestamp {
if x != nil {
return x.EndTime
}
return nil
}
func (x *JobRun) GetCompletedCount() uint32 {
if x != nil {
return x.CompletedCount
}
return 0
}
func (x *JobRun) GetTotalCount() uint32 {
if x != nil {
return x.TotalCount
}
return 0
}
func (x *JobRun) GetStatus() string {
if x != nil {
return x.Status
}
return ""
}
var File_controller_storage_job_store_v1_job_proto protoreflect.FileDescriptor
var file_controller_storage_job_store_v1_job_proto_rawDesc = []byte{
0x0a, 0x29, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f,
0x72, 0x61, 0x67, 0x65, 0x2f, 0x6a, 0x6f, 0x62, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2f, 0x76,
0x31, 0x2f, 0x6a, 0x6f, 0x62, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x1f, 0x63, 0x6f, 0x6e,
0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e,
0x6a, 0x6f, 0x62, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x1a, 0x2f, 0x63, 0x6f,
0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65,
0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2f, 0x76, 0x31, 0x2f, 0x74, 0x69,
0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x2a, 0x63,
0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x63, 0x75, 0x73, 0x74, 0x6f, 0x6d,
0x5f, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2f, 0x76, 0x31, 0x2f, 0x6f, 0x70, 0x74, 0x69,
0x6f, 0x6e, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xb2, 0x01, 0x0a, 0x03, 0x4a, 0x6f,
0x62, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x01,
0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x49, 0x64, 0x12, 0x12,
0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61,
0x6d, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f,
0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70,
0x74, 0x69, 0x6f, 0x6e, 0x12, 0x58, 0x0a, 0x12, 0x6e, 0x65, 0x78, 0x74, 0x5f, 0x73, 0x63, 0x68,
0x65, 0x64, 0x75, 0x6c, 0x65, 0x64, 0x5f, 0x72, 0x75, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b,
0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74,
0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e,
0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x10, 0x6e, 0x65,
0x78, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x64, 0x52, 0x75, 0x6e, 0x22, 0xc6,
0x03, 0x0a, 0x06, 0x4a, 0x6f, 0x62, 0x52, 0x75, 0x6e, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x72, 0x69,
0x76, 0x61, 0x74, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x70,
0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x49, 0x64, 0x12, 0x22, 0x0a, 0x0d, 0x6a, 0x6f, 0x62, 0x5f,
0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
0x0b, 0x6a, 0x6f, 0x62, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x49, 0x64, 0x12, 0x19, 0x0a, 0x08,
0x6a, 0x6f, 0x62, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07,
0x6a, 0x6f, 0x62, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x73, 0x65, 0x72, 0x76, 0x65,
0x72, 0x5f, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x73, 0x65, 0x72, 0x76,
0x65, 0x72, 0x49, 0x64, 0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74,
0x69, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74,
0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74,
0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65,
0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d,
0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65,
0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c,
0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65,
0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61,
0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x45,
0x0a, 0x08, 0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b,
0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74,
0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e,
0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x07, 0x65, 0x6e,
0x64, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x27, 0x0a, 0x0f, 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74,
0x65, 0x64, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e,
0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1f,
0x0a, 0x0b, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x09, 0x20,
0x01, 0x28, 0x0d, 0x52, 0x0a, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12,
0x16, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52,
0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x42, 0x42, 0x5a, 0x40, 0x67, 0x69, 0x74, 0x68, 0x75,
0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f,
0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61,
0x6c, 0x2f, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x2f, 0x6a, 0x6f, 0x62, 0x2f,
0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f,
0x74, 0x6f, 0x33,
}
var (
file_controller_storage_job_store_v1_job_proto_rawDescOnce sync.Once
file_controller_storage_job_store_v1_job_proto_rawDescData = file_controller_storage_job_store_v1_job_proto_rawDesc
)
func file_controller_storage_job_store_v1_job_proto_rawDescGZIP() []byte {
file_controller_storage_job_store_v1_job_proto_rawDescOnce.Do(func() {
file_controller_storage_job_store_v1_job_proto_rawDescData = protoimpl.X.CompressGZIP(file_controller_storage_job_store_v1_job_proto_rawDescData)
})
return file_controller_storage_job_store_v1_job_proto_rawDescData
}
var file_controller_storage_job_store_v1_job_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
var file_controller_storage_job_store_v1_job_proto_goTypes = []interface{}{
(*Job)(nil), // 0: controller.storage.job.store.v1.Job
(*JobRun)(nil), // 1: controller.storage.job.store.v1.JobRun
(*timestamp.Timestamp)(nil), // 2: controller.storage.timestamp.v1.Timestamp
}
var file_controller_storage_job_store_v1_job_proto_depIdxs = []int32{
2, // 0: controller.storage.job.store.v1.Job.next_scheduled_run:type_name -> controller.storage.timestamp.v1.Timestamp
2, // 1: controller.storage.job.store.v1.JobRun.create_time:type_name -> controller.storage.timestamp.v1.Timestamp
2, // 2: controller.storage.job.store.v1.JobRun.update_time:type_name -> controller.storage.timestamp.v1.Timestamp
2, // 3: controller.storage.job.store.v1.JobRun.end_time:type_name -> controller.storage.timestamp.v1.Timestamp
4, // [4:4] is the sub-list for method output_type
4, // [4:4] is the sub-list for method input_type
4, // [4:4] is the sub-list for extension type_name
4, // [4:4] is the sub-list for extension extendee
0, // [0:4] is the sub-list for field type_name
}
func init() { file_controller_storage_job_store_v1_job_proto_init() }
func file_controller_storage_job_store_v1_job_proto_init() {
if File_controller_storage_job_store_v1_job_proto != nil {
return
}
if !protoimpl.UnsafeEnabled {
file_controller_storage_job_store_v1_job_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Job); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_controller_storage_job_store_v1_job_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*JobRun); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_controller_storage_job_store_v1_job_proto_rawDesc,
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_controller_storage_job_store_v1_job_proto_goTypes,
DependencyIndexes: file_controller_storage_job_store_v1_job_proto_depIdxs,
MessageInfos: file_controller_storage_job_store_v1_job_proto_msgTypes,
}.Build()
File_controller_storage_job_store_v1_job_proto = out.File
file_controller_storage_job_store_v1_job_proto_rawDesc = nil
file_controller_storage_job_store_v1_job_proto_goTypes = nil
file_controller_storage_job_store_v1_job_proto_depIdxs = nil
}
+108
View File
@@ -0,0 +1,108 @@
package job
import (
"context"
"fmt"
"testing"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/servers"
wrapping "github.com/hashicorp/go-kms-wrapping"
"github.com/hashicorp/go-uuid"
"github.com/jinzhu/gorm"
"github.com/stretchr/testify/require"
)
func testJob(t *testing.T, conn *gorm.DB, name, description string, wrapper wrapping.Wrapper, opt ...Option) *Job {
t.Helper()
require := require.New(t)
rw := db.New(conn)
kms := kms.TestKms(t, conn, wrapper)
repo, err := NewRepository(rw, rw, kms)
require.NoError(err)
job, err := repo.CreateJob(context.Background(), name, description, opt...)
require.NoError(err)
require.NotNil(job)
return job
}
func testRun(conn *gorm.DB, pluginId, name, cId string) (*Run, error) {
query := `
insert into job_run (
job_plugin_id, job_name, server_id
)
values (?,?,?)
returning *;
`
rw := db.New(conn)
run := allocRun()
rows, err := rw.Query(context.Background(), query, []interface{}{pluginId, name, cId})
if err != nil {
return nil, err
}
if !rows.Next() {
return nil, fmt.Errorf("expected to rows")
}
err = rw.ScanRows(rows, run)
if err != nil {
return nil, err
}
_ = rows.Close()
return run, nil
}
func testRunWithUpdateTime(conn *gorm.DB, pluginId, name, cId string, updateTime time.Time) (*Run, error) {
query := `
insert into job_run (
job_plugin_id, job_name, server_id, update_time
)
values (?,?,?,?)
returning *;
`
rw := db.New(conn)
run := allocRun()
rows, err := rw.Query(context.Background(), query, []interface{}{pluginId, name, cId, updateTime})
if err != nil {
return nil, err
}
if !rows.Next() {
return nil, fmt.Errorf("expected to rows")
}
err = rw.ScanRows(rows, run)
if err != nil {
return nil, err
}
_ = rows.Close()
return run, nil
}
func testController(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper) *servers.Server {
t.Helper()
rw := db.New(conn)
kms := kms.TestKms(t, conn, wrapper)
serversRepo, err := servers.NewRepository(rw, rw, kms)
require.NoError(t, err)
id, err := uuid.GenerateUUID()
require.NoError(t, err)
controller := &servers.Server{
PrivateId: "test-job-server-" + id,
Type: servers.ServerTypeController.String(),
Description: "Test Job Controller",
Address: "127.0.0.1",
}
_, _, err = serversRepo.UpsertServer(context.Background(), controller)
require.NoError(t, err)
return controller
}
+80
View File
@@ -0,0 +1,80 @@
package job
import (
"context"
"fmt"
"testing"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/iam"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func Test_TestJob(t *testing.T) {
t.Parallel()
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
wrapper := db.TestWrapper(t)
iam.TestRepo(t, conn, wrapper)
job := testJob(t, conn, "testJob", "testDescription", wrapper)
require.NotNil(job)
rw := db.New(conn)
var got Job
err := rw.LookupWhere(context.Background(), &got, "name = ?", job.Name)
require.NoError(err)
assert.Equal("testJob", got.Name)
assert.Equal("testDescription", got.Description)
assert.Equal(defaultPluginId, got.PluginId)
assert.NotEmpty(got.NextScheduledRun)
job1 := testJob(t, conn, "testJob1", "testDescription1", wrapper, WithNextRunIn(time.Hour))
require.NotNil(job1)
var got1 Job
err = rw.LookupWhere(context.Background(), &got1, "name = ?", job1.Name)
require.NoError(err)
assert.Equal("testJob1", got1.Name)
assert.Equal("testDescription1", got1.Description)
assert.Equal(defaultPluginId, got.PluginId)
assert.NotEmpty(got1.NextScheduledRun)
// The previous job next scheduled run should have been created with the current database time,
// while the current job should be an hour later
previousJobNextRun := got.NextScheduledRun.Timestamp.GetSeconds()
currentJobNextRun := got1.NextScheduledRun.Timestamp.GetSeconds()
nextRunIn := int64(time.Hour.Seconds())
assert.True(currentJobNextRun >= previousJobNextRun+nextRunIn,
fmt.Sprintf("expected next run (%d) to be greater than or equal to the previous run end time (%d) incremented by %d",
currentJobNextRun, previousJobNextRun, nextRunIn))
}
func Test_TestRun(t *testing.T) {
t.Parallel()
assert, require := assert.New(t), require.New(t)
conn, _ := db.TestSetup(t, "postgres")
wrapper := db.TestWrapper(t)
iam.TestRepo(t, conn, wrapper)
job := testJob(t, conn, "testJob", "testDescription", wrapper)
require.NotNil(job)
server := testController(t, conn, wrapper)
run, err := testRun(conn, job.PluginId, job.Name, server.PrivateId)
require.NoError(err)
require.NotNil(run)
rw := db.New(conn)
var got Run
err = rw.LookupWhere(context.Background(), &got, "private_id = ?", run.PrivateId)
require.NoError(err)
assert.Equal(server.PrivateId, got.ServerId)
assert.Equal(job.Name, got.JobName)
assert.Equal(job.PluginId, got.JobPluginId)
assert.NotEmpty(got.CreateTime)
assert.Equal(Running.string(), got.Status)
}
+87
View File
@@ -0,0 +1,87 @@
package scheduler
import "time"
const (
defaultRunJobsLimit = 1
defaultRunJobsInterval = time.Minute
defaultMonitorInterval = 30 * time.Second
defaultInterruptThreshold = 5 * time.Minute
)
// getOpts - iterate the inbound Options and return a struct
func getOpts(opt ...Option) options {
opts := getDefaultOptions()
for _, o := range opt {
o(&opts)
}
return opts
}
// Option - how Options are passed as arguments
type Option func(*options)
// options = how options are represented
type options struct {
withRunJobsLimit uint
withRunJobInterval time.Duration
withMonitorInterval time.Duration
withInterruptThreshold time.Duration
}
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.
func WithRunJobsLimit(l uint) 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.
func WithRunJobsInterval(l time.Duration) Option {
return func(o *options) {
o.withRunJobInterval = l
if o.withRunJobInterval == 0 {
o.withRunJobInterval = defaultRunJobsInterval
}
}
}
// WithMonitorInterval provides an option to provide the interval at which the scheduler
// will query running jobs for status and update the repository accordingly.
// If WithMonitorInterval == 0, then default interval is used.
func WithMonitorInterval(l time.Duration) Option {
return func(o *options) {
o.withMonitorInterval = l
if o.withMonitorInterval == 0 {
o.withMonitorInterval = defaultMonitorInterval
}
}
}
// WithInterruptThreshold provides an option to provide the duration after which a controller
// will interrupt a running job that is not updating its status.
// If WithInterruptThreshold == 0, then default duration is used.
func WithInterruptThreshold(l time.Duration) Option {
return func(o *options) {
o.withInterruptThreshold = l
if o.withInterruptThreshold == 0 {
o.withInterruptThreshold = defaultInterruptThreshold
}
}
}
+69
View File
@@ -0,0 +1,69 @@
package scheduler
import (
"testing"
"time"
"github.com/stretchr/testify/assert"
)
// 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))
testOpts := getDefaultOptions()
assert.NotEqual(opts, testOpts)
testOpts.withRunJobInterval = time.Hour
assert.Equal(opts, testOpts)
})
t.Run("WithZeroRunJobsInterval", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithRunJobsInterval(0))
testOpts := getDefaultOptions()
assert.Equal(opts, testOpts)
})
t.Run("WithMonitorInterval", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithMonitorInterval(time.Hour))
testOpts := getDefaultOptions()
assert.NotEqual(opts, testOpts)
testOpts.withMonitorInterval = time.Hour
assert.Equal(opts, testOpts)
})
t.Run("WithZeroMonitorInterval", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithMonitorInterval(0))
testOpts := getDefaultOptions()
assert.Equal(opts, testOpts)
})
t.Run("WithInterruptThreshold", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithInterruptThreshold(time.Hour))
testOpts := getDefaultOptions()
assert.NotEqual(opts, testOpts)
testOpts.withInterruptThreshold = time.Hour
assert.Equal(opts, testOpts)
})
t.Run("WithZeroInterruptThreshold", func(t *testing.T) {
assert := assert.New(t)
opts := getOpts(WithInterruptThreshold(0))
testOpts := getDefaultOptions()
assert.Equal(opts, testOpts)
})
}
+294
View File
@@ -0,0 +1,294 @@
package scheduler
import (
"context"
"fmt"
"sync"
"time"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/scheduler/job"
"github.com/hashicorp/go-hclog"
ua "go.uber.org/atomic"
)
type jobRepoFactory func() (*job.Repository, error)
type runningJob struct {
runId string
cancelCtx context.CancelFunc
status func() JobStatus
}
// Scheduler is used to register and run recurring jobs on the controller.
type Scheduler struct {
serverId string
jobRepoFn jobRepoFactory
logger hclog.Logger
registeredJobs *sync.Map
runningJobs *sync.Map
started ua.Bool
runJobsLimit uint
runJobsInterval time.Duration
monitorInterval time.Duration
interruptThreshold time.Duration
}
// New creates a new Scheduler
//
// • serverId must be provided and is the private id of the server that will run the scheduler
//
// • jobRepoFn must be provided and is a function that returns the job repository
//
// • logger must be provided and is used to log errors that occur during scheduling and running of jobs
//
// WithRunJobsLimit, WithRunJobsInterval, WithMonitorInterval and WithInterruptThreshold are
// the only valid options.
func New(serverId string, jobRepoFn jobRepoFactory, logger hclog.Logger, opt ...Option) (*Scheduler, error) {
const op = "scheduler.New"
if serverId == "" {
return nil, errors.New(errors.InvalidParameter, op, "missing server id")
}
if jobRepoFn == nil {
return nil, errors.New(errors.InvalidParameter, op, "missing job repo function")
}
if logger == nil {
return nil, errors.New(errors.InvalidParameter, op, "missing logger")
}
opts := getOpts(opt...)
return &Scheduler{
serverId: serverId,
jobRepoFn: jobRepoFn,
logger: logger,
registeredJobs: new(sync.Map),
runningJobs: new(sync.Map),
runJobsLimit: opts.withRunJobsLimit,
runJobsInterval: opts.withRunJobInterval,
monitorInterval: opts.withMonitorInterval,
interruptThreshold: opts.withInterruptThreshold,
}, nil
}
// RegisterJob registers a job with the scheduler and persists the job into the repository.
//
// • job must be provided and is an implementer of the Job interface.
//
// All options are ignored.
func (s *Scheduler) RegisterJob(ctx context.Context, j Job, _ ...Option) error {
const op = "scheduler.(Scheduler).RegisterJob"
if err := validateJob(j); err != nil {
return errors.Wrap(err, op)
}
if _, ok := s.registeredJobs.Load(j.Name()); ok {
// Job already registered
return nil
}
repo, err := s.jobRepoFn()
if err != nil {
return errors.Wrap(err, op)
}
_, err = repo.CreateJob(ctx, j.Name(), j.Description())
if err != nil && !errors.IsUniqueError(err) {
return errors.Wrap(err, op)
}
s.registeredJobs.Store(j.Name(), j)
return nil
}
// UpdateJobNextRun sets the next scheduled run time for the provided name
// to the current database time incremented by the nextRunIn parameter. If
// nextRunIn == 0 the job will be available to run immediately.
//
// All options are ignored.
func (s *Scheduler) UpdateJobNextRun(ctx context.Context, name string, nextRunIn time.Duration, _ ...Option) error {
const op = "scheduler.(Scheduler).UpdateJobNextRun"
if name == "" {
return errors.New(errors.InvalidParameter, op, "missing name")
}
repo, err := s.jobRepoFn()
if err != nil {
return errors.Wrap(err, op)
}
_, err = repo.UpdateJobNextRun(ctx, name, nextRunIn)
if err != nil {
return errors.Wrap(err, op)
}
return nil
}
// Start begins the scheduling loop that will query the repository for jobs to run and
// run them in a goroutine, the scheduler will stop all running jobs and stop requesting
// new jobs once the ctx past in is cancelled.
// The scheduler cannot be started again once the ctx is cancelled, a new scheduler will
// need to be instantiated in order to begin scheduling again.
func (s *Scheduler) Start(ctx context.Context) error {
const op = "scheduler.(Scheduler).Start"
if !s.started.CAS(s.started.Load(), true) {
s.logger.Debug("scheduler already started, skipping")
return nil
}
if err := ctx.Err(); err != nil {
return err
}
repo, err := s.jobRepoFn()
if err != nil {
return errors.Wrap(err, op)
}
// Interrupt all runs previously allocated to this server
_, err = repo.InterruptRuns(ctx, 0, job.WithServerId(s.serverId))
if err != nil {
return errors.Wrap(err, op)
}
go s.start(ctx)
go s.monitorJobs(ctx)
return nil
}
func (s *Scheduler) start(ctx context.Context) {
s.logger.Debug("starting scheduling loop")
timer := time.NewTimer(0)
for {
select {
case <-ctx.Done():
s.logger.Debug("scheduling loop shutting down")
return
case <-timer.C:
s.logger.Debug("waking up to run jobs")
repo, err := s.jobRepoFn()
if err != nil {
s.logger.Error("error creating job repo", "error", err)
break
}
runs, err := repo.RunJobs(ctx, s.serverId, job.WithRunJobsLimit(s.runJobsLimit))
if err != nil {
s.logger.Error("error getting jobs to run from repo", "error", err)
break
}
for _, r := range runs {
err := s.runJob(ctx, r)
if err != nil {
s.logger.Error("error starting job", "error", err)
if _, inner := repo.FailRun(ctx, r.PrivateId); inner != nil {
s.logger.Error("error updating failed job run", "error", inner)
}
}
}
}
s.logger.Debug("scheduling loop going back to sleep")
timer.Reset(s.runJobsInterval)
}
}
func (s *Scheduler) runJob(ctx context.Context, r *job.Run) error {
regJob, ok := s.registeredJobs.Load(r.JobName)
if !ok {
return fmt.Errorf("job %q not registered on scheduler", r.JobName)
}
s.logger.Debug("starting job run", "run id", r.PrivateId, "job name", r.JobName)
repo, err := s.jobRepoFn()
if err != nil {
return fmt.Errorf("failed to create job repository: %w", err)
}
j := regJob.(Job)
rj := &runningJob{runId: r.PrivateId, status: j.Status}
_, loaded := s.runningJobs.LoadOrStore(r.JobName, rj)
if loaded {
return fmt.Errorf("job %q is already running", r.JobName)
}
var jobContext context.Context
jobContext, rj.cancelCtx = context.WithCancel(ctx)
go func() {
defer rj.cancelCtx()
runErr := j.Run(jobContext)
switch runErr {
case nil:
s.logger.Debug("job run complete", "run id", r.PrivateId, "name", j.Name())
if _, inner := repo.CompleteRun(jobContext, r.PrivateId, j.NextRunIn()); inner != nil {
s.logger.Error("error updating completed job run", "error", inner)
}
default:
s.logger.Debug("job run failed", "run id", r.PrivateId, "name", j.Name())
if _, inner := repo.FailRun(jobContext, r.PrivateId); inner != nil {
s.logger.Error("error updating failed job run", "error", inner)
}
}
s.runningJobs.Delete(j.Name())
}()
return nil
}
func (s *Scheduler) monitorJobs(ctx context.Context) {
s.logger.Debug("starting job monitor loop")
timer := time.NewTimer(0)
for {
select {
case <-ctx.Done():
s.logger.Debug("job monitor loop shutting down")
return
case <-timer.C:
// Update progress of all running jobs
s.runningJobs.Range(func(_, v interface{}) bool {
err := s.updateRunningJobProgress(ctx, v.(*runningJob))
if err != nil {
s.logger.Error("error updating job progress", "error", err)
}
return true
})
// Check for defunct runs to interrupt
repo, err := s.jobRepoFn()
if err != nil {
s.logger.Error("error creating job repo", "error", err)
break
}
_, err = repo.InterruptRuns(ctx, s.interruptThreshold)
if err != nil {
s.logger.Error("error interrupting job runs", "error", err)
}
}
timer.Reset(s.monitorInterval)
}
}
func (s *Scheduler) updateRunningJobProgress(ctx context.Context, j *runningJob) error {
repo, err := s.jobRepoFn()
if err != nil {
return fmt.Errorf("error creating job repo %w", err)
}
status := j.status()
_, err = repo.UpdateProgress(ctx, j.runId, status.Completed, status.Total)
if errors.Match(errors.T(errors.InvalidJobRunState), err) {
// Job has been persisted with a final run status, cancel job context to trigger early exit.
j.cancelCtx()
return nil
}
if err != nil {
return err
}
return nil
}
+338
View File
@@ -0,0 +1,338 @@
package scheduler
import (
"context"
"testing"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/scheduler/job"
"github.com/hashicorp/go-hclog"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestScheduler_New(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kmsCache := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
jobRepoFn := func() (*job.Repository, error) {
return job.NewRepository(rw, rw, kmsCache)
}
type args struct {
serverId string
jobRepo jobRepoFactory
logger hclog.Logger
runLimit uint
runInterval time.Duration
}
tests := []struct {
name string
args args
opts []Option
want args
wantErr bool
wantErrCode errors.Code
wantErrMsg string
}{
{
name: "with-no-server-id",
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "scheduler.New: missing server id: parameter violation: error #100",
},
{
name: "with-no-job-repo",
args: args{
serverId: "test-server",
},
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "scheduler.New: missing job repo function: parameter violation: error #100",
},
{
name: "with-no-logger",
args: args{
serverId: "test-server",
jobRepo: jobRepoFn,
},
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "scheduler.New: missing logger: parameter violation: error #100",
},
{
name: "valid",
args: args{
serverId: "test-server",
jobRepo: jobRepoFn,
logger: hclog.L(),
},
want: args{
serverId: "test-server",
runLimit: 1,
runInterval: time.Minute,
},
},
{
name: "valid-with-interval",
args: args{
serverId: "test-server",
jobRepo: jobRepoFn,
logger: hclog.L(),
},
opts: []Option{
WithRunJobsInterval(time.Hour),
},
want: args{
serverId: "test-server",
runLimit: 1,
runInterval: time.Hour,
},
},
{
name: "valid-with-limit",
args: args{
serverId: "test-server",
jobRepo: jobRepoFn,
logger: hclog.L(),
},
opts: []Option{
WithRunJobsLimit(20),
},
want: args{
serverId: "test-server",
runLimit: 20,
runInterval: time.Minute,
},
},
{
name: "valid-with-limit-and-interval",
args: args{
serverId: "test-server",
jobRepo: jobRepoFn,
logger: hclog.L(),
},
opts: []Option{
WithRunJobsInterval(time.Hour),
WithRunJobsLimit(20),
},
want: args{
serverId: "test-server",
runLimit: 20,
runInterval: time.Hour,
},
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
got, err := New(tt.args.serverId, tt.args.jobRepo, tt.args.logger, tt.opts...)
if tt.wantErr {
require.Error(err)
assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err)
assert.Equal(tt.wantErrMsg, err.Error())
assert.Nil(got)
return
}
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.NotNil(got.jobRepoFn)
assert.NotNil(got.runningJobs)
assert.NotNil(got.registeredJobs)
})
}
}
func TestScheduler_RegisterJob(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
iam.TestRepo(t, conn, wrapper)
server := testController(t, conn, wrapper)
sched := testScheduler(t, conn, wrapper, server.PrivateId)
tests := []struct {
name string
job Job
wantErr bool
wantErrCode errors.Code
wantErrMsg string
}{
{
name: "nil-job",
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "scheduler.(Scheduler).RegisterJob: scheduler.validateJob: missing job: parameter violation: error #100",
},
{
name: "missing-job-name",
job: testJob{},
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "scheduler.(Scheduler).RegisterJob: scheduler.validateJob: missing name: parameter violation: error #100",
},
{
name: "missing-job-description",
job: testJob{
name: "name",
},
wantErr: true,
wantErrCode: errors.InvalidParameter,
wantErrMsg: "scheduler.(Scheduler).RegisterJob: scheduler.validateJob: missing description: parameter violation: error #100",
},
{
name: "valid",
job: testJob{
name: "name",
description: "description",
},
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
err := sched.RegisterJob(context.Background(), tt.job)
if tt.wantErr {
require.Error(err)
assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err)
assert.Equal(tt.wantErrMsg, err.Error())
return
}
require.NoError(err)
j, ok := sched.registeredJobs.Load(tt.job.Name())
require.True(ok)
require.NotNil(j)
assert.Equal(tt.job, j)
// Verify job has been persisted
var dbJob job.Job
err = rw.LookupWhere(context.Background(), &dbJob, "name = ?", []interface{}{tt.job.Name()})
require.NoError(err)
require.NotNil(dbJob)
assert.Equal(tt.job.Name(), dbJob.Name)
assert.Equal(tt.job.Description(), dbJob.Description)
})
}
t.Run("multiple-same-job", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
tj := testJob{
name: "multiple-same-job",
description: "description",
}
// Job should not be registered
_, ok := sched.registeredJobs.Load(tj.name)
assert.False(ok)
err := sched.RegisterJob(context.Background(), tj)
require.NoError(err)
// Job should now be registered
_, ok = sched.registeredJobs.Load(tj.name)
assert.True(ok)
// Registering the same job should not return an error
err = sched.RegisterJob(context.Background(), tj)
require.NoError(err)
// Job should still be registered
_, ok = sched.registeredJobs.Load(tj.name)
assert.True(ok)
})
t.Run("multiple-schedulers-registering-same-job", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
tj := testJob{
name: "name",
description: "description",
}
err := sched.RegisterJob(context.Background(), tj)
require.NoError(err)
_, ok := sched.registeredJobs.Load(tj.name)
assert.True(ok)
server1 := testController(t, conn, wrapper)
sched1 := testScheduler(t, conn, wrapper, server1.PrivateId)
// Verify job is not registered on second scheduler
_, ok = sched1.registeredJobs.Load(tj.name)
assert.False(ok)
// Registering job on second scheduler should not return an error and should be registered
err = sched1.RegisterJob(context.Background(), tj)
require.NoError(err)
_, ok = sched.registeredJobs.Load(tj.name)
assert.True(ok)
})
}
func TestScheduler_UpdateJobNextRun(t *testing.T) {
t.Parallel()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
iam.TestRepo(t, conn, wrapper)
server := testController(t, conn, wrapper)
sched := testScheduler(t, conn, wrapper, server.PrivateId)
t.Run("missing-job-name", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
err := sched.UpdateJobNextRun(context.Background(), "", time.Hour)
require.Error(err)
assert.Truef(errors.Match(errors.T(errors.InvalidParameter), err), "Unexpected error %s", err)
assert.Equal("scheduler.(Scheduler).UpdateJobNextRun: missing name: parameter violation: error #100", err.Error())
})
t.Run("job-not-found", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
// Insert fake job to bypass registration check
sched.registeredJobs.Store("fake-job-name", nil)
err := sched.UpdateJobNextRun(context.Background(), "fake-job-name", time.Hour)
require.Error(err)
assert.Truef(errors.Match(errors.T(errors.RecordNotFound), err), "Unexpected error %s", err)
assert.Equal("scheduler.(Scheduler).UpdateJobNextRun: job.(Repository).UpdateJobNextRun: db.DoTx: job.(Repository).UpdateJobNextRun: job \"fake-job-name\" does not exist: search issue: error #1100", err.Error())
})
t.Run("valid", func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
tj := testJob{
name: "name",
description: "description",
}
err := sched.RegisterJob(context.Background(), tj)
require.NoError(err)
// get job from repo
var dbJob job.Job
err = rw.LookupWhere(context.Background(), &dbJob, "name = ?", []interface{}{tj.name})
require.NoError(err)
previousNextRun := dbJob.NextScheduledRun.Timestamp.GetSeconds()
err = sched.UpdateJobNextRun(context.Background(), tj.name, time.Hour)
require.NoError(err)
err = rw.LookupWhere(context.Background(), &dbJob, "name = ?", []interface{}{tj.name})
require.NoError(err)
// Verify job run time in repo is at least 1 hour later than the previous run time
assert.True(previousNextRun+int64(time.Hour.Seconds()) <= dbJob.NextScheduledRun.Timestamp.GetSeconds())
})
}
+109
View File
@@ -0,0 +1,109 @@
package scheduler
import (
"context"
"sync"
"testing"
"time"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/scheduler/job"
"github.com/hashicorp/boundary/internal/servers"
"github.com/hashicorp/go-hclog"
wrapping "github.com/hashicorp/go-kms-wrapping"
"github.com/hashicorp/go-uuid"
"github.com/jinzhu/gorm"
"github.com/stretchr/testify/require"
)
func testScheduler(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper, serverId string, opts ...Option) *Scheduler {
t.Helper()
rw := db.New(conn)
kmsCache := kms.TestKms(t, conn, wrapper)
iam.TestRepo(t, conn, wrapper)
jobRepoFn := func() (*job.Repository, error) {
return job.NewRepository(rw, rw, kmsCache)
}
s, err := New(serverId, jobRepoFn, hclog.L(), opts...)
require.NoError(t, err)
return s
}
func testController(t *testing.T, conn *gorm.DB, wrapper wrapping.Wrapper) *servers.Server {
t.Helper()
rw := db.New(conn)
kms := kms.TestKms(t, conn, wrapper)
serversRepo, err := servers.NewRepository(rw, rw, kms)
require.NoError(t, err)
id, err := uuid.GenerateUUID()
require.NoError(t, err)
controller := &servers.Server{
PrivateId: "test-job-server-" + id,
Type: servers.ServerTypeController.String(),
Description: "Test Job Controller",
Address: "127.0.0.1",
}
_, _, err = serversRepo.UpsertServer(context.Background(), controller)
require.NoError(t, err)
return controller
}
func testJobFn() (func(ctx context.Context) error, chan struct{}, chan struct{}) {
jobReady := make(chan struct{})
jobDone := make(chan struct{})
fn := func(ctx context.Context) error {
jobReady <- struct{}{}
// Block until context is cancelled
<-ctx.Done()
jobDone <- struct{}{}
return nil
}
return fn, jobReady, jobDone
}
type testJob struct {
nextRunIn time.Duration
name, description string
fn func(context.Context) error
statusFn func() JobStatus
}
func (j testJob) Status() JobStatus {
if j.statusFn == nil {
return JobStatus{}
}
return j.statusFn()
}
func (j testJob) Run(ctx context.Context) error {
return j.fn(ctx)
}
func (j testJob) NextRunIn() time.Duration {
return j.nextRunIn
}
func (j testJob) Name() string {
return j.name
}
func (j testJob) Description() string {
return j.description
}
func mapLen(sm *sync.Map) int {
count := 0
sm.Range(func(key, value interface{}) bool {
count++
return true
})
return count
}
+15
View File
@@ -14,6 +14,8 @@ import (
"github.com/hashicorp/boundary/internal/host/static"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/scheduler"
"github.com/hashicorp/boundary/internal/scheduler/job"
"github.com/hashicorp/boundary/internal/servers"
"github.com/hashicorp/boundary/internal/servers/controller/common"
"github.com/hashicorp/boundary/internal/session"
@@ -48,6 +50,8 @@ type Controller struct {
StaticHostRepoFn common.StaticRepoFactory
TargetRepoFn common.TargetRepoFactory
scheduler *scheduler.Scheduler
kms *kms.Kms
}
@@ -137,6 +141,14 @@ func New(conf *Config) (*Controller, error) {
c.workerAuthCache = cache.New(0, 0)
jobRepoFn := func() (*job.Repository, error) {
return job.NewRepository(dbase, dbase, c.kms)
}
c.scheduler, err = scheduler.New(c.conf.RawConfig.Controller.Name, jobRepoFn, c.logger)
if err != nil {
return nil, fmt.Errorf("error creating new scheduler: %w", err)
}
return c, nil
}
@@ -146,6 +158,9 @@ func (c *Controller) Start() error {
return nil
}
c.baseContext, c.baseCancel = context.WithCancel(context.Background())
if err := c.scheduler.Start(c.baseContext); err != nil {
return fmt.Errorf("error starting scheduler: %w", err)
}
if err := c.startListeners(); err != nil {
return fmt.Errorf("error starting controller listeners: %w", err)