blob: 41b554a58adf2eaba0ef1c2b4c6d96f90a90bdde [file] [log] [blame]
// Copyright 2015 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package engine
import (
"encoding/json"
"math/rand"
"sort"
"strings"
"testing"
"time"
"github.com/golang/protobuf/proto"
"golang.org/x/net/context"
"google.golang.org/api/pubsub/v1"
"github.com/luci/gae/impl/memory"
ds "github.com/luci/gae/service/datastore"
tq "github.com/luci/gae/service/taskqueue"
"github.com/luci/luci-go/common/clock"
"github.com/luci/luci-go/common/clock/testclock"
"github.com/luci/luci-go/common/data/rand/mathrand"
"github.com/luci/luci-go/common/data/stringset"
"github.com/luci/luci-go/common/errors"
"github.com/luci/luci-go/common/retry/transient"
"github.com/luci/luci-go/server/auth"
"github.com/luci/luci-go/server/auth/authtest"
"github.com/luci/luci-go/server/secrets/testsecrets"
"github.com/luci/luci-go/scheduler/appengine/acl"
"github.com/luci/luci-go/scheduler/appengine/catalog"
"github.com/luci/luci-go/scheduler/appengine/messages"
"github.com/luci/luci-go/scheduler/appengine/task"
"github.com/luci/luci-go/scheduler/appengine/task/noop"
. "github.com/luci/luci-go/common/testing/assertions"
. "github.com/smartystreets/goconvey/convey"
)
func TestGetAllProjects(t *testing.T) {
Convey("works", t, func() {
c := newTestContext(epoch)
e, _ := newTestEngine()
// Empty.
projects, err := e.GetAllProjects(c)
So(err, ShouldBeNil)
So(len(projects), ShouldEqual, 0)
// Non empty.
So(ds.Put(c,
&Job{JobID: "abc/1", ProjectID: "abc", Enabled: true},
&Job{JobID: "abc/2", ProjectID: "abc", Enabled: true},
&Job{JobID: "def/1", ProjectID: "def", Enabled: true},
&Job{JobID: "xyz/1", ProjectID: "xyz", Enabled: false},
), ShouldBeNil)
ds.GetTestable(c).CatchupIndexes()
projects, err = e.GetAllProjects(c)
So(err, ShouldBeNil)
So(projects, ShouldResemble, []string{"abc", "def"})
})
}
func TestUpdateProjectJobs(t *testing.T) {
Convey("works", t, func() {
c := newTestContext(epoch)
e, _ := newTestEngine()
// Doing nothing.
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{}), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{})
// Adding a new job (ticks every 5 sec).
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{
{
JobID: "abc/1",
Revision: "rev1",
Schedule: "*/5 * * * * * *",
}}), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
State: JobState{
State: "SCHEDULED",
TickNonce: 6278013164014963328,
TickTime: epoch.Add(5 * time.Second),
},
},
})
// TODO(tandrii): delete and update above definition after no-ACL -> ACL transition.
// Simulate ACL version roll without change of project config.
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{
{
JobID: "abc/1",
Revision: "rev1",
Schedule: "*/5 * * * * * *",
Acls: acl.GrantsByRole{Readers: []string{"group:r"}, Owners: []string{"groups:o"}},
}}), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Acls: acl.GrantsByRole{Readers: []string{"group:r"}, Owners: []string{"groups:o"}},
Schedule: "*/5 * * * * * *",
State: JobState{
State: "SCHEDULED",
TickNonce: 6278013164014963328,
TickTime: epoch.Add(5 * time.Second),
},
},
})
// Enqueued timer task to launch it.
task := ensureOneTask(c, "timers-q")
So(task.Path, ShouldEqual, "/timers")
So(task.ETA, ShouldResemble, epoch.Add(5*time.Second))
tq.GetTestable(c).ResetTasks()
// Readding same job in with exact same config revision -> noop.
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{
{
JobID: "abc/1",
Revision: "rev1",
Schedule: "*/5 * * * * * *",
}}), ShouldBeNil)
ensureZeroTasks(c, "timers-q")
ensureZeroTasks(c, "invs-q")
// Changing schedule to tick earlier -> rescheduled.
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{
{
JobID: "abc/1",
Revision: "rev2",
Schedule: "*/1 * * * * * *",
}}), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev2",
Enabled: true,
Schedule: "*/1 * * * * * *",
State: JobState{
State: "SCHEDULED",
TickNonce: 886585524575582446,
TickTime: epoch.Add(1 * time.Second),
},
},
})
// Enqueued timer task to launch it.
task = ensureOneTask(c, "timers-q")
So(task.Path, ShouldEqual, "/timers")
So(task.ETA, ShouldResemble, epoch.Add(1*time.Second))
tq.GetTestable(c).ResetTasks()
// Removed -> goes to disabled state.
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{}), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev2",
Enabled: false,
Schedule: "*/1 * * * * * *",
State: JobState{
State: "DISABLED",
},
},
})
ensureZeroTasks(c, "timers-q")
ensureZeroTasks(c, "invs-q")
})
}
func TestTransactionRetries(t *testing.T) {
Convey("retry works", t, func() {
c := newTestContext(epoch)
e, _ := newTestEngine()
// Adding a new job with transaction retry, should enqueue one task.
ds.GetTestable(c).SetTransactionRetryCount(2)
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{
{
JobID: "abc/1",
Revision: "rev1",
Schedule: "*/5 * * * * * *",
}}), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
State: JobState{
State: "SCHEDULED",
TickNonce: 928953616732700780,
TickTime: epoch.Add(5 * time.Second),
},
},
})
// Enqueued timer task to launch it.
task := ensureOneTask(c, "timers-q")
So(task.Path, ShouldEqual, "/timers")
So(task.ETA, ShouldResemble, epoch.Add(5*time.Second))
tq.GetTestable(c).ResetTasks()
})
Convey("collision is handled", t, func() {
c := newTestContext(epoch)
e, _ := newTestEngine()
// Pretend collision happened in all retries.
ds.GetTestable(c).SetTransactionRetryCount(15)
err := e.UpdateProjectJobs(c, "abc", []catalog.Definition{
{
JobID: "abc/1",
Revision: "rev1",
Schedule: "*/5 * * * * * *",
}})
So(transient.Tag.In(err), ShouldBeTrue)
So(allJobs(c), ShouldResemble, []Job{})
ensureZeroTasks(c, "timers-q")
ensureZeroTasks(c, "invs-q")
})
}
func TestResetAllJobsOnDevServer(t *testing.T) {
Convey("works", t, func() {
c := newTestContext(epoch)
e, _ := newTestEngine()
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{
{
JobID: "abc/1",
Revision: "rev1",
Schedule: "*/5 * * * * * *",
}}), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
State: JobState{
State: "SCHEDULED",
TickNonce: 6278013164014963328,
TickTime: epoch.Add(5 * time.Second),
},
},
})
clock.Get(c).(testclock.TestClock).Add(1 * time.Minute)
// ResetAllJobsOnDevServer should reschedule the job.
So(e.ResetAllJobsOnDevServer(c), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
State: JobState{
State: "SCHEDULED",
TickNonce: 886585524575582446,
TickTime: epoch.Add(65 * time.Second),
},
},
})
})
}
func TestFullFlow(t *testing.T) {
Convey("full flow", t, func() {
c := newTestContext(epoch)
e, mgr := newTestEngine()
taskBytes := noopTaskBytes()
// Adding a new job (ticks every 5 sec).
So(e.UpdateProjectJobs(c, "abc", []catalog.Definition{
{
JobID: "abc/1",
Revision: "rev1",
Schedule: "*/5 * * * * * *",
Task: taskBytes,
}}), ShouldBeNil)
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
Task: taskBytes,
State: JobState{
State: "SCHEDULED",
TickNonce: 6278013164014963328,
TickTime: epoch.Add(5 * time.Second),
},
},
})
// Enqueued timer task to launch it.
tsk := ensureOneTask(c, "timers-q")
So(tsk.Path, ShouldEqual, "/timers")
So(tsk.ETA, ShouldResemble, epoch.Add(5*time.Second))
tq.GetTestable(c).ResetTasks()
// Tick time comes, the tick task is executed, job is added to queue.
clock.Get(c).(testclock.TestClock).Add(5 * time.Second)
So(e.ExecuteSerializedAction(c, tsk.Payload, 0), ShouldBeNil)
// Job is in queued state now.
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
Task: taskBytes,
State: JobState{
State: "QUEUED",
TickNonce: 886585524575582446,
TickTime: epoch.Add(10 * time.Second),
InvocationNonce: 928953616732700780,
InvocationTime: epoch.Add(5 * time.Second),
},
},
})
// Next tick task is added.
tickTask := ensureOneTask(c, "timers-q")
So(tickTask.Path, ShouldEqual, "/timers")
So(tickTask.ETA, ShouldResemble, epoch.Add(10*time.Second))
// Invocation task (ETA is 1 sec in the future).
invTask := ensureOneTask(c, "invs-q")
So(invTask.Path, ShouldEqual, "/invs")
So(invTask.ETA, ShouldResemble, epoch.Add(6*time.Second))
tq.GetTestable(c).ResetTasks()
// Time to run the job and it fails to launch with a transient error.
mgr.launchTask = func(ctx context.Context, ctl task.Controller) error {
// Check data provided via the controller.
So(ctl.JobID(), ShouldEqual, "abc/1")
So(ctl.InvocationID(), ShouldEqual, int64(9200093518582198800))
So(ctl.InvocationNonce(), ShouldEqual, int64(928953616732700780))
So(ctl.Task(), ShouldResemble, &messages.NoopTask{})
ctl.DebugLog("oops, fail")
return errors.New("oops", transient.Tag)
}
So(transient.Tag.In(e.ExecuteSerializedAction(c, invTask.Payload, 0)), ShouldBeTrue)
// Still in QUEUED state, but with InvocatioID assigned.
jobs := allJobs(c)
So(jobs, ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
Task: taskBytes,
State: JobState{
State: "QUEUED",
TickNonce: 886585524575582446,
TickTime: epoch.Add(10 * time.Second),
InvocationNonce: 928953616732700780,
InvocationTime: epoch.Add(5 * time.Second),
InvocationID: 9200093518582198800,
},
},
})
jobKey := ds.KeyForObj(c, &jobs[0])
// Check Invocation fields.
inv := Invocation{ID: 9200093518582198800, JobKey: jobKey}
So(ds.Get(c, &inv), ShouldBeNil)
inv.JobKey = nil // for easier ShouldResemble below
debugLog := inv.DebugLog
inv.DebugLog = ""
So(inv, ShouldResemble, Invocation{
ID: 9200093518582198800,
InvocationNonce: 928953616732700780,
Revision: "rev1",
Started: epoch.Add(5 * time.Second),
Finished: epoch.Add(5 * time.Second),
Task: taskBytes,
DebugLog: "",
Status: task.StatusFailed,
MutationsCount: 1,
})
So(debugLog, ShouldContainSubstring, "[22:42:05.000] Invocation initiated (attempt 1)")
So(debugLog, ShouldContainSubstring, "[22:42:05.000] oops, fail")
So(debugLog, ShouldContainSubstring, "with status FAILED")
So(debugLog, ShouldContainSubstring, "[22:42:05.000] It will probably be retried")
// Second attempt. Now starts, hangs midway, they finishes.
mgr.launchTask = func(ctx context.Context, ctl task.Controller) error {
// Make sure Save() checkpoints the progress.
ctl.DebugLog("Starting")
ctl.State().Status = task.StatusRunning
So(ctl.Save(ctx), ShouldBeNil)
// After first Save the job and the invocation are in running state.
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
Task: taskBytes,
State: JobState{
State: "RUNNING",
TickNonce: 886585524575582446,
TickTime: epoch.Add(10 * time.Second),
InvocationNonce: 928953616732700780,
InvocationRetryCount: 1,
InvocationTime: epoch.Add(5 * time.Second),
InvocationID: 9200093518582296192,
},
},
})
inv := Invocation{ID: 9200093518582296192, JobKey: jobKey}
So(ds.Get(c, &inv), ShouldBeNil)
inv.JobKey = nil // for easier ShouldResemble below
So(inv, ShouldResemble, Invocation{
ID: 9200093518582296192,
InvocationNonce: 928953616732700780,
Revision: "rev1",
Started: epoch.Add(5 * time.Second),
Task: taskBytes,
DebugLog: "[22:42:05.000] Invocation initiated (attempt 2)\n[22:42:05.000] Starting\n",
RetryCount: 1,
Status: task.StatusRunning,
MutationsCount: 1,
})
// Noop save, just for the code coverage.
So(ctl.Save(ctx), ShouldBeNil)
// Change state to the final one.
ctl.State().Status = task.StatusSucceeded
ctl.State().ViewURL = "http://view_url"
ctl.State().TaskData = []byte("blah")
return nil
}
So(e.ExecuteSerializedAction(c, invTask.Payload, 1), ShouldBeNil)
// After final save.
inv = Invocation{ID: 9200093518582296192, JobKey: jobKey}
So(ds.Get(c, &inv), ShouldBeNil)
inv.JobKey = nil // for easier ShouldResemble below
debugLog = inv.DebugLog
inv.DebugLog = ""
So(inv, ShouldResemble, Invocation{
ID: 9200093518582296192,
InvocationNonce: 928953616732700780,
Revision: "rev1",
Started: epoch.Add(5 * time.Second),
Finished: epoch.Add(5 * time.Second),
Task: taskBytes,
DebugLog: "",
RetryCount: 1,
Status: task.StatusSucceeded,
ViewURL: "http://view_url",
TaskData: []byte("blah"),
MutationsCount: 2,
})
So(debugLog, ShouldContainSubstring, "[22:42:05.000] Invocation initiated (attempt 2)")
So(debugLog, ShouldContainSubstring, "[22:42:05.000] Starting")
So(debugLog, ShouldContainSubstring, "with status SUCCEEDED")
// Previous invocation is canceled.
inv = Invocation{ID: 9200093518582198800, JobKey: jobKey}
So(ds.Get(c, &inv), ShouldBeNil)
inv.JobKey = nil // for easier ShouldResemble below
debugLog = inv.DebugLog
inv.DebugLog = ""
So(inv, ShouldResemble, Invocation{
ID: 9200093518582198800,
InvocationNonce: 928953616732700780,
Revision: "rev1",
Started: epoch.Add(5 * time.Second),
Finished: epoch.Add(5 * time.Second),
Task: taskBytes,
DebugLog: "",
Status: task.StatusFailed,
MutationsCount: 1,
})
So(debugLog, ShouldContainSubstring, "[22:42:05.000] Invocation initiated (attempt 1)")
So(debugLog, ShouldContainSubstring, "[22:42:05.000] oops, fail")
So(debugLog, ShouldContainSubstring, "with status FAILED")
So(debugLog, ShouldContainSubstring, "[22:42:05.000] It will probably be retried")
// Job is in scheduled state again.
So(allJobs(c), ShouldResemble, []Job{
{
JobID: "abc/1",
ProjectID: "abc",
Revision: "rev1",
Enabled: true,
Schedule: "*/5 * * * * * *",
Task: taskBytes,
State: JobState{
State: "SCHEDULED",
TickNonce: 886585524575582446,
TickTime: epoch.Add(10 * time.Second),
PrevTime: epoch.Add(5 * time.Second),
},
},
})
})
}
func TestGenerateInvocationID(t *testing.T) {
Convey("generateInvocationID does not collide", t, func() {
c := newTestContext(epoch)
k := ds.NewKey(c, "Job", "", 123, nil)
// Bunch of ids generated at the exact same moment in time do not collide.
ids := map[int64]struct{}{}
for i := 0; i < 20; i++ {
id, err := generateInvocationID(c, k)
So(err, ShouldBeNil)
ids[id] = struct{}{}
}
So(len(ids), ShouldEqual, 20)
})
Convey("generateInvocationID gen IDs with most recent first", t, func() {
c := newTestContext(epoch)
k := ds.NewKey(c, "Job", "", 123, nil)
older, err := generateInvocationID(c, k)
So(err, ShouldBeNil)
clock.Get(c).(testclock.TestClock).Add(5 * time.Second)
newer, err := generateInvocationID(c, k)
So(err, ShouldBeNil)
So(newer, ShouldBeLessThan, older)
})
}
func TestQueries(t *testing.T) {
Convey("with mock data", t, func() {
c := newTestContext(epoch)
e, _ := newTestEngine()
// TODO(tandrii): remove aclDefault once all Jobs have ACLs.
aclDefault := acl.GrantsByRole{}
aclSome := acl.GrantsByRole{Readers: []string{"group:some"}}
aclOne := acl.GrantsByRole{Owners: []string{"one@example.com"}}
aclAdmin := acl.GrantsByRole{Readers: []string{"group:administrators"}, Owners: []string{"group:administrators"}}
ctxAnon := auth.WithState(c, &authtest.FakeState{
Identity: "anonymous:anonymous",
})
ctxOne := auth.WithState(c, &authtest.FakeState{
Identity: "user:one@example.com",
IdentityGroups: []string{"all"},
})
ctxSome := auth.WithState(c, &authtest.FakeState{
Identity: "user:some@example.com",
IdentityGroups: []string{"some"},
})
ctxAdmin := auth.WithState(c, &authtest.FakeState{
Identity: "user:admin@example.com",
IdentityGroups: []string{"administrators"},
})
So(ds.Put(c,
&Job{JobID: "abc/1", ProjectID: "abc", Enabled: true, Acls: aclOne},
&Job{JobID: "abc/2", ProjectID: "abc", Enabled: true, Acls: aclSome},
&Job{JobID: "abc/3", ProjectID: "abc", Enabled: true, Acls: aclDefault},
&Job{JobID: "def/1", ProjectID: "def", Enabled: true, Acls: aclDefault},
&Job{JobID: "def/2", ProjectID: "def", Enabled: false, Acls: aclDefault},
&Job{JobID: "secret/1", ProjectID: "secret", Enabled: true, Acls: aclAdmin},
), ShouldBeNil)
job1 := ds.NewKey(c, "Job", "abc/1", 0, nil)
job2 := ds.NewKey(c, "Job", "abc/2", 0, nil)
job3 := ds.NewKey(c, "Job", "abc/3", 0, nil)
So(ds.Put(c,
&Invocation{ID: 1, JobKey: job1, InvocationNonce: 123},
&Invocation{ID: 2, JobKey: job1, InvocationNonce: 123},
&Invocation{ID: 3, JobKey: job1},
&Invocation{ID: 1, JobKey: job2},
&Invocation{ID: 2, JobKey: job2},
&Invocation{ID: 3, JobKey: job2},
&Invocation{ID: 1, JobKey: job3},
), ShouldBeNil)
ds.GetTestable(c).CatchupIndexes()
Convey("GetAllProjects ignores ACLs and CurrentIdentity", func() {
test := func(ctx context.Context) {
r, err := e.GetAllProjects(c)
So(err, ShouldBeNil)
So(r, ShouldResemble, []string{"abc", "def", "secret"})
}
test(c)
test(ctxAnon)
test(ctxAdmin)
})
Convey("GetVisibleJobs works", func() {
get := func(ctx context.Context) []string {
jobs, err := e.GetVisibleJobs(ctx)
So(err, ShouldBeNil)
return sortedJobIds(jobs)
}
Convey("Anonymous users see only public jobs", func() {
// Only 3 jobs with default ACLs granting READER access to everyone, but
// def/2 is disabled and so shouldn't be returned.
So(get(ctxAnon), ShouldResemble, []string{"abc/3", "def/1"})
})
Convey("Owners can see their own jobs + public jobs", func() {
// abc/1 is owned by one@example.com.
So(get(ctxOne), ShouldResemble, []string{"abc/1", "abc/3", "def/1"})
})
Convey("Explicit readers", func() {
So(get(ctxSome), ShouldResemble, []string{"abc/2", "abc/3", "def/1"})
})
Convey("Admins have implicit READER access to all jobs", func() {
So(get(ctxAdmin), ShouldResemble, []string{"abc/1", "abc/2", "abc/3", "def/1", "secret/1"})
})
})
Convey("GetProjectJobsRA works", func() {
get := func(ctx context.Context, project string) []string {
jobs, err := e.GetVisibleProjectJobs(ctx, project)
So(err, ShouldBeNil)
return sortedJobIds(jobs)
}
Convey("Anonymous can still see public jobs", func() {
So(get(ctxAnon, "def"), ShouldResemble, []string{"def/1"})
})
Convey("Admin have implicit READER access to all jobs", func() {
So(get(ctxAdmin, "abc"), ShouldResemble, []string{"abc/1", "abc/2", "abc/3"})
})
Convey("Owners can still see their jobs", func() {
So(get(ctxOne, "abc"), ShouldResemble, []string{"abc/1", "abc/3"})
})
Convey("Readers can see their jobs", func() {
So(get(ctxSome, "abc"), ShouldResemble, []string{"abc/2", "abc/3"})
})
})
Convey("GetVisibleJob works", func() {
_, err := e.GetVisibleJob(ctxAdmin, "missing/job")
So(err, ShouldEqual, ErrNoSuchJob)
_, err = e.GetVisibleJob(ctxAnon, "abc/1") // no READER permission.
So(err, ShouldEqual, ErrNoSuchJob)
job, err := e.GetVisibleJob(ctxAnon, "def/1") // OK.
So(job, ShouldNotBeNil)
So(err, ShouldBeNil)
job, err = e.GetVisibleJob(ctxAnon, "def/2") // OK, even though not enabled.
So(job, ShouldNotBeNil)
So(err, ShouldBeNil)
})
Convey("ListVisibleInvocations works", func() {
Convey("Anonymous can't see non-public job invocations", func() {
_, _, err := e.ListVisibleInvocations(ctxAnon, "abc/1", 2, "")
So(err, ShouldResemble, ErrNoSuchJob)
})
Convey("With paging", func() {
invs, cursor, err := e.ListVisibleInvocations(ctxOne, "abc/1", 2, "")
So(err, ShouldBeNil)
So(len(invs), ShouldEqual, 2)
So(invs[0].ID, ShouldEqual, 1)
So(invs[1].ID, ShouldEqual, 2)
So(cursor, ShouldNotEqual, "")
invs, cursor, err = e.ListVisibleInvocations(ctxOne, "abc/1", 2, cursor)
So(err, ShouldBeNil)
So(len(invs), ShouldEqual, 1)
So(invs[0].ID, ShouldEqual, 3)
So(cursor, ShouldEqual, "")
})
})
Convey("GetInvocation works", func() {
Convey("Anonymous can't see non-public job invocation", func() {
_, err := e.GetVisibleInvocation(ctxAnon, "abc/1", 1)
So(err, ShouldResemble, ErrNoSuchInvocation)
})
Convey("NoSuchInvocation", func() {
_, err := e.GetVisibleInvocation(ctxAdmin, "missing/job", 1)
So(err, ShouldResemble, ErrNoSuchInvocation)
})
Convey("Reader sees", func() {
inv, err := e.GetVisibleInvocation(ctxOne, "abc/1", 1)
So(inv, ShouldNotBeNil)
So(err, ShouldBeNil)
})
})
Convey("GetInvocationsByNonce works", func() {
Convey("Anonymous can't see non-public job invocations", func() {
invs, err := e.GetVisibleInvocationsByNonce(ctxAnon, 123)
So(len(invs), ShouldEqual, 0)
So(err, ShouldBeNil)
})
Convey("NoSuchInvocation", func() {
invs, err := e.GetVisibleInvocationsByNonce(ctxAdmin, 11111) // unknown
So(len(invs), ShouldEqual, 0)
So(err, ShouldBeNil)
})
Convey("Reader sees", func() {
invs, err := e.GetVisibleInvocationsByNonce(ctxOne, 123)
So(len(invs), ShouldEqual, 2)
So(err, ShouldBeNil)
})
})
})
}
func TestPrepareTopic(t *testing.T) {
Convey("PrepareTopic works", t, func(ctx C) {
c := newTestContext(epoch)
e, _ := newTestEngine()
pubSubCalls := 0
e.configureTopic = func(c context.Context, topic, sub, pushURL, publisher string) error {
pubSubCalls++
ctx.So(topic, ShouldEqual, "projects/app/topics/dev-scheduler+noop+some~publisher.com")
ctx.So(sub, ShouldEqual, "projects/app/subscriptions/dev-scheduler+noop+some~publisher.com")
ctx.So(pushURL, ShouldEqual, "") // pull on dev server
ctx.So(publisher, ShouldEqual, "some@publisher.com")
return nil
}
ctl := &taskController{
ctx: c,
eng: e,
manager: &noop.TaskManager{},
saved: Invocation{
ID: 123456,
JobKey: ds.NewKey(c, "Job", "job_id", 0, nil),
},
}
ctl.populateState()
// Once.
topic, token, err := ctl.PrepareTopic(c, "some@publisher.com")
So(err, ShouldBeNil)
So(topic, ShouldEqual, "projects/app/topics/dev-scheduler+noop+some~publisher.com")
So(token, ShouldNotEqual, "")
So(pubSubCalls, ShouldEqual, 1)
// Again. 'configureTopic' should not be called anymore.
_, _, err = ctl.PrepareTopic(c, "some@publisher.com")
So(err, ShouldBeNil)
So(pubSubCalls, ShouldEqual, 1)
// Make sure memcache-based deduplication also works.
e.doneFlags = make(map[string]bool)
_, _, err = ctl.PrepareTopic(c, "some@publisher.com")
So(err, ShouldBeNil)
So(pubSubCalls, ShouldEqual, 1)
})
}
func TestProcessPubSubPush(t *testing.T) {
Convey("with mock invocation", t, func() {
c := newTestContext(epoch)
e, mgr := newTestEngine()
So(ds.Put(c, &Job{
JobID: "abc/1",
ProjectID: "abc",
Enabled: true,
}), ShouldBeNil)
task, err := proto.Marshal(&messages.TaskDefWrapper{
Noop: &messages.NoopTask{},
})
So(err, ShouldBeNil)
inv := Invocation{
ID: 1,
JobKey: ds.NewKey(c, "Job", "abc/1", 0, nil),
Task: task,
}
So(ds.Put(c, &inv), ShouldBeNil)
// Skip talking to PubSub for real.
e.configureTopic = func(c context.Context, topic, sub, pushURL, publisher string) error {
return nil
}
ctl, err := e.controllerForInvocation(c, &inv)
So(err, ShouldBeNil)
// Grab the working auth token.
_, token, err := ctl.PrepareTopic(c, "some@publisher.com")
So(err, ShouldBeNil)
So(token, ShouldNotEqual, "")
Convey("ProcessPubSubPush works", func() {
msg := struct {
Message pubsub.PubsubMessage `json:"message"`
}{
Message: pubsub.PubsubMessage{
Attributes: map[string]string{"auth_token": token},
Data: "blah",
},
}
blob, err := json.Marshal(&msg)
So(err, ShouldBeNil)
handled := false
mgr.handleNotification = func(ctx context.Context, msg *pubsub.PubsubMessage) error {
So(msg.Data, ShouldEqual, "blah")
handled = true
return nil
}
So(e.ProcessPubSubPush(c, blob), ShouldBeNil)
So(handled, ShouldBeTrue)
})
Convey("ProcessPubSubPush handles bad token", func() {
msg := struct {
Message pubsub.PubsubMessage `json:"message"`
}{
Message: pubsub.PubsubMessage{
Attributes: map[string]string{"auth_token": token + "blah"},
Data: "blah",
},
}
blob, err := json.Marshal(&msg)
So(err, ShouldBeNil)
So(e.ProcessPubSubPush(c, blob), ShouldErrLike, "bad token")
})
Convey("ProcessPubSubPush handles missing invocation", func() {
ds.Delete(c, ds.KeyForObj(c, &inv))
msg := pubsub.PubsubMessage{
Attributes: map[string]string{"auth_token": token},
}
blob, err := json.Marshal(&msg)
So(err, ShouldBeNil)
So(transient.Tag.In(e.ProcessPubSubPush(c, blob)), ShouldBeFalse)
})
})
}
func TestAborts(t *testing.T) {
Convey("with mock invocation", t, func() {
c := newTestContext(epoch)
e, mgr := newTestEngine()
ctxAnon := auth.WithState(c, &authtest.FakeState{
Identity: "anonymous:anonymous",
})
ctxReader := auth.WithState(c, &authtest.FakeState{
Identity: "user:reader@example.com",
IdentityGroups: []string{"readers"},
})
ctxOwner := auth.WithState(c, &authtest.FakeState{
Identity: "user:owner@example.com",
IdentityGroups: []string{"owners"},
})
// A job in "QUEUED" state (about to run an invocation).
const jobID = "abc/1"
const invNonce = int64(12345)
prepareQueuedJob(c, jobID, invNonce)
launchInv := func() int64 {
var invID int64
mgr.launchTask = func(ctx context.Context, ctl task.Controller) error {
invID = ctl.InvocationID()
ctl.State().Status = task.StatusRunning
So(ctl.Save(ctx), ShouldBeNil)
return nil
}
So(e.startInvocation(c, jobID, invNonce, "", 0), ShouldBeNil)
// It is alive and the job entity tracks it.
inv, err := e.getInvocation(c, jobID, invID)
So(err, ShouldBeNil)
So(inv.Status, ShouldEqual, task.StatusRunning)
job, err := e.getJob(c, jobID)
So(err, ShouldBeNil)
So(job.State.State, ShouldEqual, JobStateRunning)
So(job.State.InvocationID, ShouldEqual, invID)
return invID
}
Convey("AbortInvocation works", func() {
// Actually launch the queued invocation.
invID := launchInv()
// Try to kill it w/o permission.
So(e.AbortInvocation(c, jobID, invID), ShouldNotBeNil) // No current identity.
So(e.AbortInvocation(ctxAnon, jobID, invID), ShouldResemble, ErrNoSuchJob)
So(e.AbortInvocation(ctxReader, jobID, invID), ShouldResemble, ErrNoOwnerPermission)
// Now kill it.
So(e.AbortInvocation(ctxOwner, jobID, invID), ShouldBeNil)
// It is dead.
inv, err := e.getInvocation(c, jobID, invID)
So(err, ShouldBeNil)
So(inv.Status, ShouldEqual, task.StatusAborted)
// The job moved on with its life.
job, err := e.getJob(c, jobID)
So(err, ShouldBeNil)
So(job.State.State, ShouldEqual, JobStateSuspended)
So(job.State.InvocationID, ShouldEqual, 0)
})
Convey("AbortJob kills running invocation", func() {
// Actually launch the queued invocation.
invID := launchInv()
// Try to kill it w/o permission.
So(e.AbortJob(c, jobID), ShouldNotBeNil) // No current identity.
So(e.AbortJob(ctxAnon, jobID), ShouldResemble, ErrNoSuchJob)
So(e.AbortJob(ctxReader, jobID), ShouldResemble, ErrNoOwnerPermission)
// Kill it.
So(e.AbortJob(ctxOwner, jobID), ShouldBeNil)
// It is dead.
inv, err := e.getInvocation(c, jobID, invID)
So(err, ShouldBeNil)
So(inv.Status, ShouldEqual, task.StatusAborted)
// The job moved on with its life.
job, err := e.getJob(c, jobID)
So(err, ShouldBeNil)
So(job.State.State, ShouldEqual, JobStateSuspended)
So(job.State.InvocationID, ShouldEqual, 0)
})
Convey("AbortJob kills queued invocation", func() {
So(e.AbortJob(ctxOwner, jobID), ShouldBeNil)
// The job moved on with its life.
job, err := e.getJob(c, jobID)
So(err, ShouldBeNil)
So(job.State.State, ShouldEqual, JobStateSuspended)
So(job.State.InvocationID, ShouldEqual, 0)
})
Convey("AbortJob fails on non-existing job", func() {
So(e.AbortJob(ctxOwner, "not/exists"), ShouldResemble, ErrNoSuchJob)
})
})
}
func TestAddTimer(t *testing.T) {
Convey("with mock job", t, func() {
c := newTestContext(epoch)
e, mgr := newTestEngine()
// A job in "QUEUED" state (about to run an invocation).
const jobID = "abc/1"
const invNonce = int64(12345)
prepareQueuedJob(c, jobID, invNonce)
Convey("AddTimer works", func() {
// Start an invocation that adds a timer.
mgr.launchTask = func(ctx context.Context, ctl task.Controller) error {
ctl.AddTimer(ctx, time.Minute, "timer-name", []byte{1, 2, 3})
ctl.State().Status = task.StatusRunning
return nil
}
So(e.startInvocation(c, jobID, invNonce, "", 0), ShouldBeNil)
// The job is running.
job, err := e.getJob(c, jobID)
So(err, ShouldBeNil)
So(job.State.State, ShouldEqual, JobStateRunning)
// Added a task to the timers task queue.
tasks := tq.GetTestable(c).GetScheduledTasks()["timers-q"]
So(len(tasks), ShouldEqual, 1)
var tqt *tq.Task
for _, tqt = range tasks {
}
So(tqt.ETA, ShouldResemble, clock.Now(c).Add(time.Minute))
// Verify task body.
payload := actionTaskPayload{}
So(json.Unmarshal(tqt.Payload, &payload), ShouldBeNil)
So(payload, ShouldResemble, actionTaskPayload{
JobID: "abc/1",
InvID: 9200093523825174512,
InvTimer: &invocationTimer{
Delay: time.Minute,
Name: "timer-name",
Payload: []byte{1, 2, 3},
},
})
// Clear the queue.
tq.GetTestable(c).ResetTasks()
// Time comes to execute the task.
mgr.handleTimer = func(ctx context.Context, ctl task.Controller, name string, payload []byte) error {
So(name, ShouldEqual, "timer-name")
So(payload, ShouldResemble, []byte{1, 2, 3})
ctl.AddTimer(ctx, time.Minute, "ignored-timer", nil)
ctl.State().Status = task.StatusSucceeded
return nil
}
clock.Get(c).(testclock.TestClock).Add(time.Minute)
So(e.ExecuteSerializedAction(c, tqt.Payload, 0), ShouldBeNil)
// The job has finished (by timer handler). Moves back to SUSPENDED state.
job, err = e.getJob(c, jobID)
So(err, ShouldBeNil)
So(job.State.State, ShouldEqual, JobStateSuspended)
// No new timers added for finished job.
tasks = tq.GetTestable(c).GetScheduledTasks()["timers-q"]
So(len(tasks), ShouldEqual, 0)
})
})
}
func TestTrimDebugLog(t *testing.T) {
ctx := clock.Set(context.Background(), testclock.New(epoch))
junk := strings.Repeat("a", 1000)
genLines := func(start, end int) string {
inv := Invocation{}
for i := start; i < end; i++ {
inv.debugLog(ctx, "Line %d - %s", i, junk)
}
return inv.DebugLog
}
Convey("small log is not trimmed", t, func() {
inv := Invocation{
DebugLog: genLines(0, 100),
}
inv.trimDebugLog()
So(inv.DebugLog, ShouldEqual, genLines(0, 100))
})
Convey("huge log is trimmed", t, func() {
inv := Invocation{
DebugLog: genLines(0, 500),
}
inv.trimDebugLog()
So(inv.DebugLog, ShouldEqual,
genLines(0, 94)+"--- the log has been cut here ---\n"+genLines(400, 500))
})
Convey("writing lines to huge log and trimming", t, func() {
inv := Invocation{
DebugLog: genLines(0, 500),
}
inv.trimDebugLog()
for i := 0; i < 10; i++ {
inv.debugLog(ctx, "Line %d - %s", i, junk)
inv.trimDebugLog()
}
// Still single cut only. New 10 lines are at the end.
So(inv.DebugLog, ShouldEqual,
genLines(0, 94)+"--- the log has been cut here ---\n"+genLines(410, 500)+genLines(0, 10))
})
Convey("one huge line", t, func() {
inv := Invocation{
DebugLog: strings.Repeat("z", 300000),
}
inv.trimDebugLog()
const msg = "\n--- the log has been cut here ---\n"
So(inv.DebugLog, ShouldEqual, strings.Repeat("z", debugLogSizeLimit-len(msg))+msg)
})
}
////
func newTestContext(now time.Time) context.Context {
c := memory.Use(context.Background())
c = clock.Set(c, testclock.New(now))
c = mathrand.Set(c, rand.New(rand.NewSource(1000)))
c = testsecrets.Use(c)
ds.GetTestable(c).AddIndexes(&ds.IndexDefinition{
Kind: "Job",
SortBy: []ds.IndexColumn{
{Property: "Enabled"},
{Property: "ProjectID"},
},
})
ds.GetTestable(c).CatchupIndexes()
tq.GetTestable(c).CreateQueue("timers-q")
tq.GetTestable(c).CreateQueue("invs-q")
return c
}
func newTestEngine() (*engineImpl, *fakeTaskManager) {
mgr := &fakeTaskManager{}
cat := catalog.New("scheduler.cfg")
cat.RegisterTaskManager(mgr)
return NewEngine(Config{
Catalog: cat,
TimersQueuePath: "/timers",
TimersQueueName: "timers-q",
InvocationsQueuePath: "/invs",
InvocationsQueueName: "invs-q",
PubSubPushPath: "/push-url",
}).(*engineImpl), mgr
}
////
// fakeTaskManager implement task.Manager interface.
type fakeTaskManager struct {
launchTask func(ctx context.Context, ctl task.Controller) error
handleNotification func(ctx context.Context, msg *pubsub.PubsubMessage) error
handleTimer func(ctx context.Context, ctl task.Controller, name string, payload []byte) error
}
func (m *fakeTaskManager) Name() string {
return "fake"
}
func (m *fakeTaskManager) ProtoMessageType() proto.Message {
return (*messages.NoopTask)(nil)
}
func (m *fakeTaskManager) Traits() task.Traits {
return task.Traits{}
}
func (m *fakeTaskManager) ValidateProtoMessage(msg proto.Message) error {
return nil
}
func (m *fakeTaskManager) LaunchTask(c context.Context, ctl task.Controller) error {
return m.launchTask(c, ctl)
}
func (m *fakeTaskManager) AbortTask(c context.Context, ctl task.Controller) error {
return nil
}
func (m *fakeTaskManager) HandleNotification(c context.Context, ctl task.Controller, msg *pubsub.PubsubMessage) error {
return m.handleNotification(c, msg)
}
func (m fakeTaskManager) HandleTimer(c context.Context, ctl task.Controller, name string, payload []byte) error {
return m.handleTimer(c, ctl, name, payload)
}
////
func sortedJobIds(jobs []*Job) []string {
ids := stringset.New(len(jobs))
for _, j := range jobs {
ids.Add(j.JobID)
}
asSlice := ids.ToSlice()
sort.Strings(asSlice)
return asSlice
}
// prepareQueuedJob makes datastore entries for a job in QUEUED state.
func prepareQueuedJob(c context.Context, jobID string, invNonce int64) {
taskBlob, err := proto.Marshal(&messages.TaskDefWrapper{
Noop: &messages.NoopTask{},
})
if err != nil {
panic(err)
}
chunks := strings.Split(jobID, "/")
err = ds.Put(c, &Job{
JobID: jobID,
ProjectID: chunks[0],
Enabled: true,
Acls: acl.GrantsByRole{Owners: []string{"group:owners"}, Readers: []string{"group:readers"}},
Task: taskBlob,
Schedule: "triggered",
State: JobState{
State: JobStateQueued,
InvocationNonce: invNonce,
},
})
if err != nil {
panic(err)
}
}
func noopTaskBytes() []byte {
buf, _ := proto.Marshal(&messages.TaskDefWrapper{Noop: &messages.NoopTask{}})
return buf
}
func allJobs(c context.Context) []Job {
ds.GetTestable(c).CatchupIndexes()
entities := []Job{}
if err := ds.GetAll(c, ds.NewQuery("Job"), &entities); err != nil {
panic(err)
}
// Strip UTC location pointers from zero time.Time{} so that ShouldResemble
// can compare it to default time.Time{}. nil location is UTC too.
for i := range entities {
ent := &entities[i]
if ent.State.InvocationTime.IsZero() {
ent.State.InvocationTime = time.Time{}
}
if ent.State.TickTime.IsZero() {
ent.State.TickTime = time.Time{}
}
}
return entities
}
func ensureZeroTasks(c context.Context, q string) {
tqt := tq.GetTestable(c)
tasks := tqt.GetScheduledTasks()[q]
So(tasks == nil || len(tasks) == 0, ShouldBeTrue)
}
func ensureOneTask(c context.Context, q string) *tq.Task {
tqt := tq.GetTestable(c)
tasks := tqt.GetScheduledTasks()[q]
So(len(tasks), ShouldEqual, 1)
for _, t := range tasks {
return t
}
return nil
}