Google Git
Sign in
skia / buildbot / refs/heads/main / . / task_scheduler / go / scheduling / task_scheduler_test.go
blob: 94287203c26a1d8db2f596829f90890cd16d4544 [file] [log] [blame]
package scheduling
import (
"context"
"encoding/json"
"errors"
"fmt"
"math"
"os"
"path"
"sort"
"strings"
"sync"
"testing"
"time"
"cloud.google.com/go/storage"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
swarming_api "go.chromium.org/luci/common/api/swarming/swarming/v1"
"go.skia.org/infra/go/cas/mocks"
"go.skia.org/infra/go/deepequal"
"go.skia.org/infra/go/deepequal/assertdeep"
ftestutils "go.skia.org/infra/go/firestore/testutils"
"go.skia.org/infra/go/gcs/mem_gcsclient"
"go.skia.org/infra/go/git"
"go.skia.org/infra/go/git/repograph"
"go.skia.org/infra/go/git/testutils/mem_git"
"go.skia.org/infra/go/gitiles"
"go.skia.org/infra/go/gitstore"
"go.skia.org/infra/go/gitstore/mem_gitstore"
"go.skia.org/infra/go/mockhttpclient"
"go.skia.org/infra/go/now"
"go.skia.org/infra/go/sktest"
"go.skia.org/infra/go/swarming"
"go.skia.org/infra/go/testutils"
"go.skia.org/infra/go/util"
"go.skia.org/infra/go/vcsinfo"
"go.skia.org/infra/task_scheduler/go/db"
"go.skia.org/infra/task_scheduler/go/db/cache"
cache_mocks "go.skia.org/infra/task_scheduler/go/db/cache/mocks"
"go.skia.org/infra/task_scheduler/go/db/memory"
"go.skia.org/infra/task_scheduler/go/skip_tasks"
"go.skia.org/infra/task_scheduler/go/specs"
"go.skia.org/infra/task_scheduler/go/task_cfg_cache"
tcc_mocks "go.skia.org/infra/task_scheduler/go/task_cfg_cache/mocks"
tcc_testutils "go.skia.org/infra/task_scheduler/go/task_cfg_cache/testutils"
swarming_task_execution "go.skia.org/infra/task_scheduler/go/task_execution/swarming"
swarming_testutils "go.skia.org/infra/task_scheduler/go/testutils"
"go.skia.org/infra/task_scheduler/go/types"
"go.skia.org/infra/task_scheduler/go/window"
window_mocks "go.skia.org/infra/task_scheduler/go/window/mocks"
)
const (
scoreDelta = 0.000001
)
var (
androidTaskDims = map[string]string{
"pool": "Skia",
"os": "Android",
"device_type": "grouper",
}
linuxTaskDims = map[string]string{
"os": "Ubuntu",
"pool": "Skia",
}
androidBotDims = map[string][]string{
"pool": {"Skia"},
"os": {"Android"},
"device_type": {"grouper"},
}
linuxBotDims = map[string][]string{
"os": {"Ubuntu"},
"pool": {"Skia"},
}
// The following are commits and RepoStates used in tests. The commit
// hashes were chosen to be equal to those generated by MemGit, but the
// contents are completely arbitrary.
lc1 = &vcsinfo.LongCommit{
ShortCommit: &vcsinfo.ShortCommit{
Hash: mem_git.Commit0,
Author: "me@google.com",
Subject: "First Commit",
},
Body: "My first commit",
Timestamp: time.Unix(1571926390, 0),
Index: 0,
Branches: map[string]bool{
git.MainBranch: true,
},
}
lc2 = &vcsinfo.LongCommit{
ShortCommit: &vcsinfo.ShortCommit{
Hash: mem_git.Commit1,
Author: "you@google.com",
Subject: "Second Commit",
},
Body: "My second commit",
Parents: []string{lc1.Hash},
Timestamp: time.Unix(1571926450, 0),
Index: 1,
Branches: map[string]bool{
git.MainBranch: true,
},
}
ic1 = &vcsinfo.IndexCommit{
Hash: lc1.Hash,
Index: lc1.Index,
Timestamp: lc1.Timestamp,
}
ic2 = &vcsinfo.IndexCommit{
Hash: lc2.Hash,
Index: lc2.Index,
Timestamp: lc2.Timestamp,
}
rs1 = types.RepoState{
Repo: "fake.git",
Revision: lc1.Hash,
}
rs2 = types.RepoState{
Repo: "fake.git",
Revision: lc2.Hash,
}
)
func makeTask(ctx context.Context, name, repo, revision string) *types.Task {
return &types.Task{
Commits: []string{revision},
Created: now.Now(ctx),
TaskKey: types.TaskKey{
RepoState: types.RepoState{
Repo: repo,
Revision: revision,
},
Name: name,
},
MaxAttempts: types.DEFAULT_MAX_TASK_ATTEMPTS,
SwarmingTaskId: "swarmid",
}
}
func makeSwarmingRpcsTaskRequestMetadata(t *testing.T, task *types.Task, dims map[string]string) *swarming_api.SwarmingRpcsTaskRequestMetadata {
tag := func(k, v string) string {
return fmt.Sprintf("%s:%s", k, v)
}
ts := func(t time.Time) string {
if util.TimeIsZero(t) {
return ""
}
return t.Format(swarming.TIMESTAMP_FORMAT)
}
abandoned := ""
state := swarming.TASK_STATE_PENDING
failed := false
switch task.Status {
case types.TASK_STATUS_MISHAP:
state = swarming.TASK_STATE_BOT_DIED
abandoned = ts(task.Finished)
case types.TASK_STATUS_RUNNING:
state = swarming.TASK_STATE_RUNNING
case types.TASK_STATUS_FAILURE:
state = swarming.TASK_STATE_COMPLETED
failed = true
case types.TASK_STATUS_SUCCESS:
state = swarming.TASK_STATE_COMPLETED
case types.TASK_STATUS_PENDING:
// noop
default:
require.FailNow(t, "Unknown task status: %s", task.Status)
}
tags := []string{
tag(types.SWARMING_TAG_ID, task.Id),
tag(types.SWARMING_TAG_FORCED_JOB_ID, task.ForcedJobId),
tag(types.SWARMING_TAG_NAME, task.Name),
tag(swarming.DIMENSION_POOL_KEY, swarming.DIMENSION_POOL_VALUE_SKIA),
tag(types.SWARMING_TAG_REPO, task.Repo),
tag(types.SWARMING_TAG_REVISION, task.Revision),
}
for _, p := range task.ParentTaskIds {
tags = append(tags, tag(types.SWARMING_TAG_PARENT_TASK_ID, p))
}
dimensions := make([]*swarming_api.SwarmingRpcsStringPair, 0, len(dims))
for k, v := range dims {
dimensions = append(dimensions, &swarming_api.SwarmingRpcsStringPair{
Key: k,
Value: v,
})
}
var casOutput *swarming_api.SwarmingRpcsCASReference
if task.IsolatedOutput != "" {
var err error
casOutput, err = swarming.MakeCASReference(task.IsolatedOutput, "fake-cas-instance")
require.NoError(t, err)
}
return &swarming_api.SwarmingRpcsTaskRequestMetadata{
Request: &swarming_api.SwarmingRpcsTaskRequest{
CreatedTs: ts(task.Created),
TaskSlices: []*swarming_api.SwarmingRpcsTaskSlice{
{
Properties: &swarming_api.SwarmingRpcsTaskProperties{
Dimensions: dimensions,
},
},
},
Tags: tags,
},
TaskId: task.SwarmingTaskId,
TaskResult: &swarming_api.SwarmingRpcsTaskResult{
AbandonedTs: abandoned,
BotId: task.SwarmingBotId,
CreatedTs: ts(task.Created),
CompletedTs: ts(task.Finished),
Failure: failed,
CasOutputRoot: casOutput,
StartedTs: ts(task.Started),
State: state,
Tags: tags,
TaskId: task.SwarmingTaskId,
},
}
}
// Insert the given data into the caches.
func fillCaches(t sktest.TestingT, ctx context.Context, taskCfgCache task_cfg_cache.TaskCfgCache, rs types.RepoState, cfg *specs.TasksCfg) {
require.NoError(t, taskCfgCache.Set(ctx, rs, cfg, nil))
}
// Add jobs for the given RepoStates. Assumes that the RepoStates have already
// been added to the caches.
func insertJobs(t sktest.TestingT, ctx context.Context, s *TaskScheduler, rss ...types.RepoState) {
jobs := []*types.Job{}
for _, rs := range rss {
cfg, cachedErr, err := s.taskCfgCache.Get(ctx, rs)
require.NoError(t, err)
require.NoError(t, cachedErr)
for name := range cfg.Jobs {
j, err := task_cfg_cache.MakeJob(ctx, s.taskCfgCache, rs, name)
require.NoError(t, err)
jobs = append(jobs, j)
}
}
require.NoError(t, s.putJobsInChunks(ctx, jobs))
}
// Common setup for TaskScheduler tests.
func setup(t *testing.T) (context.Context, *mem_git.MemGit, *memory.InMemoryDB, *swarming_testutils.TestClient, *TaskScheduler, *mockhttpclient.URLMock, *mocks.CAS, func()) {
ctx, cancel := context.WithCancel(context.Background())
tmp, err := os.MkdirTemp("", "")
require.NoError(t, err)
d := memory.NewInMemoryDB()
swarmingClient := swarming_testutils.NewTestClient()
urlMock := mockhttpclient.NewURLMock()
mg, repo := newMemRepo(t)
hashes := mg.CommitN(2)
// Sanity check.
require.Equal(t, lc1.Hash, hashes[1])
require.Equal(t, lc2.Hash, hashes[0])
repos := repograph.Map{
rs1.Repo: repo,
}
btProject, btInstance, btCleanup := tcc_testutils.SetupBigTable(t)
taskCfgCache, err := task_cfg_cache.NewTaskCfgCache(ctx, repos, btProject, btInstance, nil)
require.NoError(t, err)
// Cache the RepoStates. This is normally done by the JobCreator.
fillCaches(t, ctx, taskCfgCache, rs1, tcc_testutils.TasksCfg1)
fillCaches(t, ctx, taskCfgCache, rs2, tcc_testutils.TasksCfg2)
cas := &mocks.CAS{}
cas.On("Close").Return(nil)
// Go ahead and mock the single-input merge calls.
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.CompileCASDigest}).Return(tcc_testutils.CompileCASDigest, nil)
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.TestCASDigest}).Return(tcc_testutils.TestCASDigest, nil)
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.PerfCASDigest}).Return(tcc_testutils.PerfCASDigest, nil)
taskExec := swarming_task_execution.NewSwarmingTaskExecutor(swarmingClient, "fake-cas-instance", "")
taskExecs := map[string]types.TaskExecutor{
types.TaskExecutor_Swarming: taskExec,
types.TaskExecutor_UseDefault: taskExec,
}
s, err := NewTaskScheduler(ctx, d, nil, time.Duration(math.MaxInt64), 0, repos, cas, "fake-cas-instance", taskExecs, urlMock.Client(), 1.0, swarming.POOLS_PUBLIC, "", taskCfgCache, nil, mem_gcsclient.New("diag_unit_tests"), btInstance, false)
require.NoError(t, err)
// Insert jobs. This is normally done by the JobCreator.
insertJobs(t, ctx, s, rs1, rs2)
return ctx, mg, d, swarmingClient, s, urlMock, cas, func() {
testutils.AssertCloses(t, s)
testutils.RemoveAll(t, tmp)
btCleanup()
cancel()
}
}
// runMainLoop calls s.MainLoop, asserts there was no error, and waits for
// background goroutines to finish.
func runMainLoop(t *testing.T, s *TaskScheduler, ctx context.Context) {
require.NoError(t, s.MainLoop(ctx))
s.testWaitGroup.Wait()
}
func lastDiagnostics(t *testing.T, s *TaskScheduler) taskSchedulerMainLoopDiagnostics {
ctx := context.Background()
lastname := ""
require.NoError(t, s.diagClient.AllFilesInDirectory(ctx, path.Join(s.diagInstance, GCS_MAIN_LOOP_DIAGNOSTICS_DIR), func(item *storage.ObjectAttrs) error {
if lastname == "" || item.Name > lastname {
lastname = item.Name
}
return nil
}))
require.NotEqual(t, lastname, "")
reader, err := s.diagClient.FileReader(ctx, lastname)
require.NoError(t, err)
defer testutils.AssertCloses(t, reader)
rv := taskSchedulerMainLoopDiagnostics{}
require.NoError(t, json.NewDecoder(reader).Decode(&rv))
return rv
}
func TestFindTaskCandidatesForJobs(t *testing.T) {
ctx, _, _, _, s, _, _, cleanup := setup(t)
defer cleanup()
test := func(jobs []*types.Job, expect map[types.TaskKey]*TaskCandidate) {
actual, err := s.findTaskCandidatesForJobs(ctx, jobs)
require.NoError(t, err)
assertdeep.Equal(t, expect, actual)
}
cfg1, cachedErr, err := s.taskCfgCache.Get(ctx, rs1)
require.NoError(t, err)
require.NoError(t, cachedErr)
cfg2, cachedErr, err := s.taskCfgCache.Get(ctx, rs2)
require.NoError(t, err)
require.NoError(t, cachedErr)
// Run on an empty job list, ensure empty list returned.
test([]*types.Job{}, map[types.TaskKey]*TaskCandidate{})
currentTime := now.Now(ctx).UTC()
// Run for one job, ensure that we get the right set of task specs
// returned (ie. all dependencies and their dependencies).
j1 := &types.Job{
Created: currentTime,
Id: "job1id",
Name: tcc_testutils.TestTaskName,
Dependencies: map[string][]string{tcc_testutils.TestTaskName: {tcc_testutils.BuildTaskName}, tcc_testutils.BuildTaskName: {}},
Priority: 0.5,
RepoState: rs1.Copy(),
}
tc1 := &TaskCandidate{
CasDigests: []string{tcc_testutils.CompileCASDigest},
Jobs: []*types.Job{j1},
TaskKey: types.TaskKey{
RepoState: rs1.Copy(),
Name: tcc_testutils.BuildTaskName,
},
TaskSpec: cfg1.Tasks[tcc_testutils.BuildTaskName].Copy(),
}
tc2 := &TaskCandidate{
CasDigests: []string{tcc_testutils.TestCASDigest},
Jobs: []*types.Job{j1},
TaskKey: types.TaskKey{
RepoState: rs1.Copy(),
Name: tcc_testutils.TestTaskName,
},
TaskSpec: cfg1.Tasks[tcc_testutils.TestTaskName].Copy(),
}
test([]*types.Job{j1}, map[types.TaskKey]*TaskCandidate{
tc1.TaskKey: tc1,
tc2.TaskKey: tc2,
})
// Add a job, ensure that its dependencies are added and that the right
// dependencies are de-duplicated.
j2 := &types.Job{
Created: currentTime,
Id: "job2id",
Name: tcc_testutils.TestTaskName,
Dependencies: map[string][]string{tcc_testutils.TestTaskName: {tcc_testutils.BuildTaskName}, tcc_testutils.BuildTaskName: {}},
Priority: 0.6,
RepoState: rs2,
}
j3 := &types.Job{
Created: currentTime,
Id: "job3id",
Name: tcc_testutils.PerfTaskName,
Dependencies: map[string][]string{tcc_testutils.PerfTaskName: {tcc_testutils.BuildTaskName}, tcc_testutils.BuildTaskName: {}},
Priority: 0.6,
RepoState: rs2,
}
tc3 := &TaskCandidate{
CasDigests: []string{tcc_testutils.CompileCASDigest},
Jobs: []*types.Job{j2, j3},
TaskKey: types.TaskKey{
RepoState: rs2.Copy(),
Name: tcc_testutils.BuildTaskName,
},
TaskSpec: cfg2.Tasks[tcc_testutils.BuildTaskName].Copy(),
}
tc4 := &TaskCandidate{
CasDigests: []string{tcc_testutils.TestCASDigest},
Jobs: []*types.Job{j2},
TaskKey: types.TaskKey{
RepoState: rs2.Copy(),
Name: tcc_testutils.TestTaskName,
},
TaskSpec: cfg2.Tasks[tcc_testutils.TestTaskName].Copy(),
}
tc5 := &TaskCandidate{
CasDigests: []string{tcc_testutils.PerfCASDigest},
Jobs: []*types.Job{j3},
TaskKey: types.TaskKey{
RepoState: rs2.Copy(),
Name: tcc_testutils.PerfTaskName,
},
TaskSpec: cfg2.Tasks[tcc_testutils.PerfTaskName].Copy(),
}
allCandidates := map[types.TaskKey]*TaskCandidate{
tc1.TaskKey: tc1,
tc2.TaskKey: tc2,
tc3.TaskKey: tc3,
tc4.TaskKey: tc4,
tc5.TaskKey: tc5,
}
test([]*types.Job{j1, j2, j3}, allCandidates)
// Finish j3, ensure that its task specs no longer show up.
delete(allCandidates, j3.MakeTaskKey(tcc_testutils.PerfTaskName))
// This is hacky, but findTaskCandidatesForJobs accepts an already-
// filtered list of jobs, so we have to pretend it never existed.
tc3.Jobs = tc3.Jobs[:1]
test([]*types.Job{j1, j2}, allCandidates)
// Ensure that we don't generate candidates for jobs at nonexistent commits.
j4 := &types.Job{
Created: currentTime,
Id: "job4id",
Name: tcc_testutils.PerfTaskName,
Dependencies: map[string][]string{tcc_testutils.PerfTaskName: {tcc_testutils.BuildTaskName}, tcc_testutils.BuildTaskName: {}},
Priority: 0.6,
RepoState: types.RepoState{
Repo: rs2.Repo,
Revision: "aaaaabbbbbcccccdddddeeeeefffff1111122222",
},
}
test([]*types.Job{j4}, map[types.TaskKey]*TaskCandidate{})
}
func TestFilterTaskCandidates(t *testing.T) {
ctx, _, _, _, s, _, _, cleanup := setup(t)
defer cleanup()
c1 := rs1.Revision
c2 := rs2.Revision
// Fake out the initial candidates.
k1 := types.TaskKey{
RepoState: rs1,
Name: tcc_testutils.BuildTaskName,
}
k2 := types.TaskKey{
RepoState: rs1,
Name: tcc_testutils.TestTaskName,
}
k3 := types.TaskKey{
RepoState: rs2,
Name: tcc_testutils.BuildTaskName,
}
k4 := types.TaskKey{
RepoState: rs2,
Name: tcc_testutils.TestTaskName,
}
k5 := types.TaskKey{
RepoState: rs2,
Name: tcc_testutils.PerfTaskName,
}
candidates := map[types.TaskKey]*TaskCandidate{
k1: {
TaskKey: k1,
TaskSpec: &specs.TaskSpec{},
},
k2: {
TaskKey: k2,
TaskSpec: &specs.TaskSpec{
Dependencies: []string{tcc_testutils.BuildTaskName},
},
},
k3: {
TaskKey: k3,
TaskSpec: &specs.TaskSpec{},
},
k4: {
TaskKey: k4,
TaskSpec: &specs.TaskSpec{
Dependencies: []string{tcc_testutils.BuildTaskName},
},
},
k5: {
TaskKey: k5,
TaskSpec: &specs.TaskSpec{
Dependencies: []string{tcc_testutils.BuildTaskName},
},
},
}
clearDiagnostics := func(candidates map[types.TaskKey]*TaskCandidate) {
for _, c := range candidates {
c.Diagnostics = nil
}
}
// Check the initial set of task candidates. The two Build tasks
// should be the only ones available.
c, err := s.filterTaskCandidates(ctx, candidates)
require.NoError(t, err)
require.Len(t, c, 1)
require.Equal(t, 1, len(c[rs1.Repo]))
require.Equal(t, 2, len(c[rs1.Repo][tcc_testutils.BuildTaskName]))
for _, byRepo := range c {
for _, byName := range byRepo {
for _, candidate := range byName {
require.Equal(t, candidate.Name, tcc_testutils.BuildTaskName)
}
}
}
// Check filtering diagnostics. Non-Build tasks have unmet dependencies.
for _, candidate := range candidates {
if candidate.Name != tcc_testutils.BuildTaskName {
require.Equal(t, candidate.Diagnostics.Filtering.UnmetDependencies, []string{tcc_testutils.BuildTaskName})
}
}
// Insert a the Build task at c1 (1 dependent) into the database,
// transition through various states.
var t1 *types.Task
for _, byRepo := range c { // Order not guaranteed, find the right candidate.
for _, byName := range byRepo {
for _, candidate := range byName {
if candidate.Revision == c1 {
t1 = makeTask(ctx, candidate.Name, candidate.Repo, candidate.Revision)
break
}
}
}
}
require.NotNil(t, t1)
// We shouldn't duplicate pending or running tasks.
for _, status := range []types.TaskStatus{types.TASK_STATUS_PENDING, types.TASK_STATUS_RUNNING} {
clearDiagnostics(candidates)
t1.Status = status
require.NoError(t, s.putTask(ctx, t1))
c, err = s.filterTaskCandidates(ctx, candidates)
require.NoError(t, err)
require.Len(t, c, 1)
for _, byRepo := range c {
for _, byName := range byRepo {
for _, candidate := range byName {
require.Equal(t, candidate.Name, tcc_testutils.BuildTaskName)
require.Equal(t, c2, candidate.Revision)
}
}
}
// Check filtering diagnostics.
for _, candidate := range candidates {
if candidate.Name != tcc_testutils.BuildTaskName {
// Non-Build tasks have unmet dependencies.
require.Equal(t, candidate.Diagnostics.Filtering.UnmetDependencies, []string{tcc_testutils.BuildTaskName})
} else if candidate.Revision == c1 {
// Blocked by t1
require.Equal(t, candidate.Diagnostics.Filtering.SupersededByTask, t1.Id)
}
}
}
clearDiagnostics(candidates)
// The task failed. Ensure that its dependents are not candidates, but
// the task itself is back in the list of candidates, in case we want
// to retry.
t1.Status = types.TASK_STATUS_FAILURE
require.NoError(t, s.putTask(ctx, t1))
c, err = s.filterTaskCandidates(ctx, candidates)
require.NoError(t, err)
require.Len(t, c, 1)
for _, byRepo := range c {
require.Len(t, byRepo, 1)
for _, byName := range byRepo {
require.Len(t, byName, 2)
for _, candidate := range byName {
require.Equal(t, candidate.Name, tcc_testutils.BuildTaskName)
}
}
}
// Check filtering diagnostics.
for _, candidate := range candidates {
if candidate.Name != tcc_testutils.BuildTaskName {
// Non-Build tasks have unmet dependencies.
require.Equal(t, candidate.Diagnostics.Filtering.UnmetDependencies, []string{tcc_testutils.BuildTaskName})
}
}
clearDiagnostics(candidates)
// The task succeeded. Ensure that its dependents are candidates and
// the task itself is not.
t1.Status = types.TASK_STATUS_SUCCESS
t1.IsolatedOutput = "fake isolated hash"
require.NoError(t, s.putTask(ctx, t1))
c, err = s.filterTaskCandidates(ctx, candidates)
require.NoError(t, err)
require.Len(t, c, 1)
for _, byRepo := range c {
require.Len(t, byRepo, 2)
for _, byName := range byRepo {
for _, candidate := range byName {
require.False(t, t1.Name == candidate.Name && t1.Revision == candidate.Revision)
}
}
}
// Candidate with k1 is blocked by t1.
require.Equal(t, candidates[k1].Diagnostics.Filtering.SupersededByTask, t1.Id)
clearDiagnostics(candidates)
// Create the other Build task.
var t2 *types.Task
for _, byRepo := range c {
for _, byName := range byRepo {
for _, candidate := range byName {
if candidate.Revision == c2 && strings.HasPrefix(candidate.Name, "Build-") {
t2 = makeTask(ctx, candidate.Name, candidate.Repo, candidate.Revision)
break
}
}
}
}
require.NotNil(t, t2)
t2.Status = types.TASK_STATUS_SUCCESS
t2.IsolatedOutput = "fake isolated hash"
require.NoError(t, s.putTask(ctx, t2))
// All test and perf tasks are now candidates, no build tasks.
c, err = s.filterTaskCandidates(ctx, candidates)
require.NoError(t, err)
require.Len(t, c, 1)
require.Equal(t, 2, len(c[rs1.Repo][tcc_testutils.TestTaskName]))
require.Equal(t, 1, len(c[rs1.Repo][tcc_testutils.PerfTaskName]))
for _, byRepo := range c {
for _, byName := range byRepo {
for _, candidate := range byName {
require.NotEqual(t, candidate.Name, tcc_testutils.BuildTaskName)
}
}
}
// Build candidates are blocked by completed tasks.
require.Equal(t, candidates[k1].Diagnostics.Filtering.SupersededByTask, t1.Id)
require.Equal(t, candidates[k3].Diagnostics.Filtering.SupersededByTask, t2.Id)
clearDiagnostics(candidates)
// Add a try job. Ensure that no deps have been incorrectly satisfied.
tryKey := k4.Copy()
tryKey.Server = "dummy-server"
tryKey.Issue = "dummy-issue"
tryKey.Patchset = "dummy-patchset"
candidates[tryKey] = &TaskCandidate{
TaskKey: tryKey,
TaskSpec: &specs.TaskSpec{
Dependencies: []string{tcc_testutils.BuildTaskName},
},
}
c, err = s.filterTaskCandidates(ctx, candidates)
require.NoError(t, err)
require.Len(t, c, 1)
require.Equal(t, 2, len(c[rs1.Repo][tcc_testutils.TestTaskName]))
require.Equal(t, 1, len(c[rs1.Repo][tcc_testutils.PerfTaskName]))
for _, byRepo := range c {
for _, byName := range byRepo {
for _, candidate := range byName {
require.NotEqual(t, candidate.Name, tcc_testutils.BuildTaskName)
require.False(t, candidate.IsTryJob())
}
}
}
// Check diagnostics for tryKey
require.Equal(t, candidates[tryKey].Diagnostics.Filtering.UnmetDependencies, []string{tcc_testutils.BuildTaskName})
}
// processTaskCandidate is a helper function for processing a single task
// candidate.
func processTaskCandidate(ctx context.Context, s *TaskScheduler, c *TaskCandidate) error {
candidates := map[string]map[string][]*TaskCandidate{
c.Repo: {
c.Name: []*TaskCandidate{c},
},
}
_, err := s.processTaskCandidates(ctx, candidates)
return err
}
func TestProcessTaskCandidate(t *testing.T) {
_, _, _, _, s, _, _, cleanup := setup(t)
defer cleanup()
currentTime := time.Unix(0, 1470674884000000)
ctx := now.TimeTravelingContext(currentTime)
checkDiagTryForced := func(c *TaskCandidate) {
require.NotNil(t, c.Diagnostics)
diag := c.Diagnostics.Scoring
require.NotNil(t, diag)
require.Equal(t, c.Jobs[0].Priority, diag.Priority)
require.Equal(t, currentTime.Sub(c.Jobs[0].Created).Hours(), diag.JobCreatedHours)
// The remaining fields should always be 0 for try/forced jobs.
require.Equal(t, 0, diag.StoleFromCommits)
require.Equal(t, 0.0, diag.TestednessIncrease)
require.Equal(t, 0.0, diag.TimeDecay)
}
tryjobRs := types.RepoState{
Patch: types.Patch{
Server: "my-server",
Issue: "my-issue",
Patchset: "my-patchset",
},
Repo: rs1.Repo,
Revision: rs1.Revision,
}
tryjob := &types.Job{
Id: "tryjobId",
Created: currentTime.Add(-1 * time.Hour),
Name: "job",
Priority: 0.5,
RepoState: tryjobRs,
}
c := &TaskCandidate{
Jobs: []*types.Job{tryjob},
TaskKey: types.TaskKey{
Name: tcc_testutils.BuildTaskName,
RepoState: tryjobRs,
},
TaskSpec: tcc_testutils.TasksCfg1.Tasks[tcc_testutils.BuildTaskName],
}
require.NoError(t, processTaskCandidate(ctx, s, c))
// Try job candidates have a specific score and no blamelist.
require.InDelta(t, (CANDIDATE_SCORE_TRY_JOB+1.0)*0.5, c.Score, scoreDelta)
require.Nil(t, c.Commits)
checkDiagTryForced(c)
// Retries are scored lower.
c = &TaskCandidate{
Attempt: 1,
Jobs: []*types.Job{tryjob},
TaskKey: types.TaskKey{
Name: tcc_testutils.BuildTaskName,
RepoState: tryjobRs,
},
TaskSpec: tcc_testutils.TasksCfg1.Tasks[tcc_testutils.BuildTaskName],
}
require.NoError(t, processTaskCandidate(ctx, s, c))
require.InDelta(t, (CANDIDATE_SCORE_TRY_JOB+1.0)*0.5*CANDIDATE_SCORE_TRY_JOB_RETRY_MULTIPLIER, c.Score, scoreDelta)
require.Nil(t, c.Commits)
checkDiagTryForced(c)
forcedJob := &types.Job{
Id: "forcedJobId",
Created: currentTime.Add(-2 * time.Hour),
Name: tcc_testutils.BuildTaskName,
Priority: 0.5,
RepoState: rs2,
}
// Manually forced candidates have a blamelist and a specific score.
c = &TaskCandidate{
Jobs: []*types.Job{forcedJob},
TaskKey: types.TaskKey{
Name: tcc_testutils.BuildTaskName,
RepoState: rs2,
ForcedJobId: forcedJob.Id,
},
TaskSpec: tcc_testutils.TasksCfg2.Tasks[tcc_testutils.BuildTaskName],
}
require.NoError(t, processTaskCandidate(ctx, s, c))
require.InDelta(t, (CANDIDATE_SCORE_FORCE_RUN+2.0)*0.5, c.Score, scoreDelta)
require.Equal(t, 2, len(c.Commits))
checkDiagTryForced(c)
// All other candidates have a blamelist and a time-decayed score.
regularJob := &types.Job{
Id: "regularJobId",
Created: currentTime.Add(-1 * time.Hour),
Name: tcc_testutils.BuildTaskName,
Priority: 0.5,
RepoState: rs2,
}
c = &TaskCandidate{
Jobs: []*types.Job{regularJob},
TaskKey: types.TaskKey{
Name: tcc_testutils.BuildTaskName,
RepoState: rs2,
},
TaskSpec: tcc_testutils.TasksCfg2.Tasks[tcc_testutils.BuildTaskName],
}
require.NoError(t, processTaskCandidate(ctx, s, c))
require.True(t, c.Score > 0)
require.Equal(t, 2, len(c.Commits))
diag := c.GetDiagnostics().Scoring
require.Equal(t, 0.5, diag.Priority)
require.Equal(t, 1.0, diag.JobCreatedHours)
require.Equal(t, 0, diag.StoleFromCommits)
require.Equal(t, 3.5, diag.TestednessIncrease)
require.InDelta(t, 1.0, diag.TimeDecay, scoreDelta)
}
func TestRegularJobRetryScoring(t *testing.T) {
_, _, _, _, s, _, _, cleanup := setup(t)
defer cleanup()
ctx := now.TimeTravelingContext(time.Date(2021, time.October, 15, 0, 0, 0, 0, time.UTC))
currentTime := now.Now(ctx)
checkDiag := func(c *TaskCandidate) {
require.NotNil(t, c.Diagnostics)
diag := c.Diagnostics.Scoring
require.NotNil(t, diag)
// All candidates in this test have a single Job.
require.Equal(t, c.Jobs[0].Priority, diag.Priority)
require.InDelta(t, currentTime.Sub(c.Jobs[0].Created).Hours(), diag.JobCreatedHours, scoreDelta)
// The commits are added close enough to "now" that there is no time decay.
require.Equal(t, 1.0, diag.TimeDecay)
}
j1 := &types.Job{
Id: "regularJobId1",
Created: currentTime.Add(-1 * time.Hour),
Name: tcc_testutils.BuildTaskName,
Priority: 0.5,
RepoState: rs1,
}
j2 := &types.Job{
Id: "regularJobId2",
Created: currentTime.Add(-1 * time.Hour),
Name: tcc_testutils.BuildTaskName,
Priority: 0.5,
RepoState: rs2,
}
// Candidates at rs1 and rs2
c1 := &TaskCandidate{
Jobs: []*types.Job{j1},
TaskKey: types.TaskKey{
Name: tcc_testutils.BuildTaskName,
RepoState: rs1,
},
TaskSpec: tcc_testutils.TasksCfg1.Tasks[tcc_testutils.BuildTaskName],
}
c2 := &TaskCandidate{
Jobs: []*types.Job{j2},
TaskKey: types.TaskKey{
Name: tcc_testutils.BuildTaskName,
RepoState: rs2,
},
TaskSpec: tcc_testutils.TasksCfg2.Tasks[tcc_testutils.BuildTaskName],
}
// Regular task at HEAD with 2 commits has score 3.5 scaled by priority 0.5.
require.NoError(t, processTaskCandidate(ctx, s, c2))
require.InDelta(t, 3.5*0.5, c2.Score, scoreDelta)
require.Equal(t, 2, len(c2.Commits))
diag := c2.GetDiagnostics().Scoring
require.Equal(t, 0, diag.StoleFromCommits)
require.Equal(t, 3.5, diag.TestednessIncrease)
checkDiag(c2)
// Regular task at HEAD^ (no backfill) with 1 commit has score 2 scaled by
// priority 0.5.
require.NoError(t, processTaskCandidate(ctx, s, c1))
diag = c1.GetDiagnostics().Scoring
require.InDelta(t, 2*0.5, c1.Score, scoreDelta)
require.Equal(t, 1, len(c1.Commits))
require.Equal(t, 0, diag.StoleFromCommits)
require.Equal(t, 2.0, diag.TestednessIncrease)
checkDiag(c1)
// Add a task at rs2 that failed.
t2 := makeTask(ctx, c2.Name, c2.Repo, c2.Revision)
t2.Status = types.TASK_STATUS_FAILURE
t2.Commits = util.CopyStringSlice(c2.Commits)
require.NoError(t, s.putTask(ctx, t2))
// Update Attempt and RetryOf before calling processTaskCandidate.
c2.Attempt = 1
c2.RetryOf = t2.Id
// Retry task at rs2 with 2 commits for 2nd of 2 attempts has score 0.75
// scaled by priority 0.5.
require.NoError(t, processTaskCandidate(ctx, s, c2))
require.InDelta(t, 0.75*0.5, c2.Score, scoreDelta)
require.Equal(t, 2, len(c2.Commits))
diag = c2.GetDiagnostics().Scoring
require.Equal(t, 2, diag.StoleFromCommits)
require.Equal(t, 0.0, diag.TestednessIncrease)
checkDiag(c2)
// Regular task at rs1 (backfilling failed task) with 1 commit has score 1.25
// scaled by priority 0.5.
diag = &taskCandidateScoringDiagnostics{}
require.NoError(t, processTaskCandidate(ctx, s, c1))
diag = c1.GetDiagnostics().Scoring
require.InDelta(t, 1.25*0.5, c1.Score, scoreDelta)
require.Equal(t, 1, len(c1.Commits))
require.Equal(t, 2, diag.StoleFromCommits)
require.Equal(t, 0.5, diag.TestednessIncrease)
checkDiag(c1)
// Actually, the task at rs2 had a mishap.
t2.Status = types.TASK_STATUS_MISHAP
require.NoError(t, s.putTask(ctx, t2))
// Scores should be same as for FAILURE.
require.NoError(t, processTaskCandidate(ctx, s, c2))
diag = c2.GetDiagnostics().Scoring
require.InDelta(t, 0.75*0.5, c2.Score, scoreDelta)
require.Equal(t, 2, len(c2.Commits))
require.Equal(t, 2, diag.StoleFromCommits)
require.Equal(t, 0.0, diag.TestednessIncrease)
checkDiag(c2)
require.NoError(t, processTaskCandidate(ctx, s, c1))
diag = c1.GetDiagnostics().Scoring
require.InDelta(t, 1.25*0.5, c1.Score, scoreDelta)
require.Equal(t, 1, len(c1.Commits))
require.Equal(t, 2, diag.StoleFromCommits)
require.Equal(t, 0.5, diag.TestednessIncrease)
checkDiag(c1)
}
func TestProcessTaskCandidates(t *testing.T) {
ctx, _, _, _, s, _, _, cleanup := setup(t)
defer cleanup()
ts := now.Now(ctx)
// Processing of individual candidates is already tested; just verify
// that if we pass in a bunch of candidates they all get processed.
// The JobSpecs do not specify priority, so they use the default of 0.5.
assertProcessed := func(c *TaskCandidate) {
if c.IsTryJob() {
require.True(t, c.Score > CANDIDATE_SCORE_TRY_JOB*0.5)
require.Nil(t, c.Commits)
} else if c.IsForceRun() {
require.True(t, c.Score > CANDIDATE_SCORE_FORCE_RUN*0.5)
require.Equal(t, 2, len(c.Commits))
} else if c.Revision == rs2.Revision {
if c.Name == tcc_testutils.PerfTaskName {
require.InDelta(t, 2.0*0.5, c.Score, scoreDelta)
require.Equal(t, 1, len(c.Commits))
} else if c.Name == tcc_testutils.BuildTaskName {
// Already covered by the forced job, so zero score.
require.InDelta(t, 0, c.Score, scoreDelta)
// Scores below the BuildTask at rs1, so it has a blamelist of 1 commit.
require.Equal(t, 1, len(c.Commits))
} else {
require.InDelta(t, 3.5*0.5, c.Score, scoreDelta)
require.Equal(t, 2, len(c.Commits))
}
} else {
require.InDelta(t, 0.5*0.5, c.Score, scoreDelta) // These will be backfills.
require.Equal(t, 1, len(c.Commits))
}
}
testJob1 := &types.Job{
Id: "testJob1",
Created: ts,
Name: tcc_testutils.TestTaskName,
Priority: 0.5,
RepoState: rs1,
}
testJob2 := &types.Job{
Id: "testJob2",
Created: ts,
Name: tcc_testutils.TestTaskName,
Priority: 0.5,
RepoState: rs2,
}
perfJob2 := &types.Job{
Id: "perfJob2",
Created: ts,
Name: tcc_testutils.PerfTaskName,
Priority: 0.5,
RepoState: rs2,
}
forcedBuildJob2 := &types.Job{
Id: "forcedBuildJob2",
Created: ts,
Name: tcc_testutils.BuildTaskName,
Priority: 0.5,
RepoState: rs2,
}
tryjobRs := types.RepoState{
Patch: types.Patch{
Server: "my-server",
Issue: "my-issue",
Patchset: "my-patchset",
},
Repo: rs1.Repo,
Revision: rs1.Revision,
}
perfTryjob2 := &types.Job{
Id: "perfJob2",
Created: ts,
Name: tcc_testutils.PerfTaskName,
Priority: 0.5,
RepoState: tryjobRs,
}
candidates := map[string]map[string][]*TaskCandidate{
rs1.Repo: {
tcc_testutils.BuildTaskName: {
{
Jobs: []*types.Job{testJob1},
TaskKey: types.TaskKey{
RepoState: rs1,
Name: tcc_testutils.BuildTaskName,
},
TaskSpec: &specs.TaskSpec{},
},
{
Jobs: []*types.Job{testJob2, perfJob2},
TaskKey: types.TaskKey{
RepoState: rs2,
Name: tcc_testutils.BuildTaskName,
},
TaskSpec: &specs.TaskSpec{},
},
{
Jobs: []*types.Job{forcedBuildJob2},
TaskKey: types.TaskKey{
RepoState: rs2,
Name: tcc_testutils.BuildTaskName,
ForcedJobId: forcedBuildJob2.Id,
},
TaskSpec: &specs.TaskSpec{},
},
},
tcc_testutils.TestTaskName: {
{
Jobs: []*types.Job{testJob1},
TaskKey: types.TaskKey{
RepoState: rs1,
Name: tcc_testutils.TestTaskName,
},
TaskSpec: &specs.TaskSpec{},
},
{
Jobs: []*types.Job{testJob2},
TaskKey: types.TaskKey{
RepoState: rs2,
Name: tcc_testutils.TestTaskName,
},
TaskSpec: &specs.TaskSpec{},
},
},
tcc_testutils.PerfTaskName: {
{
Jobs: []*types.Job{perfJob2},
TaskKey: types.TaskKey{
RepoState: rs2,
Name: tcc_testutils.PerfTaskName,
},
TaskSpec: &specs.TaskSpec{},
},
{
Jobs: []*types.Job{perfTryjob2},
TaskKey: types.TaskKey{
RepoState: tryjobRs,
Name: tcc_testutils.PerfTaskName,
},
TaskSpec: &specs.TaskSpec{},
},
},
},
}
processed, err := s.processTaskCandidates(ctx, candidates)
require.NoError(t, err)
require.Len(t, processed, 7)
for _, c := range processed {
assertProcessed(c)
}
}
func TestTestedness(t *testing.T) {
tc := []struct {
in int
out float64
}{
{
in: -1,
out: -1.0,
},
{
in: 0,
out: 0.0,
},
{
in: 1,
out: 1.0,
},
{
in: 2,
out: 1.0 + 1.0/2.0,
},
{
in: 3,
out: 1.0 + float64(2.0)/float64(3.0),
},
{
in: 4,
out: 1.0 + 3.0/4.0,
},
{
in: 4096,
out: 1.0 + float64(4095)/float64(4096),
},
}
for i, c := range tc {
require.Equal(t, c.out, testedness(c.in), fmt.Sprintf("test case #%d", i))
}
}
func TestTestednessIncrease(t *testing.T) {
tc := []struct {
a int
b int
out float64
}{
// Invalid cases.
{
a: -1,
b: 10,
out: -1.0,
},
{
a: 10,
b: -1,
out: -1.0,
},
{
a: 0,
b: -1,
out: -1.0,
},
{
a: 0,
b: 0,
out: -1.0,
},
// Invalid because if we're re-running at already-tested commits
// then we should have a blamelist which is at most the size of
// the blamelist of the previous task. We naturally get negative
// testedness increase in these cases.
{
a: 2,
b: 1,
out: -0.5,
},
// Testing only new commits.
{
a: 1,
b: 0,
out: 1.0 + 1.0,
},
{
a: 2,
b: 0,
out: 2.0 + (1.0 + 1.0/2.0),
},
{
a: 3,
b: 0,
out: 3.0 + (1.0 + float64(2.0)/float64(3.0)),
},
{
a: 4096,
b: 0,
out: 4096.0 + (1.0 + float64(4095.0)/float64(4096.0)),
},
// Retries.
{
a: 1,
b: 1,
out: 0.0,
},
{
a: 2,
b: 2,
out: 0.0,
},
{
a: 3,
b: 3,
out: 0.0,
},
{
a: 4096,
b: 4096,
out: 0.0,
},
// Bisect/backfills.
{
a: 1,
b: 2,
out: 0.5, // (1 + 1) - (1 + 1/2)
},
{
a: 1,
b: 3,
out: float64(2.5) - (1.0 + float64(2.0)/float64(3.0)),
},
{
a: 5,
b: 10,
out: 2.0*(1.0+float64(4.0)/float64(5.0)) - (1.0 + float64(9.0)/float64(10.0)),
},
}
for i, c := range tc {
require.Equal(t, c.out, testednessIncrease(c.a, c.b), fmt.Sprintf("test case #%d", i))
}
}
func TestComputeBlamelist(t *testing.T) {
// Setup.
nowCtx := now.TimeTravelingContext(mem_git.BaseTime.Add(time.Hour))
ctx, cancel := context.WithCancel(nowCtx)
defer cancel()
mg, repo := newMemRepo(t)
repos := repograph.Map{
rs1.Repo: repo,
}
d := memory.NewInMemoryTaskDB()
w, err := window.New(ctx, 2*time.Hour, 0, nil)
cache, err := cache.NewTaskCache(ctx, d, w, nil)
require.NoError(t, err)
btProject, btInstance, btCleanup := tcc_testutils.SetupBigTable(t)
defer btCleanup()
tcc, err := task_cfg_cache.NewTaskCfgCache(ctx, repos, btProject, btInstance, nil)
require.NoError(t, err)
// The test repo is laid out like this:
// * T (HEAD, main, Case #12)
// * S (Time travel commit; before the start of the window)
// * R (Case #11)
// |\
// * | Q
// | * P
// |/
// * O (Case #9)
// * N
// * M (Case #10)
// * L
// * K (Case #6)
// * J (Case #5)
// |\
// | * I
// | * H (Case #4)
// * | G
// * | F (Case #3)
// * | E (Case #8, previously #7)
// |/
// * D (Case #2)
// * C (Case #1)
// ...
// * B (Case #0)
// * A
// * _ (No TasksCfg; blamelists shouldn't include this)
//
hashes := map[string]string{}
name := "Test-Ubuntu12-ShuttleA-GTX660-x86-Release"
taskCfg := &specs.TasksCfg{
Tasks: map[string]*specs.TaskSpec{
name: {},
},
}
commit := func(name string) {
hashes[name] = mg.Commit(name)
require.NoError(t, tcc.Set(ctx, types.RepoState{
Repo: rs1.Repo,
Revision: hashes[name],
}, taskCfg, nil))
}
// Initial commit.
hashes["_"] = mg.Commit("_")
// First commit containing TasksCfg.
commit("A")
type testCase struct {
Revision string
Expected []string
StoleFromIdx int
TaskName string
}
ids := []string{}
commitsBuf := make([]*repograph.Commit, 0, MAX_BLAMELIST_COMMITS)
test := func(tc *testCase) {
// Update the repo.
require.NoError(t, repo.Update(ctx))
// Self-check: make sure we don't pass in empty commit hashes.
for _, h := range tc.Expected {
require.NotEqual(t, h, "")
}
// Ensure that we get the expected blamelist.
revision := repo.Get(tc.Revision)
require.NotNil(t, revision)
taskName := tc.TaskName
if taskName == "" {
taskName = name
}
commits, stoleFrom, err := ComputeBlamelist(ctx, cache, repo, taskName, rs1.Repo, revision, commitsBuf, tcc, w)
if tc.Revision == "" {
require.Error(t, err)
return
} else {
require.NoError(t, err)
}
sort.Strings(commits)
sort.Strings(tc.Expected)
assertdeep.Equal(t, tc.Expected, commits)
if tc.StoleFromIdx >= 0 {
require.NotNil(t, stoleFrom)
require.Equal(t, ids[tc.StoleFromIdx], stoleFrom.Id)
} else {
require.Nil(t, stoleFrom)
}
// Insert the task into the DB.
c := &TaskCandidate{
TaskKey: types.TaskKey{
RepoState: types.RepoState{
Repo: rs1.Repo,
Revision: tc.Revision,
},
Name: taskName,
},
TaskSpec: &specs.TaskSpec{},
}
task := c.MakeTask()
task.Commits = commits
task.Created = now.Now(ctx)
if stoleFrom != nil {
// Re-insert the stoleFrom task without the commits
// which were stolen from it.
stoleFromCommits := make([]string, 0, len(stoleFrom.Commits)-len(commits))
for _, commit := range stoleFrom.Commits {
if !util.In(commit, task.Commits) {
stoleFromCommits = append(stoleFromCommits, commit)
}
}
stoleFrom.Commits = stoleFromCommits
require.NoError(t, d.PutTasks(ctx, []*types.Task{task, stoleFrom}))
cache.AddTasks([]*types.Task{task, stoleFrom})
} else {
require.NoError(t, d.PutTask(ctx, task))
cache.AddTasks([]*types.Task{task})
}
ids = append(ids, task.Id)
require.NoError(t, cache.Update(ctx))
}
// Commit B.
commit("B")
// Test cases. Each test case builds on the previous cases.
// 0. The first task, at HEAD.
test(&testCase{
Revision: hashes["B"],
Expected: []string{hashes["B"], hashes["A"]},
StoleFromIdx: -1,
})
// Test the blamelist too long case by creating a bunch of commits.
for i := 0; i < MAX_BLAMELIST_COMMITS+1; i++ {
commit("C")
}
commit("D")
// 1. Blamelist too long, not a branch head.
test(&testCase{
Revision: hashes["C"],
Expected: []string{hashes["C"]},
StoleFromIdx: -1,
})
// 2. Blamelist too long, is a branch head.
test(&testCase{
Revision: hashes["D"],
Expected: []string{hashes["D"]},
StoleFromIdx: -1,
})
// Create the remaining commits.
commit("E")
commit("F")
commit("G")
mg.NewBranch("otherbranch", hashes["D"])
mg.CheckoutBranch("otherbranch")
commit("H")
commit("I")
mg.CheckoutBranch(git.MainBranch)
hashes["J"] = mg.Merge("otherbranch")
require.NoError(t, tcc.Set(ctx, types.RepoState{
Repo: rs1.Repo,
Revision: hashes["J"],
}, taskCfg, nil))
commit("K")
// 3. On a linear set of commits, with at least one previous task.
test(&testCase{
Revision: hashes["F"],
Expected: []string{hashes["E"], hashes["F"]},
StoleFromIdx: -1,
})
// 4. The first task on a new branch.
test(&testCase{
Revision: hashes["H"],
Expected: []string{hashes["H"]},
StoleFromIdx: -1,
})
// 5. After a merge.
test(&testCase{
Revision: hashes["J"],
Expected: []string{hashes["G"], hashes["I"], hashes["J"]},
StoleFromIdx: -1,
})
// 6. One last "normal" task.
test(&testCase{
Revision: hashes["K"],
Expected: []string{hashes["K"]},
StoleFromIdx: -1,
})
// 7. Steal commits from a previously-ingested task.
test(&testCase{
Revision: hashes["E"],
Expected: []string{hashes["E"]},
StoleFromIdx: 3,
})
// Ensure that task #8 really stole the commit from #3.
task, err := cache.GetTask(ids[3])
require.NoError(t, err)
require.False(t, util.In(hashes["E"], task.Commits), fmt.Sprintf("Expected not to find %s in %v", hashes["E"], task.Commits))
// 8. Retry #7.
test(&testCase{
Revision: hashes["E"],
Expected: []string{hashes["E"]},
StoleFromIdx: 7,
})
// Ensure that task #8 really stole the commit from #7.
task, err = cache.GetTask(ids[7])
require.NoError(t, err)
require.Equal(t, 0, len(task.Commits))
// Four more commits.
commit("L")
commit("M")
commit("N")
commit("O")
// 9. Not really a test case, but setting up for #10.
test(&testCase{
Revision: hashes["O"],
Expected: []string{hashes["L"], hashes["M"], hashes["N"], hashes["O"]},
StoleFromIdx: -1,
})
// 10. Steal *two* commits from #9.
test(&testCase{
Revision: hashes["M"],
Expected: []string{hashes["L"], hashes["M"]},
StoleFromIdx: 9,
})
// 11. Verify that we correctly track when task specs were added.
mg.NewBranch("otherbranch2", hashes["O"])
commit("P")
mg.CheckoutBranch(git.MainBranch)
commit("Q")
newTaskCfg := taskCfg.Copy()
newTaskCfg.Tasks["added-task"] = &specs.TaskSpec{}
require.NoError(t, tcc.Set(ctx, types.RepoState{
Repo: rs1.Repo,
Revision: hashes["Q"],
}, newTaskCfg, nil))
hashes["R"] = mg.Merge("otherbranch2")
require.NoError(t, tcc.Set(ctx, types.RepoState{
Repo: rs1.Repo,
Revision: hashes["R"],
}, newTaskCfg, nil))
// Existing task should get a normal blamelist of all three new commits.
test(&testCase{
Revision: hashes["R"],
Expected: []string{hashes["P"], hashes["Q"], hashes["R"]},
StoleFromIdx: -1,
})
// The added task's blamelist should only include commits at which the
// task was defined, ie. not P.
test(&testCase{
Revision: hashes["R"],
Expected: []string{hashes["Q"], hashes["R"]},
StoleFromIdx: -1,
TaskName: "added-task",
})
// 12. Stop computing blamelists when we reach a commit outside of the
// scheduling window.
hashes["S"] = mg.CommitAt("S", w.EarliestStart().Add(-time.Hour))
require.NoError(t, tcc.Set(ctx, types.RepoState{
Repo: rs1.Repo,
Revision: hashes["S"],
}, taskCfg, nil))
commit("T")
test(&testCase{
Revision: hashes["T"],
Expected: []string{hashes["T"]},
StoleFromIdx: -1,
})
}
func TestTimeDecay24Hr(t *testing.T) {
tc := []struct {
decayAmt24Hr float64
elapsed time.Duration
out float64
}{
{
decayAmt24Hr: 1.0,
elapsed: 10 * time.Hour,
out: 1.0,
},
{
decayAmt24Hr: 0.5,
elapsed: 0 * time.Hour,
out: 1.0,
},
{
decayAmt24Hr: 0.5,
elapsed: 24 * time.Hour,
out: 0.5,
},
{
decayAmt24Hr: 0.5,
elapsed: 12 * time.Hour,
out: 0.75,
},
{
decayAmt24Hr: 0.5,
elapsed: 36 * time.Hour,
out: 0.25,
},
{
decayAmt24Hr: 0.5,
elapsed: 48 * time.Hour,
out: 0.0,
},
{
decayAmt24Hr: 0.5,
elapsed: 72 * time.Hour,
out: 0.0,
},
}
for i, c := range tc {
require.Equal(t, c.out, timeDecay24Hr(c.decayAmt24Hr, c.elapsed), fmt.Sprintf("test case #%d", i))
}
}
func TestRegenerateTaskQueue(t *testing.T) {
ctx, _, _, _, s, _, _, cleanup := setup(t)
defer cleanup()
// Ensure that the queue is initially empty.
require.Equal(t, 0, len(s.queue))
c1 := rs1.Revision
c2 := rs2.Revision
// Regenerate the task queue.
queue, _, err := s.regenerateTaskQueue(ctx)
require.NoError(t, err)
require.Len(t, queue, 2) // Two Build tasks.
testSort := func() {
// Ensure that we sorted correctly.
if len(queue) == 0 {
return
}
highScore := queue[0].Score
for _, c := range queue {
require.True(t, highScore >= c.Score)
highScore = c.Score
}
}
testSort()
// Since we haven't run any task yet, we should have the two Build
// tasks.
// The one at HEAD should have a two-commit blamelist and a
// score of 3.5, scaled by a priority of 0.875 due to three jobs
// depending on it (1 - 0.5^3).
require.Equal(t, tcc_testutils.BuildTaskName, queue[0].Name)
require.Equal(t, []string{c2, c1}, queue[0].Commits)
require.Equal(t, 3, len(queue[0].Jobs))
require.InDelta(t, 3.5*0.875, queue[0].Score, scoreDelta)
// The other should have one commit in its blamelist and
// a score of 0.5, scaled by a priority of 0.75 due to two jobs.
require.Equal(t, tcc_testutils.BuildTaskName, queue[1].Name)
require.Equal(t, []string{c1}, queue[1].Commits)
require.InDelta(t, 0.5*0.75, queue[1].Score, scoreDelta)
// Insert the task at c1, even though it scored lower.
t1 := makeTask(ctx, queue[1].Name, queue[1].Repo, queue[1].Revision)
require.NotNil(t, t1)
t1.Status = types.TASK_STATUS_SUCCESS
t1.IsolatedOutput = "fake isolated hash"
require.NoError(t, s.putTask(ctx, t1))
// Regenerate the task queue.
queue, _, err = s.regenerateTaskQueue(ctx)
require.NoError(t, err)
// Now we expect the queue to contain the other Build task and the one
// Test task we unblocked by running the first Build task.
require.Len(t, queue, 2)
testSort()
for _, c := range queue {
if c.Name == tcc_testutils.TestTaskName {
require.InDelta(t, 2.0*0.5, c.Score, scoreDelta)
require.Equal(t, 1, len(c.Commits))
} else {
require.Equal(t, c.Name, tcc_testutils.BuildTaskName)
require.InDelta(t, 2.0*0.875, c.Score, scoreDelta)
require.Equal(t, []string{c.Revision}, c.Commits)
}
}
buildIdx := 0
testIdx := 1
if queue[1].Name == tcc_testutils.BuildTaskName {
buildIdx = 1
testIdx = 0
}
require.Equal(t, tcc_testutils.BuildTaskName, queue[buildIdx].Name)
require.Equal(t, c2, queue[buildIdx].Revision)
require.Equal(t, tcc_testutils.TestTaskName, queue[testIdx].Name)
require.Equal(t, c1, queue[testIdx].Revision)
// Run the other Build task.
t2 := makeTask(ctx, queue[buildIdx].Name, queue[buildIdx].Repo, queue[buildIdx].Revision)
t2.Status = types.TASK_STATUS_SUCCESS
t2.IsolatedOutput = "fake isolated hash"
require.NoError(t, s.putTask(ctx, t2))
// Regenerate the task queue.
queue, _, err = s.regenerateTaskQueue(ctx)
require.NoError(t, err)
require.Len(t, queue, 3)
testSort()
perfIdx := -1
for i, c := range queue {
if c.Name == tcc_testutils.PerfTaskName {
perfIdx = i
require.Equal(t, c2, c.Revision)
require.InDelta(t, 2.0*0.5, c.Score, scoreDelta)
require.Equal(t, []string{c.Revision}, c.Commits)
} else {
require.Equal(t, c.Name, tcc_testutils.TestTaskName)
if c.Revision == c2 {
require.InDelta(t, 3.5*0.5, c.Score, scoreDelta)
require.Equal(t, []string{c2, c1}, c.Commits)
} else {
require.InDelta(t, 0.5*0.5, c.Score, scoreDelta)
require.Equal(t, []string{c.Revision}, c.Commits)
}
}
}
require.True(t, perfIdx > -1)
// Run the Test task at tip of tree; its blamelist covers both commits.
t3 := makeTask(ctx, tcc_testutils.TestTaskName, rs1.Repo, c2)
t3.Commits = []string{c2, c1}
t3.Status = types.TASK_STATUS_SUCCESS
t3.IsolatedOutput = "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff/256"
require.NoError(t, s.putTask(ctx, t3))
// Regenerate the task queue.
queue, _, err = s.regenerateTaskQueue(ctx)
require.NoError(t, err)
// Now we expect the queue to contain one Test and one Perf task. The
// Test task is a backfill, and should have a score of 0.5, scaled by
// the priority of 0.5.
require.Len(t, queue, 2)
testSort()
// First candidate should be the perf task.
require.Equal(t, tcc_testutils.PerfTaskName, queue[0].Name)
require.InDelta(t, 2.0*0.5, queue[0].Score, scoreDelta)
// The test task is next, a backfill.
require.Equal(t, tcc_testutils.TestTaskName, queue[1].Name)
require.InDelta(t, 0.5*0.5, queue[1].Score, scoreDelta)
}
func makeTaskCandidate(name string, dims []string) *TaskCandidate {
return &TaskCandidate{
Score: 1.0,
TaskKey: types.TaskKey{
Name: name,
},
TaskSpec: &specs.TaskSpec{
Dimensions: dims,
},
}
}
func makeSwarmingBot(id string, dims []string) *types.Machine {
return &types.Machine{
ID: id,
Dimensions: dims,
}
}
func TestGetCandidatesToSchedule(t *testing.T) {
ctx := context.Background()
// Empty lists.
rv := getCandidatesToSchedule(ctx, []*types.Machine{}, []*TaskCandidate{})
require.Empty(t, rv)
// checkDiags takes a list of bots with the same dimensions and a list of
// ordered candidates that match those bots and checks the Diagnostics for
// candidates.
checkDiags := func(bots []*types.Machine, candidates []*TaskCandidate) {
var expectedBots []string
if len(bots) > 0 {
expectedBots = make([]string, len(bots), len(bots))
for i, b := range bots {
expectedBots[i] = b.ID
}
}
for i, c := range candidates {
// These conditions are not tested.
require.False(t, c.Diagnostics.Scheduling.OverSchedulingLimitPerTaskSpec)
require.False(t, c.Diagnostics.Scheduling.ScoreBelowThreshold)
// NoBotsAvailable and MatchingBots will be the same for all candidates.
require.Equal(t, len(expectedBots) == 0, c.Diagnostics.Scheduling.NoBotsAvailable)
require.Equal(t, expectedBots, c.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, i, c.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
if i == 0 {
// First candidate.
require.Nil(t, c.Diagnostics.Scheduling.LastSimilarCandidate)
} else {
last := candidates[i-1]
require.Equal(t, &last.TaskKey, c.Diagnostics.Scheduling.LastSimilarCandidate)
}
require.Equal(t, i < len(bots), c.Diagnostics.Scheduling.Selected)
}
// Clear diagnostics for next test.
for _, c := range candidates {
c.Diagnostics = nil
}
}
t1 := makeTaskCandidate("task1", []string{"k:v"})
rv = getCandidatesToSchedule(ctx, []*types.Machine{}, []*TaskCandidate{t1})
require.Empty(t, rv)
checkDiags([]*types.Machine{}, []*TaskCandidate{t1})
b1 := makeSwarmingBot("bot1", []string{"k:v"})
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1}, []*TaskCandidate{})
require.Empty(t, rv)
// Single match.
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1}, []*TaskCandidate{t1})
assertdeep.Equal(t, []*TaskCandidate{t1}, rv)
checkDiags([]*types.Machine{b1}, []*TaskCandidate{t1})
// No match.
t1.TaskSpec.Dimensions[0] = "k:v2"
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1}, []*TaskCandidate{t1})
require.Empty(t, rv)
checkDiags([]*types.Machine{}, []*TaskCandidate{t1})
// Add a task candidate to match b1.
t1 = makeTaskCandidate("task1", []string{"k:v2"})
t2 := makeTaskCandidate("task2", []string{"k:v"})
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1}, []*TaskCandidate{t1, t2})
assertdeep.Equal(t, []*TaskCandidate{t2}, rv)
checkDiags([]*types.Machine{}, []*TaskCandidate{t1})
checkDiags([]*types.Machine{b1}, []*TaskCandidate{t2})
// Switch the task order.
t1 = makeTaskCandidate("task1", []string{"k:v2"})
t2 = makeTaskCandidate("task2", []string{"k:v"})
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1}, []*TaskCandidate{t2, t1})
assertdeep.Equal(t, []*TaskCandidate{t2}, rv)
checkDiags([]*types.Machine{}, []*TaskCandidate{t1})
checkDiags([]*types.Machine{b1}, []*TaskCandidate{t2})
// Make both tasks match the bot, ensure that we pick the first one.
t1 = makeTaskCandidate("task1", []string{"k:v"})
t2 = makeTaskCandidate("task2", []string{"k:v"})
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1}, []*TaskCandidate{t1, t2})
assertdeep.Equal(t, []*TaskCandidate{t1}, rv)
checkDiags([]*types.Machine{b1}, []*TaskCandidate{t1, t2})
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1}, []*TaskCandidate{t2, t1})
assertdeep.Equal(t, []*TaskCandidate{t2}, rv)
checkDiags([]*types.Machine{b1}, []*TaskCandidate{t2, t1})
// Multiple dimensions. Ensure that different permutations of the bots
// and tasks lists give us the expected results.
dims := []string{"k:v", "k2:v2", "k3:v3"}
b1 = makeSwarmingBot("bot1", dims)
b2 := makeSwarmingBot("bot2", t1.TaskSpec.Dimensions)
t1 = makeTaskCandidate("task1", []string{"k:v"})
t2 = makeTaskCandidate("task2", dims)
// In the first two cases, the task with fewer dimensions has the
// higher priority. It gets the bot with more dimensions because it
// is first in sorted order. The second task does not get scheduled
// because there is no bot available which can run it.
// TODO(borenet): Use a more optimal solution to avoid this case.
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1, b2}, []*TaskCandidate{t1, t2})
assertdeep.Equal(t, []*TaskCandidate{t1}, rv)
// Can't use checkDiags for these cases.
require.Equal(t, []string{b1.ID, b2.ID}, t1.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, 0, t1.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
require.Nil(t, t1.Diagnostics.Scheduling.LastSimilarCandidate)
require.True(t, t1.Diagnostics.Scheduling.Selected)
t1.Diagnostics = nil
require.Equal(t, []string{b1.ID}, t2.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, 1, t2.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
require.Equal(t, &t1.TaskKey, t2.Diagnostics.Scheduling.LastSimilarCandidate)
require.False(t, t2.Diagnostics.Scheduling.Selected)
t2.Diagnostics = nil
t1 = makeTaskCandidate("task1", []string{"k:v"})
t2 = makeTaskCandidate("task2", dims)
rv = getCandidatesToSchedule(ctx, []*types.Machine{b2, b1}, []*TaskCandidate{t1, t2})
assertdeep.Equal(t, []*TaskCandidate{t1}, rv)
require.Equal(t, []string{b1.ID, b2.ID}, t1.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, 0, t1.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
require.Nil(t, t1.Diagnostics.Scheduling.LastSimilarCandidate)
require.True(t, t1.Diagnostics.Scheduling.Selected)
t1.Diagnostics = nil
require.Equal(t, []string{b1.ID}, t2.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, 1, t2.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
require.Equal(t, &t1.TaskKey, t2.Diagnostics.Scheduling.LastSimilarCandidate)
require.False(t, t2.Diagnostics.Scheduling.Selected)
t2.Diagnostics = nil
// In these two cases, the task with more dimensions has the higher
// priority. Both tasks get scheduled.
t1 = makeTaskCandidate("task1", []string{"k:v"})
t2 = makeTaskCandidate("task2", dims)
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1, b2}, []*TaskCandidate{t2, t1})
assertdeep.Equal(t, []*TaskCandidate{t2, t1}, rv)
require.Equal(t, []string{b1.ID, b2.ID}, t1.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, 1, t1.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
require.Equal(t, &t2.TaskKey, t1.Diagnostics.Scheduling.LastSimilarCandidate)
require.True(t, t1.Diagnostics.Scheduling.Selected)
t1.Diagnostics = nil
require.Equal(t, []string{b1.ID}, t2.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, 0, t2.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
require.Nil(t, t2.Diagnostics.Scheduling.LastSimilarCandidate)
require.True(t, t2.Diagnostics.Scheduling.Selected)
t2.Diagnostics = nil
t1 = makeTaskCandidate("task1", []string{"k:v"})
t2 = makeTaskCandidate("task2", dims)
rv = getCandidatesToSchedule(ctx, []*types.Machine{b2, b1}, []*TaskCandidate{t2, t1})
assertdeep.Equal(t, []*TaskCandidate{t2, t1}, rv)
require.Equal(t, []string{b1.ID, b2.ID}, t1.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, 1, t1.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
require.Equal(t, &t2.TaskKey, t1.Diagnostics.Scheduling.LastSimilarCandidate)
require.True(t, t1.Diagnostics.Scheduling.Selected)
t1.Diagnostics = nil
require.Equal(t, []string{b1.ID}, t2.Diagnostics.Scheduling.MatchingBots)
require.Equal(t, 0, t2.Diagnostics.Scheduling.NumHigherScoreSimilarCandidates)
require.Nil(t, t2.Diagnostics.Scheduling.LastSimilarCandidate)
require.True(t, t2.Diagnostics.Scheduling.Selected)
t2.Diagnostics = nil
// Matching dimensions. More bots than tasks.
b2 = makeSwarmingBot("bot2", dims)
b3 := makeSwarmingBot("bot3", dims)
t1 = makeTaskCandidate("task1", dims)
t2 = makeTaskCandidate("task2", dims)
t3 := makeTaskCandidate("task3", dims)
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1, b2, b3}, []*TaskCandidate{t1, t2})
assertdeep.Equal(t, []*TaskCandidate{t1, t2}, rv)
checkDiags([]*types.Machine{b1, b2, b3}, []*TaskCandidate{t1, t2})
// More tasks than bots.
t1 = makeTaskCandidate("task1", dims)
t2 = makeTaskCandidate("task2", dims)
t3 = makeTaskCandidate("task3", dims)
rv = getCandidatesToSchedule(ctx, []*types.Machine{b1, b2}, []*TaskCandidate{t1, t2, t3})
assertdeep.Equal(t, []*TaskCandidate{t1, t2}, rv)
checkDiags([]*types.Machine{b1, b2}, []*TaskCandidate{t1, t2, t3})
}
func makeBot(id string, dims map[string]string) *types.Machine {
dimensions := make([]string, 0, len(dims))
for k, v := range dims {
dimensions = append(dimensions, fmt.Sprintf("%s:%s", k, v))
}
return &types.Machine{
ID: id,
Dimensions: dimensions,
}
}
func mockBots(t sktest.TestingT, swarmingClient *swarming_testutils.TestClient, bots ...*types.Machine) {
swarmBots := make([]*swarming_api.SwarmingRpcsBotInfo, 0, len(bots))
for _, bot := range bots {
dims, err := swarming.ParseDimensions(bot.Dimensions)
require.NoError(t, err)
swarmBots = append(swarmBots, &swarming_api.SwarmingRpcsBotInfo{
BotId: bot.ID,
Dimensions: swarming.StringMapToBotDimensions(dims),
})
}
swarmingClient.MockBots(swarmBots)
}
func TestSchedulingE2E(t *testing.T) {
ctx, _, _, swarmingClient, s, _, cas, cleanup := setup(t)
defer cleanup()
c1 := rs1.Revision
c2 := rs2.Revision
// Start testing. No free bots, so we get a full queue with nothing
// scheduled.
runMainLoop(t, s, ctx)
tasks, err := s.tCache.GetTasksForCommits(rs1.Repo, []string{c1, c2})
require.NoError(t, err)
expect := map[string]map[string]*types.Task{
c1: {},
c2: {},
}
assertdeep.Equal(t, expect, tasks)
require.Equal(t, 2, len(s.queue)) // Two compile tasks.
// A bot is free but doesn't have all of the right dimensions to run a task.
bot1 := makeBot("bot1", map[string]string{"pool": "Skia"})
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, []string{c1, c2})
require.NoError(t, err)
expect = map[string]map[string]*types.Task{
c1: {},
c2: {},
}
assertdeep.Equal(t, expect, tasks)
require.Equal(t, 2, len(s.queue)) // Still two compile tasks.
// One bot free, schedule a task, ensure it's not in the queue.
bot1.Dimensions = append(bot1.Dimensions, "os:Ubuntu")
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, []string{c1, c2})
require.NoError(t, err)
t1 := tasks[c2][tcc_testutils.BuildTaskName]
require.NotNil(t, t1)
require.Equal(t, c2, t1.Revision)
require.Equal(t, tcc_testutils.BuildTaskName, t1.Name)
require.Equal(t, []string{c2, c1}, t1.Commits)
require.Equal(t, 1, len(s.queue))
// The task is complete.
t1.Status = types.TASK_STATUS_SUCCESS
t1.Finished = now.Now(ctx)
t1.IsolatedOutput = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd/86"
require.NoError(t, s.putTask(ctx, t1))
swarmingClient.MockTasks([]*swarming_api.SwarmingRpcsTaskRequestMetadata{
makeSwarmingRpcsTaskRequestMetadata(t, t1, linuxTaskDims),
})
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.TestCASDigest, t1.IsolatedOutput}).Return("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbabc123/56", nil)
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.PerfCASDigest, t1.IsolatedOutput}).Return("ccccccccccccccccccccccccccccccccccccccccccccccccccccccccccabc123/56", nil)
// No bots free. Ensure that the queue is correct.
mockBots(t, swarmingClient)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, []string{c1, c2})
require.NoError(t, err)
for _, c := range t1.Commits {
expect[c][t1.Name] = t1
}
assertdeep.Equal(t, expect, tasks)
expectLen := 3 // One remaining build task, plus one test task and one perf task.
require.Equal(t, expectLen, len(s.queue))
// More bots than tasks free, ensure the queue is correct.
bot2 := makeBot("bot2", androidTaskDims)
bot3 := makeBot("bot3", androidTaskDims)
bot4 := makeBot("bot4", linuxTaskDims)
mockBots(t, swarmingClient, bot1, bot2, bot3, bot4)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
_, err = s.tCache.GetTasksForCommits(rs1.Repo, []string{c1, c2})
require.NoError(t, err)
require.Equal(t, 0, len(s.queue))
// The build, test, and perf tasks should have triggered.
var t2 *types.Task
var t3 *types.Task
var t4 *types.Task
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, []string{c1, c2})
require.NoError(t, err)
require.Len(t, tasks, 2)
for commit, v := range tasks {
if commit == c1 {
// Build task at c1 and test task at c2 whose blamelist also has c1.
require.Len(t, v, 2)
for _, task := range v {
if task.Revision != commit {
continue
}
require.Equal(t, tcc_testutils.BuildTaskName, task.Name)
require.Nil(t, t4)
t4 = task
require.Equal(t, c1, task.Revision)
require.Equal(t, []string{c1}, task.Commits)
}
} else {
require.Len(t, v, 3)
for _, task := range v {
if task.Name == tcc_testutils.TestTaskName {
require.Nil(t, t2)
t2 = task
require.Equal(t, c2, task.Revision)
require.Equal(t, []string{c2, c1}, task.Commits)
} else if task.Name == tcc_testutils.PerfTaskName {
require.Nil(t, t3)
t3 = task
require.Equal(t, c2, task.Revision)
require.Equal(t, []string{c2}, task.Commits)
} else {
// This is the first task we triggered.
require.Equal(t, tcc_testutils.BuildTaskName, task.Name)
}
}
}
}
require.NotNil(t, t2)
require.NotNil(t, t3)
require.NotNil(t, t4)
t4.Status = types.TASK_STATUS_SUCCESS
t4.Finished = now.Now(ctx)
t4.IsolatedOutput = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee/99"
require.NoError(t, s.putTask(ctx, t4))
// No new bots free; only the remaining test task should be in the queue.
mockBots(t, swarmingClient)
mockTasks := []*swarming_api.SwarmingRpcsTaskRequestMetadata{
makeSwarmingRpcsTaskRequestMetadata(t, t2, linuxTaskDims),
makeSwarmingRpcsTaskRequestMetadata(t, t3, linuxTaskDims),
makeSwarmingRpcsTaskRequestMetadata(t, t4, linuxTaskDims),
}
swarmingClient.MockTasks(mockTasks)
require.NoError(t, s.updateUnfinishedTasks(ctx))
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
expectLen = 1 // Test task from c1
require.Equal(t, expectLen, len(s.queue))
// Finish the other task.
t3, err = s.tCache.GetTask(t3.Id)
require.NoError(t, err)
t3.Status = types.TASK_STATUS_SUCCESS
t3.Finished = now.Now(ctx)
t3.IsolatedOutput = "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff/256"
// Ensure that we finalize all of the tasks and insert into the DB.
mockBots(t, swarmingClient, bot1, bot2, bot3, bot4)
mockTasks = []*swarming_api.SwarmingRpcsTaskRequestMetadata{
makeSwarmingRpcsTaskRequestMetadata(t, t3, linuxTaskDims),
}
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.TestCASDigest, t4.IsolatedOutput}).Return("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbabc123/56", nil)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, []string{c1, c2})
require.NoError(t, err)
require.Equal(t, 2, len(tasks[c1]))
require.Equal(t, 3, len(tasks[c2]))
require.Equal(t, 0, len(s.queue))
// Mark everything as finished. Ensure that the queue still ends up empty.
tasksList := []*types.Task{}
for _, v := range tasks {
for _, task := range v {
if task.Status != types.TASK_STATUS_SUCCESS {
task.Status = types.TASK_STATUS_SUCCESS
task.Finished = now.Now(ctx)
task.IsolatedOutput = "abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234/56"
tasksList = append(tasksList, task)
}
}
}
mockTasks = make([]*swarming_api.SwarmingRpcsTaskRequestMetadata, 0, len(tasksList))
for _, task := range tasksList {
mockTasks = append(mockTasks, makeSwarmingRpcsTaskRequestMetadata(t, task, linuxTaskDims))
}
swarmingClient.MockTasks(mockTasks)
mockBots(t, swarmingClient, bot1, bot2, bot3, bot4)
require.NoError(t, s.updateUnfinishedTasks(ctx))
runMainLoop(t, s, ctx)
require.Equal(t, 0, len(s.queue))
}
func TestSchedulerStealingFrom(t *testing.T) {
ctx, mg, _, swarmingClient, s, _, _, cleanup := setup(t)
defer cleanup()
c1 := rs1.Revision
c2 := rs2.Revision
// Run the available compile task at c2.
bot1 := makeBot("bot1", linuxTaskDims)
bot2 := makeBot("bot2", linuxTaskDims)
mockBots(t, swarmingClient, bot1, bot2)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err := s.tCache.GetTasksForCommits(rs1.Repo, []string{c1, c2})
require.NoError(t, err)
require.Equal(t, 1, len(tasks[c1]))
require.Equal(t, 1, len(tasks[c2]))
// Finish the one task.
tasksList := []*types.Task{}
t1 := tasks[c2][tcc_testutils.BuildTaskName]
t1.Status = types.TASK_STATUS_SUCCESS
t1.Finished = now.Now(ctx)
t1.IsolatedOutput = "abc123"
tasksList = append(tasksList, t1)
// Forcibly create and insert a second task at c1.
t2 := t1.Copy()
t2.Id = "t2id"
t2.Revision = c1
t2.Commits = []string{c1}
tasksList = append(tasksList, t2)
require.NoError(t, s.putTasks(ctx, tasksList))
// Add some commits.
commits := mg.CommitN(10)
require.NoError(t, s.repos[rs1.Repo].Update(ctx))
for _, h := range commits {
rs := types.RepoState{
Repo: rs1.Repo,
Revision: h,
}
fillCaches(t, ctx, s.taskCfgCache, rs, tcc_testutils.TasksCfg2)
insertJobs(t, ctx, s, rs)
}
// Run one task. Ensure that it's at tip-of-tree.
head := s.repos[rs1.Repo].Get(git.MainBranch).Hash
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, commits)
require.NoError(t, err)
require.Len(t, tasks[head], 1)
task := tasks[head][tcc_testutils.BuildTaskName]
require.Equal(t, head, task.Revision)
expect := commits[:]
sort.Strings(expect)
sort.Strings(task.Commits)
assertdeep.Equal(t, expect, task.Commits)
task.Status = types.TASK_STATUS_SUCCESS
task.Finished = now.Now(ctx)
task.IsolatedOutput = "abc123"
require.NoError(t, s.putTask(ctx, task))
oldTasksByCommit := tasks
// Run backfills, ensuring that each one steals the right set of commits
// from previous builds, until all of the build task candidates have run.
for i := 0; i < 9; i++ {
// Now, run another task. The new task should bisect the old one.
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, commits)
require.NoError(t, err)
var newTask *types.Task
for _, v := range tasks {
for _, task := range v {
if task.Status == types.TASK_STATUS_PENDING {
require.True(t, newTask == nil || task.Id == newTask.Id)
newTask = task
}
}
}
require.NotNil(t, newTask)
oldTask := oldTasksByCommit[newTask.Revision][newTask.Name]
require.NotNil(t, oldTask)
require.True(t, util.In(newTask.Revision, oldTask.Commits))
// Find the updated old task.
updatedOldTask, err := s.tCache.GetTask(oldTask.Id)
require.NoError(t, err)
require.NotNil(t, updatedOldTask)
// Ensure that the blamelists are correct.
old := util.NewStringSet(oldTask.Commits)
new := util.NewStringSet(newTask.Commits)
updatedOld := util.NewStringSet(updatedOldTask.Commits)
assertdeep.Equal(t, old, new.Union(updatedOld))
require.Equal(t, 0, len(new.Intersect(updatedOld)))
// Finish the new task.
newTask.Status = types.TASK_STATUS_SUCCESS
newTask.Finished = now.Now(ctx)
newTask.IsolatedOutput = "abc123"
require.NoError(t, s.putTask(ctx, newTask))
oldTasksByCommit = tasks
}
// Ensure that we're really done.
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, commits)
require.NoError(t, err)
var newTask *types.Task
for _, v := range tasks {
for _, task := range v {
if task.Status == types.TASK_STATUS_PENDING {
require.True(t, newTask == nil || task.Id == newTask.Id)
newTask = task
}
}
}
require.Nil(t, newTask)
}
// spyDB calls onPutTasks before delegating PutTask(s) to DB.
type spyDB struct {
db.DB
onPutTasks func([]*types.Task)
}
func (s *spyDB) PutTask(ctx context.Context, task *types.Task) error {
s.onPutTasks([]*types.Task{task})
return s.DB.PutTask(ctx, task)
}
func (s *spyDB) PutTasks(ctx context.Context, tasks []*types.Task) error {
s.onPutTasks(tasks)
return s.DB.PutTasks(ctx, tasks)
}
func testMultipleCandidatesBackfillingEachOtherSetup(t *testing.T) (context.Context, *mem_git.MemGit, db.DB, *TaskScheduler, *swarming_testutils.TestClient, []string, func(*types.Task), *specs.TasksCfg, func()) {
ctx, cancel := context.WithCancel(context.Background())
workdir, err := os.MkdirTemp("", "")
require.NoError(t, err)
// Setup the scheduler.
d := memory.NewInMemoryDB()
swarmingClient := swarming_testutils.NewTestClient()
mg, repo := newMemRepo(t)
repos := repograph.Map{
rs1.Repo: repo,
}
btProject, btInstance, btCleanup := tcc_testutils.SetupBigTable(t)
taskCfgCache, err := task_cfg_cache.NewTaskCfgCache(ctx, repos, btProject, btInstance, nil)
require.NoError(t, err)
// Create a single task in the config.
taskName := "faketask"
cfg := &specs.TasksCfg{
CasSpecs: map[string]*specs.CasSpec{
"compile": {
Digest: tcc_testutils.CompileCASDigest,
},
},
Tasks: map[string]*specs.TaskSpec{
taskName: {
CasSpec: "compile",
CipdPackages: []*specs.CipdPackage{},
Dependencies: []string{},
Dimensions: []string{"pool:Skia"},
Priority: 1.0,
},
},
Jobs: map[string]*specs.JobSpec{
"j1": {
TaskSpecs: []string{taskName},
},
},
}
hashes := mg.CommitN(1)
mockTasks := []*swarming_api.SwarmingRpcsTaskRequestMetadata{}
mock := func(task *types.Task) {
task.Status = types.TASK_STATUS_SUCCESS
task.Finished = now.Now(ctx)
task.IsolatedOutput = tcc_testutils.CompileCASDigest
mockTasks = append(mockTasks, makeSwarmingRpcsTaskRequestMetadata(t, task, linuxTaskDims))
swarmingClient.MockTasks(mockTasks)
}
// Create the TaskScheduler.
cas := &mocks.CAS{}
cas.On("Close").Return(nil)
// Go ahead and mock the single-input merge calls.
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.CompileCASDigest}).Return(tcc_testutils.CompileCASDigest, nil)
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.TestCASDigest}).Return(tcc_testutils.TestCASDigest, nil)
cas.On("Merge", testutils.AnyContext, []string{tcc_testutils.PerfCASDigest}).Return(tcc_testutils.PerfCASDigest, nil)
taskExec := swarming_task_execution.NewSwarmingTaskExecutor(swarmingClient, "fake-cas-instance", "")
taskExecs := map[string]types.TaskExecutor{
types.TaskExecutor_Swarming: taskExec,
types.TaskExecutor_UseDefault: taskExec,
}
s, err := NewTaskScheduler(ctx, d, nil, time.Duration(math.MaxInt64), 0, repos, cas, "fake-cas-instance", taskExecs, mockhttpclient.NewURLMock().Client(), 1.0, swarming.POOLS_PUBLIC, "", taskCfgCache, nil, mem_gcsclient.New("diag_unit_tests"), btInstance, BusyBotsDebugLoggingOff)
require.NoError(t, err)
for _, h := range hashes {
rs := types.RepoState{
Repo: rs1.Repo,
Revision: h,
}
fillCaches(t, ctx, taskCfgCache, rs, cfg)
insertJobs(t, ctx, s, rs)
}
// Cycle once.
bot1 := makeBot("bot1", map[string]string{"pool": "Skia"})
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
require.Equal(t, 0, len(s.queue))
head := s.repos[rs1.Repo].Get(git.MainBranch).Hash
tasks, err := s.tCache.GetTasksForCommits(rs1.Repo, []string{head})
require.NoError(t, err)
require.Len(t, tasks[head], 1)
mock(tasks[head][taskName])
// Add some commits to the repo.
newHashes := mg.CommitN(8)
for _, h := range newHashes {
rs := types.RepoState{
Repo: rs1.Repo,
Revision: h,
}
fillCaches(t, ctx, taskCfgCache, rs, cfg)
insertJobs(t, ctx, s, rs)
}
require.NoError(t, s.repos[rs1.Repo].Update(ctx))
return ctx, mg, d, s, swarmingClient, newHashes, mock, cfg, func() {
testutils.AssertCloses(t, s)
testutils.RemoveAll(t, workdir)
btCleanup()
cancel()
}
}
func TestMultipleCandidatesBackfillingEachOther(t *testing.T) {
ctx, _, d, s, swarmingClient, commits, mock, _, cleanup := testMultipleCandidatesBackfillingEachOtherSetup(t)
defer cleanup()
// Trigger builds simultaneously.
bot1 := makeBot("bot1", map[string]string{"pool": "Skia"})
bot2 := makeBot("bot2", map[string]string{"pool": "Skia"})
bot3 := makeBot("bot3", map[string]string{"pool": "Skia"})
mockBots(t, swarmingClient, bot1, bot2, bot3)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
require.Equal(t, 5, len(s.queue))
tasks, err := s.tCache.GetTasksForCommits(rs1.Repo, commits)
require.NoError(t, err)
// If we're queueing correctly, we should've triggered tasks at
// commits[0], commits[4], and either commits[2] or commits[6].
var t1, t2, t3 *types.Task
for _, byName := range tasks {
for _, task := range byName {
if task.Revision == commits[0] {
t1 = task
} else if task.Revision == commits[4] {
t2 = task
} else if task.Revision == commits[2] || task.Revision == commits[6] {
t3 = task
} else {
require.FailNow(t, fmt.Sprintf("Task has unknown revision: %+v", task))
}
}
}
require.NotNil(t, t1)
require.NotNil(t, t2)
require.NotNil(t, t3)
mock(t1)
mock(t2)
mock(t3)
// Ensure that we got the blamelists right.
var expect1, expect2, expect3 []string
if t3.Revision == commits[2] {
expect1 = util.CopyStringSlice(commits[:2])
expect2 = util.CopyStringSlice(commits[4:])
expect3 = util.CopyStringSlice(commits[2:4])
} else {
expect1 = util.CopyStringSlice(commits[:4])
expect2 = util.CopyStringSlice(commits[4:6])
expect3 = util.CopyStringSlice(commits[6:])
}
sort.Strings(expect1)
sort.Strings(expect2)
sort.Strings(expect3)
sort.Strings(t1.Commits)
sort.Strings(t2.Commits)
sort.Strings(t3.Commits)
assertdeep.Equal(t, expect1, t1.Commits)
assertdeep.Equal(t, expect2, t2.Commits)
assertdeep.Equal(t, expect3, t3.Commits)
// Just for good measure, check the task at the head of the queue.
expectIdx := 2
if t3.Revision == commits[expectIdx] {
expectIdx = 6
}
require.Equal(t, commits[expectIdx], s.queue[0].Revision)
retryCount := 0
causeConcurrentUpdate := func(tasks []*types.Task) {
// HACK(benjaminwagner): Filter out PutTask calls from
// updateUnfinishedTasks by looking for new tasks.
anyNew := false
for _, task := range tasks {
if util.TimeIsZero(task.DbModified) {
anyNew = true
break
}
}
if !anyNew {
return
}
if retryCount < 3 {
taskToUpdate := []*types.Task{t1, t2, t3}[retryCount]
retryCount++
taskInDb, err := d.GetTaskById(ctx, taskToUpdate.Id)
require.NoError(t, err)
taskInDb.Status = types.TASK_STATUS_SUCCESS
require.NoError(t, d.PutTask(ctx, taskInDb))
s.tCache.AddTasks([]*types.Task{taskInDb})
}
}
s.db = &spyDB{
DB: d,
onPutTasks: causeConcurrentUpdate,
}
// Run again with 5 bots to check the case where we bisect the same
// task twice.
bot4 := makeBot("bot4", map[string]string{"pool": "Skia"})
bot5 := makeBot("bot5", map[string]string{"pool": "Skia"})
mockBots(t, swarmingClient, bot1, bot2, bot3, bot4, bot5)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
require.Equal(t, 0, len(s.queue))
require.Equal(t, 3, retryCount)
tasks, err = s.tCache.GetTasksForCommits(rs1.Repo, commits)
require.NoError(t, err)
for _, byName := range tasks {
for _, task := range byName {
require.Equal(t, 1, len(task.Commits))
require.Equal(t, task.Revision, task.Commits[0])
if util.In(task.Id, []string{t1.Id, t2.Id, t3.Id}) {
require.Equal(t, types.TASK_STATUS_SUCCESS, task.Status)
} else {
require.Equal(t, types.TASK_STATUS_PENDING, task.Status)
}
}
}
}
func TestSchedulingRetry(t *testing.T) {
ctx, _, _, swarmingClient, s, _, _, cleanup := setup(t)
defer cleanup()
c1 := rs1.Revision
// Run the available compile task at c2.
bot1 := makeBot("bot1", linuxTaskDims)
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err := s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, tasks, 1)
t1 := tasks[0]
require.NotNil(t, t1)
// Ensure c2, not c1.
require.NotEqual(t, c1, t1.Revision)
c2 := t1.Revision
// Forcibly add a second build task at c1.
t2 := t1.Copy()
t2.Id = "t2Id"
t2.Revision = c1
t2.Commits = []string{c1}
t1.Commits = []string{c2}
// One task successful, the other not.
t1.Status = types.TASK_STATUS_FAILURE
t1.Finished = now.Now(ctx)
t2.Status = types.TASK_STATUS_SUCCESS
t2.Finished = now.Now(ctx)
t2.IsolatedOutput = "abc123"
require.NoError(t, s.putTasks(ctx, []*types.Task{t1, t2}))
// Cycle. Ensure that we schedule a retry of t1.
prev := t1
i := 1
for {
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.UnfinishedTasks()
require.NoError(t, err)
if len(tasks) == 0 {
break
}
require.Len(t, tasks, 1)
retry := tasks[0]
require.NotNil(t, retry)
require.Equal(t, prev.Id, retry.RetryOf)
require.Equal(t, i, retry.Attempt)
require.Equal(t, c2, retry.Revision)
retry.Status = types.TASK_STATUS_FAILURE
retry.Finished = now.Now(ctx)
require.NoError(t, s.putTask(ctx, retry))
prev = retry
i++
}
require.Equal(t, 5, i)
}
func TestParentTaskId(t *testing.T) {
ctx, _, _, swarmingClient, s, _, cas, cleanup := setup(t)
defer cleanup()
// Run the available compile task at c2.
bot1 := makeBot("bot1", linuxTaskDims)
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err := s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, tasks, 1)
t1 := tasks[0]
t1.Status = types.TASK_STATUS_SUCCESS
t1.Finished = now.Now(ctx)
t1.IsolatedOutput = "abc123/45"
require.Equal(t, 0, len(t1.ParentTaskIds))
require.NoError(t, s.putTasks(ctx, []*types.Task{t1}))
// Run the dependent tasks. Ensure that their parent IDs are correct.
bot3 := makeBot("bot3", androidTaskDims)
bot4 := makeBot("bot4", androidTaskDims)
mockBots(t, swarmingClient, bot3, bot4)
cas.On("Merge", testutils.AnyContext, []string{"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb/42", "abc123/45"}).Return("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbabc123/87", nil)
cas.On("Merge", testutils.AnyContext, []string{"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc/42", "abc123/45"}).Return("ccccccccccccccccccccccccccccccccccccccccccccccccccccccccccabc123/87", nil)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, tasks, 2)
for _, task := range tasks {
require.Equal(t, 1, len(task.ParentTaskIds))
p := task.ParentTaskIds[0]
require.Equal(t, p, t1.Id)
updated, err := task.UpdateFromTaskResult(&types.TaskResult{
ID: task.SwarmingTaskId,
CasOutput: task.IsolatedOutput,
Created: task.Created,
Finished: task.Finished,
Started: task.Started,
Status: task.Status,
Tags: map[string][]string{
types.SWARMING_TAG_ID: {task.Id},
types.SWARMING_TAG_NAME: {task.Name},
types.SWARMING_TAG_REPO: {task.Repo},
types.SWARMING_TAG_REVISION: {task.Revision},
types.SWARMING_TAG_PARENT_TASK_ID: task.ParentTaskIds,
types.SWARMING_TAG_FORCED_JOB_ID: {task.ForcedJobId},
},
MachineID: task.SwarmingBotId,
})
require.NoError(t, err)
require.False(t, updated)
}
}
func TestSkipTasks(t *testing.T) {
// skip_tasks has its own tests, so this test just verifies that it's
// actually integrated into the scheduler.
ctx, _, _, swarmingClient, s, _, _, cleanup := setup(t)
defer cleanup()
c, cleanupfs := ftestutils.NewClientForTesting(context.Background(), t)
defer cleanupfs()
bl, err := skip_tasks.New(context.Background(), c)
require.NoError(t, err)
s.skipTasks = bl
c1 := rs1.Revision
// Mock some bots, add one of the build tasks to the skip_tasks list.
bot1 := makeBot("bot1", linuxTaskDims)
bot2 := makeBot("bot2", linuxTaskDims)
mockBots(t, swarmingClient, bot1, bot2)
require.NoError(t, s.GetSkipTasks().AddRule(ctx, &skip_tasks.Rule{
AddedBy: "Tests",
TaskSpecPatterns: []string{".*"},
Commits: []string{c1},
Description: "desc",
Name: "My-Rule",
}, s.repos))
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err := s.tCache.UnfinishedTasks()
require.NoError(t, err)
// The skipped commit should not have been triggered.
require.Len(t, tasks, 1)
require.NotEqual(t, c1, tasks[0].Revision)
// Candidate diagnostics should indicate the skip rule.
diag := lastDiagnostics(t, s)
foundSkipped := 0
for _, c := range diag.Candidates {
if c.Revision == c1 {
foundSkipped++
require.Equal(t, "My-Rule", c.Diagnostics.Filtering.SkippedByRule)
} else if c.TaskKey == tasks[0].TaskKey {
require.Nil(t, c.Diagnostics.Filtering)
} else {
require.Equal(t, "", c.Diagnostics.Filtering.SkippedByRule)
require.True(t, len(c.Diagnostics.Filtering.UnmetDependencies) > 0)
}
}
// Should be one Build task and one Test task skipped.
require.Equal(t, 2, foundSkipped)
}
func TestGetTasksForJob(t *testing.T) {
ctx, _, _, swarmingClient, s, _, cas, cleanup := setup(t)
defer cleanup()
c1 := rs1.Revision
c2 := rs2.Revision
// Cycle once, check that we have empty sets for all Jobs.
runMainLoop(t, s, ctx)
jobs, err := s.jCache.InProgressJobs()
require.NoError(t, err)
require.Len(t, jobs, 5)
var j1, j2, j3, j4, j5 *types.Job
for _, j := range jobs {
if j.Revision == c1 {
if j.Name == tcc_testutils.BuildTaskName {
j1 = j
} else {
j2 = j
}
} else {
if j.Name == tcc_testutils.BuildTaskName {
j3 = j
} else if j.Name == tcc_testutils.TestTaskName {
j4 = j
} else {
j5 = j
}
}
tasksByName, err := s.getTasksForJob(j)
require.NoError(t, err)
for _, tasks := range tasksByName {
require.Empty(t, tasks)
}
}
require.NotNil(t, j1)
require.NotNil(t, j2)
require.NotNil(t, j3)
require.NotNil(t, j4)
require.NotNil(t, j5)
// Run the available compile task at c2.
bot1 := makeBot("bot1", linuxTaskDims)
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err := s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, tasks, 1)
t1 := tasks[0]
require.NotNil(t, t1)
require.Equal(t, t1.Revision, c2)
// Test that we get the new tasks where applicable.
expect := map[string]map[string][]*types.Task{
j1.Id: {
tcc_testutils.BuildTaskName: {},
},
j2.Id: {
tcc_testutils.BuildTaskName: {},
tcc_testutils.TestTaskName: {},
},
j3.Id: {
tcc_testutils.BuildTaskName: {t1},
},
j4.Id: {
tcc_testutils.BuildTaskName: {t1},
tcc_testutils.TestTaskName: {},
},
j5.Id: {
tcc_testutils.BuildTaskName: {t1},
tcc_testutils.PerfTaskName: {},
},
}
for _, j := range jobs {
tasksByName, err := s.getTasksForJob(j)
require.NoError(t, err)
assertdeep.Equal(t, expect[j.Id], tasksByName)
}
// Mark the task as failed.
t1.Status = types.TASK_STATUS_FAILURE
t1.Finished = now.Now(ctx)
require.NoError(t, s.putTasks(ctx, []*types.Task{t1}))
// Test that the results propagated through.
for _, j := range jobs {
tasksByName, err := s.getTasksForJob(j)
require.NoError(t, err)
assertdeep.Equal(t, expect[j.Id], tasksByName)
}
// Cycle. Ensure that we schedule a retry of t1.
// Need two bots, since the retry will score lower than the Build task at c1.
bot2 := makeBot("bot2", linuxTaskDims)
mockBots(t, swarmingClient, bot1, bot2)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, tasks, 2)
var t2, t3 *types.Task
for _, task := range tasks {
if task.TaskKey == t1.TaskKey {
t2 = task
} else {
t3 = task
}
}
require.NotNil(t, t2)
require.Equal(t, t1.Id, t2.RetryOf)
// Verify that both the original t1 and its retry show up.
t1, err = s.tCache.GetTask(t1.Id) // t1 was updated.
require.NoError(t, err)
expect[j1.Id][tcc_testutils.BuildTaskName] = []*types.Task{t3}
expect[j2.Id][tcc_testutils.BuildTaskName] = []*types.Task{t3}
expect[j3.Id][tcc_testutils.BuildTaskName] = []*types.Task{t1, t2}
expect[j4.Id][tcc_testutils.BuildTaskName] = []*types.Task{t1, t2}
expect[j5.Id][tcc_testutils.BuildTaskName] = []*types.Task{t1, t2}
for _, j := range jobs {
tasksByName, err := s.getTasksForJob(j)
require.NoError(t, err)
assertdeep.Equal(t, expect[j.Id], tasksByName)
}
// The retry succeeded.
t2.Status = types.TASK_STATUS_SUCCESS
t2.Finished = now.Now(ctx)
t2.IsolatedOutput = "abc123/45"
// The Build at c1 failed.
t3.Status = types.TASK_STATUS_FAILURE
t3.Finished = now.Now(ctx)
require.NoError(t, s.putTasks(ctx, []*types.Task{t2, t3}))
mockBots(t, swarmingClient)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
tasks, err = s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Empty(t, tasks)
// Test that the results propagated through.
for _, j := range jobs {
tasksByName, err := s.getTasksForJob(j)
require.NoError(t, err)
assertdeep.Equal(t, expect[j.Id], tasksByName)
}
// Schedule the remaining tasks.
bot3 := makeBot("bot3", androidTaskDims)
bot4 := makeBot("bot4", androidTaskDims)
bot5 := makeBot("bot5", androidTaskDims)
mockBots(t, swarmingClient, bot3, bot4, bot5)
cas.On("Merge", testutils.AnyContext, []string{"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb/42", "abc123/45"}).Return("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbabc123/87", nil)
cas.On("Merge", testutils.AnyContext, []string{"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc/42", "abc123/45"}).Return("ccccccccccccccccccccccccccccccccccccccccccccccccccccccccccabc123/87", nil)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
// Verify that the new tasks show up.
tasks, err = s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, tasks, 2) // Test and perf at c2.
var t4, t5 *types.Task
for _, task := range tasks {
if task.Name == tcc_testutils.TestTaskName {
t4 = task
} else {
t5 = task
}
}
expect[j4.Id][tcc_testutils.TestTaskName] = []*types.Task{t4}
expect[j5.Id][tcc_testutils.PerfTaskName] = []*types.Task{t5}
for _, j := range jobs {
tasksByName, err := s.getTasksForJob(j)
require.NoError(t, err)
assertdeep.Equal(t, expect[j.Id], tasksByName)
}
}
func TestTaskTimeouts(t *testing.T) {
ctx, mg, _, swarmingClient, s, _, _, cleanup := setup(t)
defer cleanup()
// The test repo does not set any timeouts. Ensure that we get
// reasonable default values.
bot1 := makeBot("bot1", map[string]string{"pool": "Skia", "os": "Ubuntu", "gpu": "none"})
mockBots(t, swarmingClient, bot1)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
unfinished, err := s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, unfinished, 1)
task := unfinished[0]
swarmingTask, err := swarmingClient.GetTaskMetadata(ctx, task.SwarmingTaskId)
require.NoError(t, err)
// These are the defaults in go/swarming/swarming.go.
require.Equal(t, 1, len(swarmingTask.Request.TaskSlices))
require.Equal(t, int64(60*60), swarmingTask.Request.TaskSlices[0].Properties.ExecutionTimeoutSecs)
require.Equal(t, int64(20*60), swarmingTask.Request.TaskSlices[0].Properties.IoTimeoutSecs)
require.Equal(t, int64(4*60*60), swarmingTask.Request.TaskSlices[0].ExpirationSecs)
// Fail the task to get it out of the unfinished list.
task.Status = types.TASK_STATUS_FAILURE
require.NoError(t, s.putTask(ctx, task))
// Rewrite tasks.json with some timeouts.
name := "Timeout-Task"
cfg := &specs.TasksCfg{
CasSpecs: map[string]*specs.CasSpec{
"compile": {
Digest: tcc_testutils.CompileCASDigest,
},
},
Jobs: map[string]*specs.JobSpec{
"Timeout-Job": {
Priority: 1.0,
TaskSpecs: []string{name},
},
},
Tasks: map[string]*specs.TaskSpec{
name: {
CasSpec: "compile",
CipdPackages: []*specs.CipdPackage{},
Dependencies: []string{},
Dimensions: []string{
"pool:Skia",
"os:Mac",
"gpu:my-gpu",
},
ExecutionTimeout: 40 * time.Minute,
Expiration: 2 * time.Hour,
IoTimeout: 3 * time.Minute,
Priority: 1.0,
},
},
}
hashes := mg.CommitN(1)
require.NoError(t, s.repos.Update(ctx))
rs := types.RepoState{
Repo: "fake.git",
Revision: hashes[0],
}
fillCaches(t, ctx, s.taskCfgCache, rs, cfg)
insertJobs(t, ctx, s, rs)
// Cycle, ensure that we get the expected timeouts.
bot2 := makeBot("bot2", map[string]string{"pool": "Skia", "os": "Mac", "gpu": "my-gpu"})
mockBots(t, swarmingClient, bot2)
runMainLoop(t, s, ctx)
require.NoError(t, s.tCache.Update(ctx))
unfinished, err = s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, unfinished, 1)
task = unfinished[0]
require.Equal(t, name, task.Name)
swarmingTask, err = swarmingClient.GetTaskMetadata(ctx, task.SwarmingTaskId)
require.NoError(t, err)
require.Equal(t, 1, len(swarmingTask.Request.TaskSlices))
require.Equal(t, int64(40*60), swarmingTask.Request.TaskSlices[0].Properties.ExecutionTimeoutSecs)
require.Equal(t, int64(3*60), swarmingTask.Request.TaskSlices[0].Properties.IoTimeoutSecs)
require.Equal(t, int64(2*60*60), swarmingTask.Request.TaskSlices[0].ExpirationSecs)
}
func TestUpdateUnfinishedTasks(t *testing.T) {
ctx, _, _, swarmingClient, s, _, _, cleanup := setup(t)
defer cleanup()
// Create a few tasks.
now := time.Unix(1480683321, 0).UTC()
t1 := &types.Task{
Id: "t1",
Created: now.Add(-time.Minute),
Status: types.TASK_STATUS_RUNNING,
SwarmingTaskId: "swarmt1",
}
t2 := &types.Task{
Id: "t2",
Created: now.Add(-10 * time.Minute),
Status: types.TASK_STATUS_PENDING,
SwarmingTaskId: "swarmt2",
}
t3 := &types.Task{
Id: "t3",
Created: now.Add(-5 * time.Hour), // Outside the 4-hour window.
Status: types.TASK_STATUS_PENDING,
SwarmingTaskId: "swarmt3",
}
// Include a fake task to ensure it's ignored.
t4 := &types.Task{
Id: "t4",
Created: now.Add(-time.Minute),
Status: types.TASK_STATUS_PENDING,
}
// Insert the tasks into the DB.
tasks := []*types.Task{t1, t2, t3, t4}
require.NoError(t, s.putTasks(ctx, tasks))
// Update the tasks, mock in Swarming.
t1.Status = types.TASK_STATUS_SUCCESS
t1.IsolatedOutput = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd/86"
t2.Status = types.TASK_STATUS_FAILURE
t3.Status = types.TASK_STATUS_SUCCESS
t3.IsolatedOutput = "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff/256"
m1 := makeSwarmingRpcsTaskRequestMetadata(t, t1, linuxTaskDims)
m2 := makeSwarmingRpcsTaskRequestMetadata(t, t2, linuxTaskDims)
m3 := makeSwarmingRpcsTaskRequestMetadata(t, t3, linuxTaskDims)
swarmingClient.MockTasks([]*swarming_api.SwarmingRpcsTaskRequestMetadata{m1, m2, m3})
// Assert that the third task doesn't show up in the time range query.
got, err := swarmingClient.ListTasks(ctx, now.Add(-4*time.Hour), now, []string{"pool:Skia"}, "")
require.NoError(t, err)
assertdeep.Equal(t, []*swarming_api.SwarmingRpcsTaskRequestMetadata{m1, m2}, got)
// Ensure that we update the tasks as expected.
require.NoError(t, s.updateUnfinishedTasks(ctx))
for _, task := range tasks {
got, err := s.db.GetTaskById(ctx, task.Id)
require.NoError(t, err)
// Ignore DbModified when comparing.
task.DbModified = got.DbModified
assertdeep.Equal(t, task, got)
}
}
// setupAddTasksTest calls setup then adds 7 commits to the repo and returns
// their hashes.
func setupAddTasksTest(t *testing.T) (context.Context, *mem_git.MemGit, []string, *memory.InMemoryDB, *TaskScheduler, func()) {
ctx, mg, d, _, s, _, _, cleanup := setup(t)
// Add some commits to test blamelist calculation.
mg.CommitN(7)
require.NoError(t, s.repos.Update(ctx))
hashes, err := s.repos[rs1.Repo].Get(git.MainBranch).AllCommits()
require.NoError(t, err)
for _, hash := range hashes {
require.NoError(t, s.taskCfgCache.Set(ctx, types.RepoState{
Repo: rs1.Repo,
Revision: hash,
}, &specs.TasksCfg{
Tasks: map[string]*specs.TaskSpec{
"duty": {},
"onus": {},
"toil": {},
"work": {},
},
}, nil))
}
return ctx, mg, hashes, d, s, func() {
cleanup()
}
}
// assertBlamelist asserts task.Commits contains exactly the hashes at the given
// indexes of hashes, in any order.
func assertBlamelist(t *testing.T, hashes []string, task *types.Task, indexes []int) {
expected := util.NewStringSet()
for _, idx := range indexes {
expected[hashes[idx]] = true
}
require.Equal(t, expected, util.NewStringSet(task.Commits))
}
// assertModifiedTasks asserts that the result of GetModifiedTasks is deep-equal
// to expected, in any order.
func assertModifiedTasks(t *testing.T, d db.TaskReader, mod <-chan []*types.Task, expected []*types.Task) {
tasksById := map[string]*types.Task{}
require.Eventually(t, func() bool {
// Use a select so that the test will fail after 10 seconds
// rather than time out after 10 minutes (or whatever the
// overall timeout is set to).
select {
case modTasks := <-mod:
for _, task := range modTasks {
tasksById[task.Id] = task
}
for _, expectedTask := range expected {
actualTask, ok := tasksById[expectedTask.Id]
if !ok {
return false
}
if !deepequal.DeepEqual(expectedTask, actualTask) {
return false
}
}
return true
default:
// Nothing to do.
}
return false
}, 10*time.Second, 50*time.Millisecond)
}
// addTasksSingleTaskSpec should add tasks and compute simple blamelists.
func TestAddTasksSingleTaskSpecSimple(t *testing.T) {
ctx, _, hashes, d, s, cleanup := setupAddTasksTest(t)
defer cleanup()
t1 := makeTask(ctx, "toil", rs1.Repo, hashes[6])
require.NoError(t, s.putTask(ctx, t1))
mod := d.ModifiedTasksCh(ctx)
<-mod // The first batch is unused.
t2 := makeTask(ctx, "toil", rs1.Repo, hashes[5]) // Commits should be {5}
t3 := makeTask(ctx, "toil", rs1.Repo, hashes[3]) // Commits should be {3, 4}
t4 := makeTask(ctx, "toil", rs1.Repo, hashes[2]) // Commits should be {2}
t5 := makeTask(ctx, "toil", rs1.Repo, hashes[0]) // Commits should be {0, 1}
// Clear Commits on some tasks, set incorrect Commits on others to
// ensure it's ignored.
t3.Commits = nil
t4.Commits = []string{hashes[5], hashes[4], hashes[3], hashes[2]}
sort.Strings(t4.Commits)
// Specify tasks in wrong order to ensure results are deterministic.
require.NoError(t, s.addTasksSingleTaskSpec(ctx, []*types.Task{t5, t2, t3, t4}))
assertBlamelist(t, hashes, t2, []int{5})
assertBlamelist(t, hashes, t3, []int{3, 4})
assertBlamelist(t, hashes, t4, []int{2})
assertBlamelist(t, hashes, t5, []int{0, 1})
// Check that the tasks were inserted into the DB.
assertModifiedTasks(t, d, mod, []*types.Task{t2, t3, t4, t5})
}
// addTasksSingleTaskSpec should compute blamelists when new tasks bisect each
// other.
func TestAddTasksSingleTaskSpecBisectNew(t *testing.T) {
ctx, _, hashes, d, s, cleanup := setupAddTasksTest(t)
defer cleanup()
t1 := makeTask(ctx, "toil", rs1.Repo, hashes[6])
require.NoError(t, s.putTask(ctx, t1))
mod := d.ModifiedTasksCh(ctx)
<-mod // The first batch is unused.
// t2.Commits = {1, 2, 3, 4, 5}
t2 := makeTask(ctx, "toil", rs1.Repo, hashes[1])
// t3.Commits = {3, 4, 5}
// t2.Commits = {1, 2}
t3 := makeTask(ctx, "toil", rs1.Repo, hashes[3])
// t4.Commits = {4, 5}
// t3.Commits = {3}
t4 := makeTask(ctx, "toil", rs1.Repo, hashes[4])
// t5.Commits = {0}
t5 := makeTask(ctx, "toil", rs1.Repo, hashes[0])
// t6.Commits = {2}
// t2.Commits = {1}
t6 := makeTask(ctx, "toil", rs1.Repo, hashes[2])
// t7.Commits = {1}
// t2.Commits = {}
t7 := makeTask(ctx, "toil", rs1.Repo, hashes[1])
// Specify tasks in wrong order to ensure results are deterministic.
tasks := []*types.Task{t5, t2, t7, t3, t6, t4}
// Assign Ids.
for _, task := range tasks {
require.NoError(t, d.AssignId(ctx, task))
}
require.NoError(t, s.addTasksSingleTaskSpec(ctx, tasks))
assertBlamelist(t, hashes, t2, []int{})
assertBlamelist(t, hashes, t3, []int{3})
assertBlamelist(t, hashes, t4, []int{4, 5})
assertBlamelist(t, hashes, t5, []int{0})
assertBlamelist(t, hashes, t6, []int{2})
assertBlamelist(t, hashes, t7, []int{1})
// Check that the tasks were inserted into the DB.
assertModifiedTasks(t, d, mod, tasks)
}
// addTasksSingleTaskSpec should compute blamelists when new tasks bisect old
// tasks.
func TestAddTasksSingleTaskSpecBisectOld(t *testing.T) {
ctx, _, hashes, d, s, cleanup := setupAddTasksTest(t)
defer cleanup()
t1 := makeTask(ctx, "toil", rs1.Repo, hashes[6])
t2 := makeTask(ctx, "toil", rs1.Repo, hashes[1])
t2.Commits = []string{hashes[1], hashes[2], hashes[3], hashes[4], hashes[5]}
sort.Strings(t2.Commits)
require.NoError(t, s.putTasks(ctx, []*types.Task{t1, t2}))
mod := d.ModifiedTasksCh(ctx)
<-mod // The first batch is unused.
// t3.Commits = {3, 4, 5}
// t2.Commits = {1, 2}
t3 := makeTask(ctx, "toil", rs1.Repo, hashes[3])
// t4.Commits = {4, 5}
// t3.Commits = {3}
t4 := makeTask(ctx, "toil", rs1.Repo, hashes[4])
// t5.Commits = {2}
// t2.Commits = {1}
t5 := makeTask(ctx, "toil", rs1.Repo, hashes[2])
// t6.Commits = {4, 5}
// t4.Commits = {}
t6 := makeTask(ctx, "toil", rs1.Repo, hashes[4])
// Specify tasks in wrong order to ensure results are deterministic.
require.NoError(t, s.addTasksSingleTaskSpec(ctx, []*types.Task{t5, t3, t6, t4}))
t2Updated, err := d.GetTaskById(ctx, t2.Id)
require.NoError(t, err)
assertBlamelist(t, hashes, t2Updated, []int{1})
assertBlamelist(t, hashes, t3, []int{3})
assertBlamelist(t, hashes, t4, []int{})
assertBlamelist(t, hashes, t5, []int{2})
assertBlamelist(t, hashes, t6, []int{4, 5})
// Check that the tasks were inserted into the DB.
t2.Commits = t2Updated.Commits
t2.DbModified = t2Updated.DbModified
assertModifiedTasks(t, d, mod, []*types.Task{t2, t3, t4, t5, t6})
}
// addTasksSingleTaskSpec should update existing tasks, keeping the correct
// blamelist.
func TestAddTasksSingleTaskSpecUpdate(t *testing.T) {
ctx, _, hashes, d, s, cleanup := setupAddTasksTest(t)
defer cleanup()
t1 := makeTask(ctx, "toil", rs1.Repo, hashes[6])
t2 := makeTask(ctx, "toil", rs1.Repo, hashes[3])
t3 := makeTask(ctx, "toil", rs1.Repo, hashes[4])
t3.Commits = nil // Stolen by t5
t4 := makeTask(ctx, "toil", rs1.Repo, hashes[0])
t4.Commits = []string{hashes[0], hashes[1], hashes[2]}
sort.Strings(t4.Commits)
t5 := makeTask(ctx, "toil", rs1.Repo, hashes[4])
t5.Commits = []string{hashes[4], hashes[5]}
sort.Strings(t5.Commits)
tasks := []*types.Task{t1, t2, t3, t4, t5}
require.NoError(t, s.putTasks(ctx, tasks))
mod := d.ModifiedTasksCh(ctx)
<-mod // The first batch is unused.
// Make an update.
for _, task := range tasks {
task.Status = types.TASK_STATUS_MISHAP
}
// Specify tasks in wrong order to ensure results are deterministic.
require.NoError(t, s.addTasksSingleTaskSpec(ctx, []*types.Task{t5, t3, t1, t4, t2}))
// Check that blamelists did not change.
assertBlamelist(t, hashes, t1, []int{6})
assertBlamelist(t, hashes, t2, []int{3})
assertBlamelist(t, hashes, t3, []int{})
assertBlamelist(t, hashes, t4, []int{0, 1, 2})
assertBlamelist(t, hashes, t5, []int{4, 5})
// Check that the tasks were inserted into the DB.
assertModifiedTasks(t, d, mod, []*types.Task{t1, t2, t3, t4, t5})
}
// AddTasks should call addTasksSingleTaskSpec for each group of tasks.
func TestAddTasks(t *testing.T) {
ctx, _, hashes, d, s, cleanup := setupAddTasksTest(t)
defer cleanup()
toil1 := makeTask(ctx, "toil", rs1.Repo, hashes[6])
duty1 := makeTask(ctx, "duty", rs1.Repo, hashes[6])
work1 := makeTask(ctx, "work", rs1.Repo, hashes[6])
work2 := makeTask(ctx, "work", rs1.Repo, hashes[1])
work2.Commits = []string{hashes[1], hashes[2], hashes[3], hashes[4], hashes[5]}
sort.Strings(work2.Commits)
onus1 := makeTask(ctx, "onus", rs1.Repo, hashes[6])
onus2 := makeTask(ctx, "onus", rs1.Repo, hashes[3])
onus2.Commits = []string{hashes[3], hashes[4], hashes[5]}
sort.Strings(onus2.Commits)
require.NoError(t, s.putTasks(ctx, []*types.Task{toil1, duty1, work1, work2, onus1, onus2}))
mod := d.ModifiedTasksCh(ctx)
<-mod // The first batch is unused.
// toil2.Commits = {5}
toil2 := makeTask(ctx, "toil", rs1.Repo, hashes[5])
// toil3.Commits = {3, 4}
toil3 := makeTask(ctx, "toil", rs1.Repo, hashes[3])
// duty2.Commits = {1, 2, 3, 4, 5}
duty2 := makeTask(ctx, "duty", rs1.Repo, hashes[1])
// duty3.Commits = {3, 4, 5}
// duty2.Commits = {1, 2}
duty3 := makeTask(ctx, "duty", rs1.Repo, hashes[3])
// work3.Commits = {3, 4, 5}
// work2.Commits = {1, 2}
work3 := makeTask(ctx, "work", rs1.Repo, hashes[3])
// work4.Commits = {2}
// work2.Commits = {1}
work4 := makeTask(ctx, "work", rs1.Repo, hashes[2])
onus2.Status = types.TASK_STATUS_MISHAP
// onus3 steals all commits from onus2
onus3 := makeTask(ctx, "onus", rs1.Repo, hashes[3])
// onus4 steals all commits from onus3
onus4 := makeTask(ctx, "onus", rs1.Repo, hashes[3])
tasks := map[string]map[string][]*types.Task{
rs1.Repo: {
"toil": {toil2, toil3},
"duty": {duty2, duty3},
"work": {work3, work4},
"onus": {onus2, onus3, onus4},
},
}
require.NoError(t, s.addTasks(ctx, tasks))
assertBlamelist(t, hashes, toil2, []int{5})
assertBlamelist(t, hashes, toil3, []int{3, 4})
assertBlamelist(t, hashes, duty2, []int{1, 2})
assertBlamelist(t, hashes, duty3, []int{3, 4, 5})
work2Updated, err := d.GetTaskById(ctx, work2.Id)
require.NoError(t, err)
assertBlamelist(t, hashes, work2Updated, []int{1})
assertBlamelist(t, hashes, work3, []int{3, 4, 5})
assertBlamelist(t, hashes, work4, []int{2})
assertBlamelist(t, hashes, onus2, []int{})
assertBlamelist(t, hashes, onus3, []int{})
assertBlamelist(t, hashes, onus4, []int{3, 4, 5})
// Check that the tasks were inserted into the DB.
work2.Commits = work2Updated.Commits
work2.DbModified = work2Updated.DbModified
assertModifiedTasks(t, d, mod, []*types.Task{toil2, toil3, duty2, duty3, work2, work3, work4, onus2, onus3, onus4})
}
// AddTasks should not leave DB in an inconsistent state if there is a partial error.
func TestAddTasksFailure(t *testing.T) {
ctx, _, hashes, d, s, cleanup := setupAddTasksTest(t)
defer cleanup()
toil1 := makeTask(ctx, "toil", rs1.Repo, hashes[6])
duty1 := makeTask(ctx, "duty", rs1.Repo, hashes[6])
duty2 := makeTask(ctx, "duty", rs1.Repo, hashes[5])
require.NoError(t, s.putTasks(ctx, []*types.Task{toil1, duty1, duty2}))
d.Wait()
// Cause ErrConcurrentUpdate in AddTasks.
cachedDuty2 := duty2.Copy()
duty2.Status = types.TASK_STATUS_MISHAP
require.NoError(t, d.PutTask(ctx, duty2))
d.Wait()
mod := d.ModifiedTasksCh(ctx)
<-mod // The first batch is unused.
// toil2.Commits = {3, 4, 5}
toil2 := makeTask(ctx, "toil", rs1.Repo, hashes[3])
cachedDuty2.Status = types.TASK_STATUS_FAILURE
// duty3.Commits = {3, 4}
duty3 := makeTask(ctx, "duty", rs1.Repo, hashes[3])
tasks := map[string]map[string][]*types.Task{
rs1.Repo: {
"toil": {toil2},
"duty": {cachedDuty2, duty3},
},
}
// Try multiple times to reduce chance of test passing flakily.
for i := 0; i < 3; i++ {
err := s.addTasks(ctx, tasks)
require.Error(t, err)
modTasks := <-mod
// "duty" tasks should never be updated.
for _, task := range modTasks {
require.Equal(t, "toil", task.Name)
assertdeep.Equal(t, toil2, task)
assertBlamelist(t, hashes, toil2, []int{3, 4, 5})
}
}
duty2.Status = types.TASK_STATUS_FAILURE
tasks[rs1.Repo]["duty"] = []*types.Task{duty2, duty3}
require.NoError(t, s.addTasks(ctx, tasks))
assertBlamelist(t, hashes, toil2, []int{3, 4, 5})
assertBlamelist(t, hashes, duty2, []int{5})
assertBlamelist(t, hashes, duty3, []int{3, 4})
// Check that the tasks were inserted into the DB.
assertModifiedTasks(t, d, mod, []*types.Task{toil2, duty2, duty3})
}
// AddTasks should retry on ErrConcurrentUpdate.
func TestAddTasksRetries(t *testing.T) {
ctx, _, hashes, d, s, cleanup := setupAddTasksTest(t)
defer cleanup()
toil1 := makeTask(ctx, "toil", rs1.Repo, hashes[6])
duty1 := makeTask(ctx, "duty", rs1.Repo, hashes[6])
work1 := makeTask(ctx, "work", rs1.Repo, hashes[6])
toil2 := makeTask(ctx, "toil", rs1.Repo, hashes[1])
toil2.Commits = []string{hashes[1], hashes[2], hashes[3], hashes[4], hashes[5]}
sort.Strings(toil2.Commits)
duty2 := makeTask(ctx, "duty", rs1.Repo, hashes[1])
duty2.Commits = util.CopyStringSlice(toil2.Commits)
work2 := makeTask(ctx, "work", rs1.Repo, hashes[1])
work2.Commits = util.CopyStringSlice(toil2.Commits)
require.NoError(t, s.putTasks(ctx, []*types.Task{toil1, toil2, duty1, duty2, work1, work2}))
mod := d.ModifiedTasksCh(ctx)
<-mod // The first batch is unused.
// *3.Commits = {3, 4, 5}
// *2.Commits = {1, 2}
toil3 := makeTask(ctx, "toil", rs1.Repo, hashes[3])
duty3 := makeTask(ctx, "duty", rs1.Repo, hashes[3])
work3 := makeTask(ctx, "work", rs1.Repo, hashes[3])
// *4.Commits = {2}
// *2.Commits = {1}
toil4 := makeTask(ctx, "toil", rs1.Repo, hashes[2])
duty4 := makeTask(ctx, "duty", rs1.Repo, hashes[2])
work4 := makeTask(ctx, "work", rs1.Repo, hashes[2])
tasks := map[string]map[string][]*types.Task{
rs1.Repo: {
"toil": {toil3.Copy(), toil4.Copy()},
"duty": {duty3.Copy(), duty4.Copy()},
"work": {work3.Copy(), work4.Copy()},
},
}
retryCountMtx := sync.Mutex{}
retryCount := map[string]int{}
causeConcurrentUpdate := func(tasks []*types.Task) {
retryCountMtx.Lock()
defer retryCountMtx.Unlock()
retryCount[tasks[0].Name]++
if tasks[0].Name == "toil" && retryCount["toil"] < 2 {
toil2.Started = now.Now(ctx).UTC()
require.NoError(t, d.PutTasks(ctx, []*types.Task{toil2}))
s.tCache.AddTasks([]*types.Task{toil2})
}
if tasks[0].Name == "duty" && retryCount["duty"] < 3 {
duty2.Started = now.Now(ctx).UTC()
require.NoError(t, d.PutTasks(ctx, []*types.Task{duty2}))
s.tCache.AddTasks([]*types.Task{duty2})
}
if tasks[0].Name == "work" && retryCount["work"] < 4 {
work2.Started = now.Now(ctx).UTC()
require.NoError(t, d.PutTasks(ctx, []*types.Task{work2}))
s.tCache.AddTasks([]*types.Task{work2})
}
}
s.db = &spyDB{
DB: d,
onPutTasks: causeConcurrentUpdate,
}
require.NoError(t, s.addTasks(ctx, tasks))
retryCountMtx.Lock()
defer retryCountMtx.Unlock()
require.Equal(t, 2, retryCount["toil"])
require.Equal(t, 3, retryCount["duty"])
require.Equal(t, 4, retryCount["work"])
modified := []*types.Task{}
check := func(t2, t3, t4 *types.Task) {
t2InDB, err := d.GetTaskById(ctx, t2.Id)
require.NoError(t, err)
assertBlamelist(t, hashes, t2InDB, []int{1})
t3Arg := tasks[t3.Repo][t3.Name][0]
t3InDB, err := d.GetTaskById(ctx, t3Arg.Id)
require.NoError(t, err)
assertdeep.Equal(t, t3Arg, t3InDB)
assertBlamelist(t, hashes, t3InDB, []int{3, 4, 5})
t4Arg := tasks[t4.Repo][t4.Name][1]
t4InDB, err := d.GetTaskById(ctx, t4Arg.Id)
require.NoError(t, err)
assertdeep.Equal(t, t4Arg, t4InDB)
assertBlamelist(t, hashes, t4InDB, []int{2})
t2.Commits = t2InDB.Commits
t2.DbModified = t2InDB.DbModified
t3.Id = t3InDB.Id
t3.Commits = t3InDB.Commits
t3.DbModified = t3InDB.DbModified
t4.Id = t4InDB.Id
t4.Commits = t4InDB.Commits
t4.DbModified = t4InDB.DbModified
modified = append(modified, t2, t3, t4)
}
check(toil2, toil3, toil4)
check(duty2, duty3, duty4)
check(work2, work3, work4)
assertModifiedTasks(t, d, mod, modified)
}
func TestTriggerTaskFailed(t *testing.T) {
// Verify that if one task out of a set fails to trigger, the others are
// still inserted into the DB and handled properly, eg. wrt. blamelists.
ctx, _, _, s, swarmingClient, commits, _, _, cleanup := testMultipleCandidatesBackfillingEachOtherSetup(t)
defer cleanup()
// Trigger three tasks. We should attempt to trigger tasks at
// commits[0], commits[4], and either commits[2] or commits[6]. Mock
// failure to trigger the task at commits[4] and ensure that the other
// two tasks get inserted with the correct blamelists.
bot1 := makeBot("bot1", map[string]string{"pool": "Skia"})
bot2 := makeBot("bot2", map[string]string{"pool": "Skia"})
bot3 := makeBot("bot3", map[string]string{"pool": "Skia"})
makeTags := func(commit string) []string {
return []string{
"luci_project:",
"milo_host:https://ci.chromium.org/raw/build/%s",
"sk_attempt:0",
"sk_dim_pool:Skia",
"sk_retry_of:",
fmt.Sprintf("source_revision:%s", commit),
"source_repo:" + fmt.Sprintf(gitiles.CommitURL, rs1.Repo, "%s"),
fmt.Sprintf("sk_repo:%s", rs1.Repo),
fmt.Sprintf("sk_revision:%s", commit),
"sk_forced_job_id:",
"sk_name:faketask",
}
}
mockBots(t, swarmingClient, bot1, bot2, bot3)
swarmingClient.MockTriggerTaskFailure(makeTags(commits[4]))
err := s.MainLoop(ctx)
s.testWaitGroup.Wait()
require.NotNil(t, err)
require.True(t, strings.Contains(err.Error(), "Mocked trigger failure!"))
require.NoError(t, s.tCache.Update(ctx))
require.Equal(t, 6, len(s.queue))
tasks, err := s.tCache.GetTasksForCommits(rs1.Repo, commits)
require.NoError(t, err)
var t1, t2, t3 *types.Task
for _, byName := range tasks {
for _, task := range byName {
if task.Revision == commits[0] {
t1 = task
} else if task.Revision == commits[4] {
t2 = task
} else if task.Revision == commits[2] || task.Revision == commits[6] {
t3 = task
} else {
require.FailNow(t, fmt.Sprintf("Task has unknown revision: %v", task))
}
}
}
require.NotNil(t, t1)
require.Nil(t, t2)
require.NotNil(t, t3)
// Ensure that we got the blamelists right.
var expect1, expect3 []string
if t3.Revision == commits[2] {
expect1 = util.CopyStringSlice(commits[:2])
expect3 = util.CopyStringSlice(commits[2:])
} else {
expect1 = util.CopyStringSlice(commits[:6])
expect3 = util.CopyStringSlice(commits[6:])
}
sort.Strings(expect1)
sort.Strings(expect3)
sort.Strings(t1.Commits)
sort.Strings(t3.Commits)
assertdeep.Equal(t, expect1, t1.Commits)
assertdeep.Equal(t, expect3, t3.Commits)
// Check diagnostics.
diag := lastDiagnostics(t, s)
failedTrigger := 0
for _, c := range diag.Candidates {
if c.Revision == commits[4] {
require.True(t, strings.Contains(c.Diagnostics.Triggering.TriggerError, "Mocked trigger failure!"))
failedTrigger++
} else {
if c.TaskKey == t1.TaskKey {
require.Equal(t, "", c.Diagnostics.Triggering.TriggerError)
require.Equal(t, t1.Id, c.Diagnostics.Triggering.TaskId)
} else if c.TaskKey == t3.TaskKey {
require.Equal(t, "", c.Diagnostics.Triggering.TriggerError)
require.Equal(t, t3.Id, c.Diagnostics.Triggering.TaskId)
} else {
require.Nil(t, c.Diagnostics.Triggering)
}
}
}
// Should be one task that failed
require.Equal(t, 1, failedTrigger)
}
const badTaskName = "badtask"
type mockDB struct {
db.DB
failAssignId bool
failPutTasks bool
}
func (d *mockDB) AssignId(ctx context.Context, t *types.Task) error {
if t.Name == badTaskName && d.failAssignId {
return errors.New(badTaskName)
}
return d.DB.AssignId(ctx, t)
}
func (d *mockDB) PutTasks(ctx context.Context, tasks []*types.Task) error {
for _, t := range tasks {
if t.Name == badTaskName && d.failPutTasks {
return errors.New(badTaskName)
}
}
return d.DB.PutTasks(ctx, tasks)
}
func TestContinueOnTriggerTaskFailure(t *testing.T) {
// Verify that if one task out of a set fails any part of the triggering
// process, the others are still triggered, inserted into the DB, etc.
ctx, mg, _, s, swarmingClient, commits, _, cfg, cleanup := testMultipleCandidatesBackfillingEachOtherSetup(t)
defer cleanup()
// Add several more commits.
newCommits := mg.CommitN(10)
commits = append(newCommits, commits...)
badCommit := commits[0]
badTaskName := "badtask"
badPool := "BadPool"
s.pools = []string{"Skia", badPool}
require.NoError(t, s.repos.Update(ctx))
for _, hash := range newCommits {
rs := types.RepoState{
Repo: rs1.Repo,
Revision: hash,
}
c := cfg.Copy()
if hash == badCommit {
c.Tasks[badTaskName] = &specs.TaskSpec{
CasSpec: "compile",
CipdPackages: []*specs.CipdPackage{},
Dependencies: []string{},
Dimensions: []string{fmt.Sprintf("pool:%s", badPool)},
Priority: 1.0,
}
c.Jobs["badjob"] = &specs.JobSpec{
TaskSpecs: []string{badTaskName},
}
}
fillCaches(t, ctx, s.taskCfgCache, rs, c)
insertJobs(t, ctx, s, rs)
}
finishAll := func() {
tasks, err := s.tCache.UnfinishedTasks()
require.NoError(t, err)
for _, task := range tasks {
task.Finished = now.Now(ctx)
task.Status = types.TASK_STATUS_SUCCESS
}
require.NoError(t, s.db.PutTasks(ctx, tasks))
}
finishAll() // The setup func triggers a task.
badTags := []string{
"luci_project:",
"milo_host:https://ci.chromium.org/raw/build/%s",
"sk_attempt:0",
fmt.Sprintf("sk_dim_pool:%s", badPool),
"sk_retry_of:",
fmt.Sprintf("source_revision:%s", badCommit),
"source_repo:" + fmt.Sprintf(gitiles.CommitURL, rs1.Repo, "%s"),
fmt.Sprintf("sk_repo:%s", rs1.Repo),
fmt.Sprintf("sk_revision:%s", badCommit),
"sk_forced_job_id:",
fmt.Sprintf("sk_name:%s", badTaskName),
}
test := func() {
// Pretend there are two bots available.
bots := []*types.Machine{
makeBot("bot0", map[string]string{"pool": "Skia"}),
makeBot("bot1", map[string]string{"pool": badPool}),
}
mockBots(t, swarmingClient, bots...)
// Run MainLoop.
err := s.MainLoop(ctx)
s.testWaitGroup.Wait()
require.NotNil(t, err)
require.NoError(t, s.tCache.Update(ctx))
// We'll try to trigger all tasks but the one for the bad commit will
// fail. Ensure that we triggered all of the others.
tasks, err := s.tCache.UnfinishedTasks()
require.NoError(t, err)
require.Len(t, tasks, 1)
require.NotEqual(t, badTaskName, tasks[0].Name)
// Clean up for the next iteration.
finishAll()
}
// 1. Actual triggering failed.
swarmingClient.MockTriggerTaskFailure(badTags)
test()
// 2. DB.AssignId failed.
mdb := &mockDB{
DB: s.db,
failAssignId: true,
}
s.db = mdb
test()
// 3. DB.PutTasks failed.
mdb.failAssignId = false
mdb.failPutTasks = true
test()
// 4. Failure to load details of de-duplicated task after triggering.
mdb.failPutTasks = false
swarmingClient.MockTriggerTaskDeduped(badTags, "thistagwontparse")
test()
s.db = mdb.DB
// 5. Failure to merge CAS entries.
// TODO(borenet)
}
func TestTriggerTaskDeduped(t *testing.T) {
// Verify that we properly handle de-duplicated tasks.
ctx, _, _, s, swarmingClient, commits, _, _, cleanup := testMultipleCandidatesBackfillingEachOtherSetup(t)
defer cleanup()
// Trigger three tasks. We should attempt to trigger tasks at
// commits[0], commits[4], and either commits[2] or commits[6]. Mock
// deduplication of the task at commits[4] and ensure that the other
// two tasks are not deduped.
bot1 := makeBot("bot1", map[string]string{"pool": "Skia"})
bot2 := makeBot("bot2", map[string]string{"pool": "Skia"})
bot3 := makeBot("bot3", map[string]string{"pool": "Skia"})
makeTags := func(commit string) []string {
return []string{
"luci_project:",
"milo_host:https://ci.chromium.org/raw/build/%s",
"sk_attempt:0",
"sk_dim_pool:Skia",
"sk_retry_of:",
fmt.Sprintf("source_revision:%s", commit),
"source_repo:" + fmt.Sprintf(gitiles.CommitURL, rs1.Repo, "%s"),
fmt.Sprintf("sk_repo:%s", rs1.Repo),
fmt.Sprintf("sk_revision:%s", commit),
"sk_forced_job_id:",
"sk_name:faketask",
}
}
mockBots(t, swarmingClient, bot1, bot2, bot3)
swarmingClient.MockTriggerTaskDeduped(makeTags(commits[4]))
require.NoError(t, s.MainLoop(ctx))
s.testWaitGroup.Wait()
require.NoError(t, s.tCache.Update(ctx))
require.Equal(t, 5, len(s.queue))
tasks, err := s.tCache.GetTasksForCommits(rs1.Repo, commits)
require.NoError(t, err)
var t1, t2, t3 *types.Task
for _, byName := range tasks {
for _, task := range byName {
if task.Revision == commits[0] {
t1 = task
} else if task.Revision == commits[4] {
t2 = task
} else if task.Revision == commits[2] || task.Revision == commits[6] {
t3 = task
} else {
require.FailNow(t, fmt.Sprintf("Task has unknown revision: %v", task))
}
}
}
require.NotNil(t, t1)
require.NotNil(t, t2)
require.NotNil(t, t3)
// Ensure that t2 was correctly deduped, and the others weren't.
require.Equal(t, types.TASK_STATUS_PENDING, t1.Status)
require.Equal(t, types.TASK_STATUS_SUCCESS, t2.Status)
require.Equal(t, types.TASK_STATUS_PENDING, t3.Status)
}
func TestTriggerTaskNoResource(t *testing.T) {
// Verify that we properly handle tasks which are rejected due to lack
// of matching bots.
ctx, _, _, s, swarmingClient, commits, _, _, cleanup := testMultipleCandidatesBackfillingEachOtherSetup(t)
defer cleanup()
// Attempt to trigger a task. It gets rejected because the bot we
// thought was available to run it is now busy/quarantined/offline.
bot1 := makeBot("bot1", map[string]string{"pool": "Skia"})
makeTags := func(commit string) []string {
return []string{
"luci_project:",
"milo_host:https://ci.chromium.org/raw/build/%s",
"sk_attempt:0",
"sk_dim_pool:Skia",
"sk_retry_of:",
fmt.Sprintf("source_revision:%s", commit),
"source_repo:" + fmt.Sprintf(gitiles.CommitURL, rs1.Repo, "%s"),
fmt.Sprintf("sk_repo:%s", rs1.Repo),
fmt.Sprintf("sk_revision:%s", commit),
"sk_forced_job_id:",
"sk_name:faketask",
}
}
mockBots(t, swarmingClient, bot1)
swarmingClient.MockTriggerTaskNoResource(makeTags(commits[0]))
err := s.MainLoop(ctx)
require.NotNil(t, err)
require.True(t, strings.Contains(err.Error(), "No bots available to run faketask with dimensions: pool:Skia"))
s.testWaitGroup.Wait()
require.NoError(t, s.tCache.Update(ctx))
require.Equal(t, 8, len(s.queue))
tasks, err := s.tCache.GetTasksForCommits(rs1.Repo, []string{commits[0]})
require.NoError(t, err)
require.Equal(t, 0, len(tasks[commits[0]]))
}
func TestScoreCandidate_TryJob_PrioritizedHigherByWaitTime(t *testing.T) {
ctx := context.Background()
test := func(name, jobCreated, now string, expectedScore float64) {
t.Run(name, func(t *testing.T) {
s := TaskScheduler{}
job := types.Job{
Created: rfc3339(t, jobCreated),
Priority: specs.DEFAULT_JOB_SPEC_PRIORITY,
}
tc := asTryJob(TaskCandidate{
Jobs: []*types.Job{&job},
})
s.scoreCandidate(ctx, &tc, rfc3339(t, now), timeDoesNotMatter, nil)
assert.InDelta(t, expectedScore, tc.Score, 0.0001)
})
}
test("no waiting", "2021-10-01T15:00:00Z", "2021-10-01T15:00:00Z", 5)
test("10 min waiting", "2021-10-01T15:00:00Z", "2021-10-01T15:10:00Z", 5.08333)
test("30 min waiting", "2021-10-01T08:00:00Z", "2021-10-01T08:30:00Z", 5.25)
test("2 hours waiting", "2021-10-01T13:00:00Z", "2021-10-01T15:00:00Z", 6)
}
func TestScoreCandidate_TryJob_PrioritizedLowerByNumRetries(t *testing.T) {
ctx := context.Background()
test := func(name string, numRetries int, expectedScore float64) {
t.Run(name, func(t *testing.T) {
s := TaskScheduler{}
ts := rfc3339(t, "2021-10-01T15:00:00Z") // fixed time indicating no waiting
job := types.Job{
Created: ts,
Priority: specs.DEFAULT_JOB_SPEC_PRIORITY,
}
tc := asTryJob(TaskCandidate{
Jobs: []*types.Job{&job},
Attempt: numRetries,
})
s.scoreCandidate(ctx, &tc, ts, timeDoesNotMatter, nil)
assert.InDelta(t, expectedScore, tc.Score, 0.0001)
})
}
test("first try", 0, 5)
test("second try", 1, 3.75)
test("third try", 2, 2.8125)
test("tenth try", 9, 0.3754)
}
func TestScoreCandidate_TryJob_MultipleJobPrioritiesImpactScore(t *testing.T) {
ctx := context.Background()
test := func(name string, expectedScore float64, jobPriorities ...float64) {
t.Run(name, func(t *testing.T) {
s := TaskScheduler{}
ts := rfc3339(t, "2021-10-01T15:00:00Z") // fixed time indicating no waiting
tc := asTryJob(TaskCandidate{})
// For each job priority passed in, create a job with that priority that's a part
// of this task candidate.
for _, jp := range jobPriorities {
tc.Jobs = append(tc.Jobs, &types.Job{
Created: ts,
Priority: jp,
})
}
s.scoreCandidate(ctx, &tc, ts, timeDoesNotMatter, nil)
assert.InDelta(t, expectedScore, tc.Score, 0.0001)
})
}
test("one job, default priority", 5, specs.DEFAULT_JOB_SPEC_PRIORITY)
test("one job, higher priority", 7.5, 0.75)
test("one job, highest priority", 10, 1.0)
test("one job, lower priority", 2.5, 0.25)
test("one job, lowest priority", 0.001, 0.0001)
test("out of range priority results in default", 5, 1234567)
test("zero priority results in default", 5, 0)
// We score tasks higher if they have more jobs.
test("two jobs, medium priority", 7.5, 0.5, 0.5)
test("three jobs, medium priority", 8.75, 0.5, 0.5, 0.5)
test("four jobs, medium priority", 9.375, 0.5, 0.5, 0.5, 0.5)
test("four jobs, mixed priorities", 9.811, 0.1, 0.7, 0.3, 0.9)
// Double-check the math for the mixed priorities test
assert.InDelta(t, .9811, 1-(1-0.1)*(1-0.7)*(1-0.3)*(1-0.9), 0.0001)
}
func TestScoreCandidate_TryJob_OldestJobOnlyUsedForWaitTime(t *testing.T) {
ctx := context.Background()
cycleStart := rfc3339(t, "2021-10-01T15:00:00Z")
test := func(name string, expectedScore float64, jobCreated ...string) {
t.Run(name, func(t *testing.T) {
s := TaskScheduler{}
tc := asTryJob(TaskCandidate{})
for _, ts := range jobCreated {
tc.Jobs = append(tc.Jobs, &types.Job{
Created: rfc3339(t, ts),
Priority: specs.DEFAULT_JOB_SPEC_PRIORITY,
})
}
s.scoreCandidate(ctx, &tc, cycleStart, timeDoesNotMatter, nil)
assert.InDelta(t, expectedScore, tc.Score, 0.0001)
})
}
// These should all be the same because the first job (which is presumed to be the oldest)
// is the time that is used.
test("two jobs, both waited 2 hours", 9, "2021-10-01T13:00:00Z", "2021-10-01T13:00:00Z")
test("two jobs, only one waited 2 hours", 9, "2021-10-01T13:00:00Z", "2021-10-01T14:55:00Z")
}
func TestScoreCandidate_ForcedJob_PrioritizedHigherByWaitTime(t *testing.T) {
ctx := context.Background()
test := func(name, jobCreated, now string, expectedScore float64) {
t.Run(name, func(t *testing.T) {
s := TaskScheduler{}
job := types.Job{
Created: rfc3339(t, jobCreated),
Priority: specs.DEFAULT_JOB_SPEC_PRIORITY,
}
tc := asForcedJob(TaskCandidate{
Jobs: []*types.Job{&job},
})
s.scoreCandidate(ctx, &tc, rfc3339(t, now), timeDoesNotMatter, nil)
assert.InDelta(t, expectedScore, tc.Score, 0.0001)
})
}
test("no waiting", "2021-10-01T15:00:00Z", "2021-10-01T15:00:00Z", 50)
test("10 min waiting", "2021-10-01T15:00:00Z", "2021-10-01T15:10:00Z", 50.08333)
test("30 min waiting", "2021-10-01T08:00:00Z", "2021-10-01T08:30:00Z", 50.25)
test("2 hours waiting", "2021-10-01T13:00:00Z", "2021-10-01T15:00:00Z", 51)
}
func TestScoreCandidate_ForcedJob_MultipleJobPrioritiesImpactScore(t *testing.T) {
ctx := context.Background()
test := func(name string, expectedScore float64, jobPriorities ...float64) {
t.Run(name, func(t *testing.T) {
s := TaskScheduler{}
ts := rfc3339(t, "2021-10-01T15:00:00Z") // fixed time indicating no waiting
tc := asForcedJob(TaskCandidate{})
// For each job priority passed in, create a job with that priority that's a part
// of this task candidate.
for _, jp := range jobPriorities {
tc.Jobs = append(tc.Jobs, &types.Job{
Created: ts,
Priority: jp,
})
}
s.scoreCandidate(ctx, &tc, ts, timeDoesNotMatter, nil)
assert.InDelta(t, expectedScore, tc.Score, 0.0001)
})
}
test("one job, default priority", 50, specs.DEFAULT_JOB_SPEC_PRIORITY)
test("one job, higher priority", 75, 0.75)
test("one job, highest priority", 100, 1.0)
test("one job, lower priority", 25, 0.25)
test("one job, lowest priority", 0.01, 0.0001)
test("out of range priority results in default", 50, 1234567)
test("zero priority results in default", 50, 0)
// We score tasks higher if they have more jobs.
test("two jobs, medium priority", 75, 0.5, 0.5)
test("three jobs, medium priority", 87.5, 0.5, 0.5, 0.5)
test("four jobs, medium priority", 93.75, 0.5, 0.5, 0.5, 0.5)
test("four jobs, mixed priorities", 98.11, 0.1, 0.7, 0.3, 0.9)
// Double-check the math for the mixed priorities test
assert.InDelta(t, .9811, 1-(1-0.1)*(1-0.7)*(1-0.3)*(1-0.9), 0.0001)
}
func TestScoreCandidate_ForcedJob_OldestJobOnlyUsedForWaitTime(t *testing.T) {
ctx := context.Background()
cycleStart := rfc3339(t, "2021-10-01T15:00:00Z")
test := func(name string, expectedScore float64, jobCreated ...string) {
t.Run(name, func(t *testing.T) {
s := TaskScheduler{}
tc := asForcedJob(TaskCandidate{})
for _, ts := range jobCreated {
tc.Jobs = append(tc.Jobs, &types.Job{
Created: rfc3339(t, ts),
Priority: specs.DEFAULT_JOB_SPEC_PRIORITY,
})
}
s.scoreCandidate(ctx, &tc, cycleStart, timeDoesNotMatter, nil)
assert.InDelta(t, expectedScore, tc.Score, 0.0001)
})
}
// These should all be the same because the first job (which is presumed to be the oldest)
// is the time that is used.
test("two jobs, both waited 2 hours", 76.5, "2021-10-01T13:00:00Z", "2021-10-01T13:00:00Z")
test("two jobs, only one waited 2 hours", 76.5, "2021-10-01T13:00:00Z", "2021-10-01T14:55:00Z")
}
func TestComputeBlamelist_NoExistingTests(t *testing.T) {
ctx := context.Background()
const repoName = "some_repo"
const taskName = "Test-Foo-Bar"
tcg := tcc_mocks.TasksAlwaysDefined(taskName)
mg, repo := newMemRepo(t)
firstCommit := mg.Commit("a")
secondCommit := mg.Commit("b", firstCommit)
thirdCommit := mg.Commit("c", secondCommit)
mtc := &cache_mocks.TaskCache{}
mtc.On("GetTaskForCommit", repoName, mock.Anything, taskName).Return(nil, nil)
blamelistNoTask := func(name string, revision string, expectedBlamelist []string) {
t.Run(name, func(t *testing.T) {
blamedCommits, stealFromTask, err := ComputeBlamelist(ctx, mtc, repo, taskName, repoName, repo.Get(revision), commitsBuffer, tcg, window_mocks.AllInclusiveWindow())
require.NoError(t, err)
assert.Equal(t, expectedBlamelist, blamedCommits)
assert.Nil(t, stealFromTask)
})
}
// We expect this commit and all following commits to be blamed
blamelistNoTask("first commit", firstCommit, []string{firstCommit})
blamelistNoTask("second commit", secondCommit, []string{secondCommit, firstCommit})
blamelistNoTask("third commit", thirdCommit, []string{thirdCommit, secondCommit, firstCommit})
}
func TestComputeBlamelist_FirstCommitTested(t *testing.T) {
ctx := context.Background()
const repoName = "some_repo"
const taskName = "Test-Foo-Bar"
tcg := tcc_mocks.TasksAlwaysDefined(taskName)
mg, repo := newMemRepo(t)
firstCommit := mg.Commit("a")
secondCommit := mg.Commit("b", firstCommit)
thirdCommit := mg.Commit("c", secondCommit)
firstCommitTask := newTask("task-1", firstCommit)
mtc := &cache_mocks.TaskCache{}
mtc.On("GetTaskForCommit", repoName, firstCommit, taskName).Return(firstCommitTask, nil)
mtc.On("GetTaskForCommit", repoName, mock.Anything, taskName).Return(nil, nil)
blamelistNoTask := func(name string, revision string, expectedBlamelist []string) {
t.Run(name, func(t *testing.T) {
blamedCommits, stealFromTask, err := ComputeBlamelist(ctx, mtc, repo, taskName, repoName, repo.Get(revision), commitsBuffer, tcg, window_mocks.AllInclusiveWindow())
require.NoError(t, err)
assert.Equal(t, expectedBlamelist, blamedCommits)
assert.Nil(t, stealFromTask)
})
}
// For commits after the task, the blamelist should extend back, but not include that commit.
blamelistNoTask("second commit", secondCommit, []string{secondCommit})
blamelistNoTask("third commit", thirdCommit, []string{thirdCommit, secondCommit})
// Calculating a blame for a commit which already ran a task is considered a retry. As such,
// the entire blamelist is returned.
t.Run("first commit (retry)", func(t *testing.T) {
blamedCommits, stealFromTask, err := ComputeBlamelist(ctx, mtc, repo, taskName, repoName, repo.Get(firstCommit), commitsBuffer, tcg, window_mocks.AllInclusiveWindow())
require.NoError(t, err)
assert.Equal(t, []string{firstCommit}, blamedCommits)
assert.Equal(t, firstCommitTask, stealFromTask)
})
}
func TestComputeBlamelist_LastCommitTested_FollowingCommitsBlamed(t *testing.T) {
ctx := context.Background()
const repoName = "some_repo"
const taskName = "Test-Foo-Bar"
tcg := tcc_mocks.TasksAlwaysDefined(taskName)
mg, repo := newMemRepo(t)
firstCommit := mg.Commit("a")
secondCommit := mg.Commit("b", firstCommit)
thirdCommit := mg.Commit("c", secondCommit)
thirdCommitTask := newTask("task-1", thirdCommit, secondCommit, firstCommit)
mtc := &cache_mocks.TaskCache{}
// By returning this task for all commits, we are saying that every commit is covered by
// the task that ran on the last commit.
mtc.On("GetTaskForCommit", repoName, mock.Anything, taskName).Return(thirdCommitTask, nil)
blamelistAndTask := func(name string, revision string, expectedBlamelist []string, expectedTask *types.Task) {
t.Run(name, func(t *testing.T) {
blamedCommits, stealFromTask, err := ComputeBlamelist(ctx, mtc, repo, taskName, repoName, repo.Get(revision), commitsBuffer, tcg, window_mocks.AllInclusiveWindow())
require.NoError(t, err)
assert.Equal(t, expectedBlamelist, blamedCommits)
assert.Equal(t, expectedTask, stealFromTask)
})
}
blamelistAndTask("first commit", firstCommit, []string{firstCommit}, thirdCommitTask)
blamelistAndTask("second commit", secondCommit, []string{secondCommit, firstCommit}, thirdCommitTask)
// Calculating a blame for a commit which already ran a task is considered a retry. As such,
// the entire blamelist is returned.
blamelistAndTask("third commit (retry)", thirdCommit, []string{thirdCommit, secondCommit, firstCommit}, thirdCommitTask)
}
func TestComputeBlamelist_BlamelistTooLong_UseProvidedCommit(t *testing.T) {
ctx := context.Background()
const repoName = "some_repo"
const taskName = "Test-Foo-Bar"
tcg := tcc_mocks.TasksAlwaysDefined(taskName)
// Pretend we have 1000 commits in a row. We make sure their hashes are unique but predictable
// by using the integer values from 0 to 999 prefixed with enough zeros to make them 40 digits
// (the length of a real-world git SHA1 hash).
require.Greater(t, 1000, MAX_BLAMELIST_COMMITS)
mg, repo := newMemRepo(t)
mg.Commit("0")
for i := 1; i < 1000; i++ {
h := fmt.Sprintf("Commit_%d", i)
mg.Commit(h)
}
lastCommit := repo.Get(repo.Branches()[0])
require.Equal(t, "Commit_999", lastCommit.Subject)
mtc := &cache_mocks.TaskCache{}
mtc.On("GetTaskForCommit", repoName, mock.Anything, taskName).Return(nil, nil)
blamedCommits, stealFromTask, err := ComputeBlamelist(ctx, mtc, repo, taskName, repoName, lastCommit, commitsBuffer, tcg, window_mocks.AllInclusiveWindow())
require.NoError(t, err)
assert.Equal(t, []string{lastCommit.Hash}, blamedCommits)
assert.Nil(t, stealFromTask)
}
func TestComputeBlamelist_BranchInHistory_NonOverlappingCoverage(t *testing.T) {
ctx := context.Background()
const repoName = "some_repo"
const taskName = "Test-Foo-Bar"
tcg := tcc_mocks.TasksAlwaysDefined(taskName)
// This represents a git history with a branch and merge like this.
// The asterisks represent at which commits a task has run. Only one of the branches
// has a test, so commit 2 is not "double covered".
// 0
// 1*
// 2
// 3a 3b
// 4a* 4b
// 5a |
// 6
mg, repo := newMemRepo(t)
zero := mg.Commit("0")
one := mg.Commit("1", zero)
two := mg.Commit("2", one)
threeA := mg.Commit("3a", two)
threeB := mg.Commit("3b", two)
fourA := mg.Commit("4a", threeA)
fourB := mg.Commit("4b", threeB)
fiveA := mg.Commit("5a", fourA)
six := mg.Commit("6", fourB, fiveA)
commitOneTask := newTask("task-1", one, zero)
commitFourATask := newTask("task-2", fourA, threeA, two)
mtc := &cache_mocks.TaskCache{}
mtc.On("GetTaskForCommit", repoName, zero, taskName).Return(commitOneTask, nil)
mtc.On("GetTaskForCommit", repoName, one, taskName).Return(commitOneTask, nil)
mtc.On("GetTaskForCommit", repoName, two, taskName).Return(commitFourATask, nil)
mtc.On("GetTaskForCommit", repoName, threeA, taskName).Return(commitFourATask, nil)
mtc.On("GetTaskForCommit", repoName, fourA, taskName).Return(commitFourATask, nil)
mtc.On("GetTaskForCommit", repoName, mock.Anything, taskName).Return(nil, nil)
blamelistAndTask := func(name string, revision string, expectedBlamelist []string, expectedTask *types.Task) {
t.Run(name, func(t *testing.T) {
blamedCommits, stealFromTask, err := ComputeBlamelist(ctx, mtc, repo, taskName, repoName, repo.Get(revision), commitsBuffer, tcg, window_mocks.AllInclusiveWindow())
require.NoError(t, err)
assert.Equal(t, expectedBlamelist, blamedCommits)
assert.Equal(t, expectedTask, stealFromTask)
})
}
blamelistAndTask("commit zero", zero, []string{zero}, commitOneTask)
blamelistAndTask("commit one", one, []string{one, zero}, commitOneTask)
blamelistAndTask("commit two", two, []string{two}, commitFourATask)
blamelistAndTask("commit threeA", threeA, []string{threeA, two}, commitFourATask)
blamelistAndTask("commit threeB", threeB, []string{threeB}, nil)
blamelistAndTask("commit fourA", fourA, []string{fourA, threeA, two}, commitFourATask)
blamelistAndTask("commit fourB", fourB, []string{fourB, threeB}, nil)
blamelistAndTask("commit fiveA", fiveA, []string{fiveA}, nil)
blamelistAndTask("commit six", six, []string{six, fourB, threeB, fiveA}, nil)
}
var (
// Use of timeDoesNotMatter indicates that the time does not affect the outputs.
timeDoesNotMatter = time.Time{}
commitsBuffer = make([]*repograph.Commit, 0, MAX_BLAMELIST_COMMITS)
)
func rfc3339(t *testing.T, n string) time.Time {
ts, err := time.Parse(time.RFC3339, n)
require.NoError(t, err)
return ts
}
func asTryJob(c TaskCandidate) TaskCandidate {
c.Issue = "nonempty"
c.Patchset = "nonempty"
c.Server = "nonempty"
if !c.IsTryJob() {
panic("Should be a tryjob task candidate now")
}
return c
}
func asForcedJob(c TaskCandidate) TaskCandidate {
c.ForcedJobId = "nonempty"
if !c.IsForceRun() {
panic("Should be a tryjob task candidate now")
}
return c
}
func newTask(id string, ranAt string, alsoCovered ...string) *types.Task {
nt := types.Task{Id: id}
nt.Revision = ranAt
nt.Commits = append(nt.Commits, ranAt)
for _, c := range alsoCovered {
nt.Commits = append(nt.Commits, c)
}
return &nt
}
// newMemRepo is a convenience function which creates a MemGit backed by an
// in-memory GitStore and builds a repograph.Graph from it. MemGit automatically
// calls Graph.Update() whenever it is mutated.
func newMemRepo(t sktest.TestingT) (*mem_git.MemGit, *repograph.Graph) {
gs := mem_gitstore.New()
gb := mem_git.New(t, gs)
ctx := context.Background()
ri, err := gitstore.NewGitStoreRepoImpl(ctx, gs)
require.NoError(t, err)
repo, err := repograph.NewWithRepoImpl(ctx, ri)
require.NoError(t, err)
gb.AddUpdater(repo)
return gb, repo
}