mirror of https://github.com/coder/coder.git
751 lines
25 KiB
Go
751 lines
25 KiB
Go
package provisionerdserver_test
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/sqlc-dev/pqtype"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"go.uber.org/goleak"
|
|
"golang.org/x/exp/slices"
|
|
|
|
"cdr.dev/slog"
|
|
"cdr.dev/slog/sloggers/slogtest"
|
|
"github.com/coder/coder/v2/coderd/database"
|
|
"github.com/coder/coder/v2/coderd/database/dbmem"
|
|
"github.com/coder/coder/v2/coderd/database/dbtestutil"
|
|
"github.com/coder/coder/v2/coderd/database/dbtime"
|
|
"github.com/coder/coder/v2/coderd/database/provisionerjobs"
|
|
"github.com/coder/coder/v2/coderd/database/pubsub"
|
|
"github.com/coder/coder/v2/coderd/provisionerdserver"
|
|
"github.com/coder/coder/v2/testutil"
|
|
)
|
|
|
|
func TestMain(m *testing.M) {
|
|
goleak.VerifyTestMain(m)
|
|
}
|
|
|
|
// TestAcquirer_Store tests that a database.Store is accepted as a provisionerdserver.AcquirerStore
|
|
func TestAcquirer_Store(t *testing.T) {
|
|
t.Parallel()
|
|
db := dbmem.New()
|
|
ps := pubsub.NewInMemory()
|
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
|
|
defer cancel()
|
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
_ = provisionerdserver.NewAcquirer(ctx, logger.Named("acquirer"), db, ps)
|
|
}
|
|
|
|
func TestAcquirer_Single(t *testing.T) {
|
|
t.Parallel()
|
|
fs := newFakeOrderedStore()
|
|
ps := pubsub.NewInMemory()
|
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
|
|
defer cancel()
|
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
uut := provisionerdserver.NewAcquirer(ctx, logger.Named("acquirer"), fs, ps)
|
|
|
|
orgID := uuid.New()
|
|
workerID := uuid.New()
|
|
pt := []database.ProvisionerType{database.ProvisionerTypeEcho}
|
|
tags := provisionerdserver.Tags{
|
|
"environment": "on-prem",
|
|
}
|
|
acquiree := newTestAcquiree(t, orgID, workerID, pt, tags)
|
|
jobID := uuid.New()
|
|
err := fs.sendCtx(ctx, database.ProvisionerJob{ID: jobID}, nil)
|
|
require.NoError(t, err)
|
|
acquiree.startAcquire(ctx, uut)
|
|
job := acquiree.success(ctx)
|
|
require.Equal(t, jobID, job.ID)
|
|
require.Len(t, fs.params, 1)
|
|
require.Equal(t, workerID, fs.params[0].WorkerID.UUID)
|
|
}
|
|
|
|
// TestAcquirer_MultipleSameDomain tests multiple acquirees with the same provisioners and tags
|
|
func TestAcquirer_MultipleSameDomain(t *testing.T) {
|
|
t.Parallel()
|
|
fs := newFakeOrderedStore()
|
|
ps := pubsub.NewInMemory()
|
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
|
|
defer cancel()
|
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
uut := provisionerdserver.NewAcquirer(ctx, logger.Named("acquirer"), fs, ps)
|
|
|
|
acquirees := make([]*testAcquiree, 0, 10)
|
|
jobIDs := make(map[uuid.UUID]bool)
|
|
workerIDs := make(map[uuid.UUID]bool)
|
|
orgID := uuid.New()
|
|
pt := []database.ProvisionerType{database.ProvisionerTypeEcho}
|
|
tags := provisionerdserver.Tags{
|
|
"environment": "on-prem",
|
|
}
|
|
for i := 0; i < 10; i++ {
|
|
wID := uuid.New()
|
|
workerIDs[wID] = true
|
|
a := newTestAcquiree(t, orgID, wID, pt, tags)
|
|
acquirees = append(acquirees, a)
|
|
a.startAcquire(ctx, uut)
|
|
}
|
|
for i := 0; i < 10; i++ {
|
|
jobID := uuid.New()
|
|
jobIDs[jobID] = true
|
|
err := fs.sendCtx(ctx, database.ProvisionerJob{ID: jobID}, nil)
|
|
require.NoError(t, err)
|
|
}
|
|
gotJobIDs := make(map[uuid.UUID]bool)
|
|
for i := 0; i < 10; i++ {
|
|
j := acquirees[i].success(ctx)
|
|
gotJobIDs[j.ID] = true
|
|
}
|
|
require.Equal(t, jobIDs, gotJobIDs)
|
|
require.Len(t, fs.overlaps, 0)
|
|
gotWorkerCalls := make(map[uuid.UUID]bool)
|
|
for _, params := range fs.params {
|
|
gotWorkerCalls[params.WorkerID.UUID] = true
|
|
}
|
|
require.Equal(t, workerIDs, gotWorkerCalls)
|
|
}
|
|
|
|
// TestAcquirer_WaitsOnNoJobs tests that after a call that returns no jobs, Acquirer waits for a new
|
|
// job posting before retrying
|
|
func TestAcquirer_WaitsOnNoJobs(t *testing.T) {
|
|
t.Parallel()
|
|
fs := newFakeOrderedStore()
|
|
ps := pubsub.NewInMemory()
|
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
|
|
defer cancel()
|
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
uut := provisionerdserver.NewAcquirer(ctx, logger.Named("acquirer"), fs, ps)
|
|
|
|
orgID := uuid.New()
|
|
workerID := uuid.New()
|
|
pt := []database.ProvisionerType{database.ProvisionerTypeEcho}
|
|
tags := provisionerdserver.Tags{
|
|
"environment": "on-prem",
|
|
}
|
|
acquiree := newTestAcquiree(t, orgID, workerID, pt, tags)
|
|
jobID := uuid.New()
|
|
err := fs.sendCtx(ctx, database.ProvisionerJob{}, sql.ErrNoRows)
|
|
require.NoError(t, err)
|
|
err = fs.sendCtx(ctx, database.ProvisionerJob{ID: jobID}, nil)
|
|
require.NoError(t, err)
|
|
acquiree.startAcquire(ctx, uut)
|
|
require.Eventually(t, func() bool {
|
|
fs.mu.Lock()
|
|
defer fs.mu.Unlock()
|
|
return len(fs.params) == 1
|
|
}, testutil.WaitShort, testutil.IntervalFast)
|
|
acquiree.requireBlocked()
|
|
|
|
// First send in some with incompatible tags & types
|
|
postJob(t, ps, database.ProvisionerTypeEcho, provisionerdserver.Tags{
|
|
"cool": "tapes",
|
|
"strong": "bad",
|
|
})
|
|
postJob(t, ps, database.ProvisionerTypeEcho, provisionerdserver.Tags{
|
|
"environment": "fighters",
|
|
})
|
|
postJob(t, ps, database.ProvisionerTypeTerraform, provisionerdserver.Tags{
|
|
"environment": "on-prem",
|
|
})
|
|
acquiree.requireBlocked()
|
|
|
|
// compatible tags
|
|
postJob(t, ps, database.ProvisionerTypeEcho, provisionerdserver.Tags{})
|
|
job := acquiree.success(ctx)
|
|
require.Equal(t, jobID, job.ID)
|
|
}
|
|
|
|
// TestAcquirer_RetriesPending tests that if we get a job posting while a db call is in progress
|
|
// we retry to acquire a job immediately, even if the first call returned no jobs. We want this
|
|
// behavior since the query that found no jobs could have resolved before the job was posted, but
|
|
// the query result could reach us later than the posting over the pubsub.
|
|
func TestAcquirer_RetriesPending(t *testing.T) {
|
|
t.Parallel()
|
|
fs := newFakeOrderedStore()
|
|
ps := pubsub.NewInMemory()
|
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
|
|
defer cancel()
|
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
uut := provisionerdserver.NewAcquirer(ctx, logger.Named("acquirer"), fs, ps)
|
|
|
|
orgID := uuid.New()
|
|
workerID := uuid.New()
|
|
pt := []database.ProvisionerType{database.ProvisionerTypeEcho}
|
|
tags := provisionerdserver.Tags{
|
|
"environment": "on-prem",
|
|
}
|
|
acquiree := newTestAcquiree(t, orgID, workerID, pt, tags)
|
|
jobID := uuid.New()
|
|
|
|
acquiree.startAcquire(ctx, uut)
|
|
require.Eventually(t, func() bool {
|
|
fs.mu.Lock()
|
|
defer fs.mu.Unlock()
|
|
return len(fs.params) == 1
|
|
}, testutil.WaitShort, testutil.IntervalFast)
|
|
|
|
// First call to DB is in progress. Send in posting
|
|
postJob(t, ps, database.ProvisionerTypeEcho, provisionerdserver.Tags{})
|
|
// there is a race between the posting being processed and the DB call
|
|
// returning. In either case we should retry, but we're trying to hit the
|
|
// case where the posting is processed first, so sleep a little bit to give
|
|
// it a chance.
|
|
time.Sleep(testutil.IntervalMedium)
|
|
|
|
// Now, when first DB call returns ErrNoRows we retry.
|
|
err := fs.sendCtx(ctx, database.ProvisionerJob{}, sql.ErrNoRows)
|
|
require.NoError(t, err)
|
|
err = fs.sendCtx(ctx, database.ProvisionerJob{ID: jobID}, nil)
|
|
require.NoError(t, err)
|
|
|
|
job := acquiree.success(ctx)
|
|
require.Equal(t, jobID, job.ID)
|
|
}
|
|
|
|
// TestAcquirer_DifferentDomains tests that acquirees with different tags don't block each other
|
|
func TestAcquirer_DifferentDomains(t *testing.T) {
|
|
t.Parallel()
|
|
fs := newFakeTaggedStore(t)
|
|
ps := pubsub.NewInMemory()
|
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
|
|
defer cancel()
|
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
|
|
orgID := uuid.New()
|
|
pt := []database.ProvisionerType{database.ProvisionerTypeEcho}
|
|
worker0 := uuid.New()
|
|
tags0 := provisionerdserver.Tags{
|
|
"worker": "0",
|
|
}
|
|
acquiree0 := newTestAcquiree(t, orgID, worker0, pt, tags0)
|
|
worker1 := uuid.New()
|
|
tags1 := provisionerdserver.Tags{
|
|
"worker": "1",
|
|
}
|
|
acquiree1 := newTestAcquiree(t, orgID, worker1, pt, tags1)
|
|
jobID := uuid.New()
|
|
fs.jobs = []database.ProvisionerJob{
|
|
{ID: jobID, Provisioner: database.ProvisionerTypeEcho, Tags: database.StringMap{"worker": "1"}},
|
|
}
|
|
|
|
uut := provisionerdserver.NewAcquirer(ctx, logger.Named("acquirer"), fs, ps)
|
|
|
|
ctx0, cancel0 := context.WithCancel(ctx)
|
|
defer cancel0()
|
|
acquiree0.startAcquire(ctx0, uut)
|
|
select {
|
|
case params := <-fs.params:
|
|
require.Equal(t, worker0, params.WorkerID.UUID)
|
|
case <-ctx.Done():
|
|
t.Fatal("timed out waiting for call to database from worker0")
|
|
}
|
|
acquiree0.requireBlocked()
|
|
|
|
// worker1 should not be blocked by worker0, as they are different tags
|
|
acquiree1.startAcquire(ctx, uut)
|
|
job := acquiree1.success(ctx)
|
|
require.Equal(t, jobID, job.ID)
|
|
|
|
cancel0()
|
|
acquiree0.requireCanceled(ctx)
|
|
}
|
|
|
|
func TestAcquirer_BackupPoll(t *testing.T) {
|
|
t.Parallel()
|
|
fs := newFakeOrderedStore()
|
|
ps := pubsub.NewInMemory()
|
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
|
|
defer cancel()
|
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
uut := provisionerdserver.NewAcquirer(
|
|
ctx, logger.Named("acquirer"), fs, ps,
|
|
provisionerdserver.TestingBackupPollDuration(testutil.IntervalMedium),
|
|
)
|
|
|
|
workerID := uuid.New()
|
|
orgID := uuid.New()
|
|
pt := []database.ProvisionerType{database.ProvisionerTypeEcho}
|
|
tags := provisionerdserver.Tags{
|
|
"environment": "on-prem",
|
|
}
|
|
acquiree := newTestAcquiree(t, orgID, workerID, pt, tags)
|
|
jobID := uuid.New()
|
|
err := fs.sendCtx(ctx, database.ProvisionerJob{}, sql.ErrNoRows)
|
|
require.NoError(t, err)
|
|
err = fs.sendCtx(ctx, database.ProvisionerJob{ID: jobID}, nil)
|
|
require.NoError(t, err)
|
|
acquiree.startAcquire(ctx, uut)
|
|
job := acquiree.success(ctx)
|
|
require.Equal(t, jobID, job.ID)
|
|
}
|
|
|
|
// TestAcquirer_UnblockOnCancel tests that a canceled call doesn't block a call
|
|
// from the same domain.
|
|
func TestAcquirer_UnblockOnCancel(t *testing.T) {
|
|
t.Parallel()
|
|
fs := newFakeOrderedStore()
|
|
ps := pubsub.NewInMemory()
|
|
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
|
|
defer cancel()
|
|
logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
|
|
pt := []database.ProvisionerType{database.ProvisionerTypeEcho}
|
|
orgID := uuid.New()
|
|
worker0 := uuid.New()
|
|
tags := provisionerdserver.Tags{
|
|
"environment": "on-prem",
|
|
}
|
|
acquiree0 := newTestAcquiree(t, orgID, worker0, pt, tags)
|
|
worker1 := uuid.New()
|
|
acquiree1 := newTestAcquiree(t, orgID, worker1, pt, tags)
|
|
jobID := uuid.New()
|
|
|
|
uut := provisionerdserver.NewAcquirer(ctx, logger.Named("acquirer"), fs, ps)
|
|
|
|
// queue up 2 responses --- we may not need both, since acquiree0 will
|
|
// usually cancel before calling, but cancel is async, so it might call.
|
|
for i := 0; i < 2; i++ {
|
|
err := fs.sendCtx(ctx, database.ProvisionerJob{ID: jobID}, nil)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
ctx0, cancel0 := context.WithCancel(ctx)
|
|
cancel0()
|
|
acquiree0.startAcquire(ctx0, uut)
|
|
acquiree1.startAcquire(ctx, uut)
|
|
job := acquiree1.success(ctx)
|
|
require.Equal(t, jobID, job.ID)
|
|
}
|
|
|
|
func TestAcquirer_MatchTags(t *testing.T) {
|
|
t.Parallel()
|
|
if testing.Short() {
|
|
t.Skip("skipping this test due to -short")
|
|
}
|
|
|
|
testCases := []struct {
|
|
name string
|
|
provisionerJobTags map[string]string
|
|
|
|
acquireJobTags map[string]string
|
|
unmatchedOrg bool // acquire will use a random org id
|
|
expectAcquire bool
|
|
}{
|
|
{
|
|
name: "untagged provisioner and untagged job",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": ""},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": ""},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "tagged provisioner and tagged job",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem"},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "double-tagged provisioner and tagged job",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem", "datacenter": "chicago"},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "double-tagged provisioner and double-tagged job",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem", "datacenter": "chicago"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem", "datacenter": "chicago"},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner and user-scoped job",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa"},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner with tags and user-scoped job",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem"},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner with tags and user-scoped job with tags",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem"},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner with multiple tags and user-scoped job with tags",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem", "datacenter": "chicago"},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner with multiple tags and user-scoped job with multiple tags",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem", "datacenter": "chicago"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem", "datacenter": "chicago"},
|
|
expectAcquire: true,
|
|
},
|
|
{
|
|
name: "untagged provisioner and tagged job",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": ""},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "tagged provisioner and untagged job",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": ""},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem"},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "tagged provisioner and double-tagged job",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem", "datacenter": "chicago"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem"},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "double-tagged provisioner and double-tagged job with differing tags",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem", "datacenter": "chicago"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem", "datacenter": "new_york"},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner and untagged job",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": ""},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa"},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner and different user-scoped job",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "bbb"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa"},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "org-scoped provisioner and user-scoped job",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": ""},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner and org-scoped job with tags",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": ""},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner and user-scoped job with tags",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa"},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner with tags and user-scoped job with multiple tags",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem", "datacenter": "chicago"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem"},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "user-scoped provisioner with tags and user-scoped job with differing tags",
|
|
provisionerJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem", "datacenter": "new_york"},
|
|
acquireJobTags: map[string]string{"scope": "user", "owner": "aaa", "environment": "on-prem", "datacenter": "chicago"},
|
|
expectAcquire: false,
|
|
},
|
|
{
|
|
name: "matching tags with unmatched org",
|
|
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem"},
|
|
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "environment": "on-prem"},
|
|
expectAcquire: false,
|
|
unmatchedOrg: true,
|
|
},
|
|
}
|
|
for _, tt := range testCases {
|
|
tt := tt
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
ctx := testutil.Context(t, testutil.WaitShort)
|
|
// NOTE: explicitly not using fake store for this test.
|
|
db, ps := dbtestutil.NewDB(t)
|
|
log := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
|
|
org, err := db.InsertOrganization(ctx, database.InsertOrganizationParams{
|
|
ID: uuid.New(),
|
|
Name: "test org",
|
|
Description: "the organization of testing",
|
|
CreatedAt: dbtime.Now(),
|
|
UpdatedAt: dbtime.Now(),
|
|
})
|
|
require.NoError(t, err)
|
|
pj, err := db.InsertProvisionerJob(ctx, database.InsertProvisionerJobParams{
|
|
ID: uuid.New(),
|
|
CreatedAt: dbtime.Now(),
|
|
UpdatedAt: dbtime.Now(),
|
|
OrganizationID: org.ID,
|
|
InitiatorID: uuid.New(),
|
|
Provisioner: database.ProvisionerTypeEcho,
|
|
StorageMethod: database.ProvisionerStorageMethodFile,
|
|
FileID: uuid.New(),
|
|
Type: database.ProvisionerJobTypeWorkspaceBuild,
|
|
Input: []byte("{}"),
|
|
Tags: tt.provisionerJobTags,
|
|
TraceMetadata: pqtype.NullRawMessage{},
|
|
})
|
|
require.NoError(t, err)
|
|
ptypes := []database.ProvisionerType{database.ProvisionerTypeEcho}
|
|
acq := provisionerdserver.NewAcquirer(ctx, log, db, ps)
|
|
|
|
acquireOrgID := org.ID
|
|
if tt.unmatchedOrg {
|
|
acquireOrgID = uuid.New()
|
|
}
|
|
aj, err := acq.AcquireJob(ctx, acquireOrgID, uuid.New(), ptypes, tt.acquireJobTags)
|
|
if tt.expectAcquire {
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, pj.ID, aj.ID)
|
|
} else {
|
|
assert.Empty(t, aj, "should not have acquired job")
|
|
assert.ErrorIs(t, err, context.DeadlineExceeded, "should have timed out")
|
|
}
|
|
})
|
|
}
|
|
|
|
t.Run("GenTable", func(t *testing.T) {
|
|
t.Parallel()
|
|
// Generate a table that can be copy-pasted into docs/admin/provisioners.md
|
|
lines := []string{
|
|
"\n",
|
|
"| Provisioner Tags | Job Tags | Can Run Job? |",
|
|
"|------------------|----------|--------------|",
|
|
}
|
|
// turn the JSON map into k=v for readability
|
|
kvs := func(m map[string]string) string {
|
|
ss := make([]string, 0, len(m))
|
|
// ensure consistent ordering of tags
|
|
for _, k := range []string{"scope", "owner", "environment", "datacenter"} {
|
|
if v, found := m[k]; found {
|
|
ss = append(ss, k+"="+v)
|
|
}
|
|
}
|
|
return strings.Join(ss, " ")
|
|
}
|
|
for _, tt := range testCases {
|
|
acquire := "✅"
|
|
if !tt.expectAcquire {
|
|
acquire = "❌"
|
|
}
|
|
s := fmt.Sprintf("| %s | %s | %s |", kvs(tt.acquireJobTags), kvs(tt.provisionerJobTags), acquire)
|
|
lines = append(lines, s)
|
|
}
|
|
t.Logf("You can paste this into docs/admin/provisioners.md")
|
|
t.Logf(strings.Join(lines, "\n"))
|
|
})
|
|
}
|
|
|
|
func postJob(t *testing.T, ps pubsub.Pubsub, pt database.ProvisionerType, tags provisionerdserver.Tags) {
|
|
t.Helper()
|
|
msg, err := json.Marshal(provisionerjobs.JobPosting{
|
|
ProvisionerType: pt,
|
|
Tags: tags,
|
|
})
|
|
require.NoError(t, err)
|
|
err = ps.Publish(provisionerjobs.EventJobPosted, msg)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
// fakeOrderedStore is a fake store that lets tests send AcquireProvisionerJob
|
|
// results in order over a channel, and tests for overlapped calls.
|
|
type fakeOrderedStore struct {
|
|
jobs chan database.ProvisionerJob
|
|
errors chan error
|
|
|
|
mu sync.Mutex
|
|
params []database.AcquireProvisionerJobParams
|
|
|
|
// inflight and overlaps track whether any calls from workers overlap with
|
|
// one another
|
|
inflight map[uuid.UUID]bool
|
|
overlaps [][]uuid.UUID
|
|
}
|
|
|
|
func newFakeOrderedStore() *fakeOrderedStore {
|
|
return &fakeOrderedStore{
|
|
// buffer the channels so that we can queue up lots of responses to
|
|
// occur nearly simultaneously
|
|
jobs: make(chan database.ProvisionerJob, 100),
|
|
errors: make(chan error, 100),
|
|
inflight: make(map[uuid.UUID]bool),
|
|
}
|
|
}
|
|
|
|
func (s *fakeOrderedStore) AcquireProvisionerJob(
|
|
_ context.Context, params database.AcquireProvisionerJobParams,
|
|
) (
|
|
database.ProvisionerJob, error,
|
|
) {
|
|
s.mu.Lock()
|
|
s.params = append(s.params, params)
|
|
for workerID := range s.inflight {
|
|
s.overlaps = append(s.overlaps, []uuid.UUID{workerID, params.WorkerID.UUID})
|
|
}
|
|
s.inflight[params.WorkerID.UUID] = true
|
|
s.mu.Unlock()
|
|
|
|
job := <-s.jobs
|
|
err := <-s.errors
|
|
|
|
s.mu.Lock()
|
|
delete(s.inflight, params.WorkerID.UUID)
|
|
s.mu.Unlock()
|
|
|
|
return job, err
|
|
}
|
|
|
|
func (s *fakeOrderedStore) sendCtx(ctx context.Context, job database.ProvisionerJob, err error) error {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case s.jobs <- job:
|
|
// OK
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case s.errors <- err:
|
|
// OK
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// fakeTaggedStore is a test store that allows tests to specify which jobs are
|
|
// available, and returns them to callers with the appropriate provisioner type
|
|
// and tags. It doesn't care about the order.
|
|
type fakeTaggedStore struct {
|
|
t *testing.T
|
|
mu sync.Mutex
|
|
jobs []database.ProvisionerJob
|
|
params chan database.AcquireProvisionerJobParams
|
|
}
|
|
|
|
func newFakeTaggedStore(t *testing.T) *fakeTaggedStore {
|
|
return &fakeTaggedStore{
|
|
t: t,
|
|
params: make(chan database.AcquireProvisionerJobParams, 100),
|
|
}
|
|
}
|
|
|
|
func (s *fakeTaggedStore) AcquireProvisionerJob(
|
|
_ context.Context, params database.AcquireProvisionerJobParams,
|
|
) (
|
|
database.ProvisionerJob, error,
|
|
) {
|
|
defer func() { s.params <- params }()
|
|
var tags provisionerdserver.Tags
|
|
err := json.Unmarshal(params.Tags, &tags)
|
|
if !assert.NoError(s.t, err) {
|
|
return database.ProvisionerJob{}, err
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
jobLoop:
|
|
for i, job := range s.jobs {
|
|
if !slices.Contains(params.Types, job.Provisioner) {
|
|
continue
|
|
}
|
|
for k, v := range job.Tags {
|
|
pv, ok := tags[k]
|
|
if !ok {
|
|
continue jobLoop
|
|
}
|
|
if v != pv {
|
|
continue jobLoop
|
|
}
|
|
}
|
|
// found a job!
|
|
s.jobs = append(s.jobs[:i], s.jobs[i+1:]...)
|
|
return job, nil
|
|
}
|
|
return database.ProvisionerJob{}, sql.ErrNoRows
|
|
}
|
|
|
|
// testAcquiree is a helper type that handles asynchronously calling AcquireJob
|
|
// and asserting whether or not it returns, blocks, or is canceled.
|
|
type testAcquiree struct {
|
|
t *testing.T
|
|
orgID uuid.UUID
|
|
workerID uuid.UUID
|
|
pt []database.ProvisionerType
|
|
tags provisionerdserver.Tags
|
|
ec chan error
|
|
jc chan database.ProvisionerJob
|
|
}
|
|
|
|
func newTestAcquiree(t *testing.T, orgID uuid.UUID, workerID uuid.UUID, pt []database.ProvisionerType, tags provisionerdserver.Tags) *testAcquiree {
|
|
return &testAcquiree{
|
|
t: t,
|
|
orgID: orgID,
|
|
workerID: workerID,
|
|
pt: pt,
|
|
tags: tags,
|
|
ec: make(chan error, 1),
|
|
jc: make(chan database.ProvisionerJob, 1),
|
|
}
|
|
}
|
|
|
|
func (a *testAcquiree) startAcquire(ctx context.Context, uut *provisionerdserver.Acquirer) {
|
|
go func() {
|
|
j, e := uut.AcquireJob(ctx, a.orgID, a.workerID, a.pt, a.tags)
|
|
a.ec <- e
|
|
a.jc <- j
|
|
}()
|
|
}
|
|
|
|
func (a *testAcquiree) success(ctx context.Context) database.ProvisionerJob {
|
|
select {
|
|
case <-ctx.Done():
|
|
a.t.Fatal("timeout waiting for AcquireJob error")
|
|
case err := <-a.ec:
|
|
require.NoError(a.t, err)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
a.t.Fatal("timeout waiting for AcquireJob job")
|
|
case job := <-a.jc:
|
|
return job
|
|
}
|
|
// unhittable
|
|
return database.ProvisionerJob{}
|
|
}
|
|
|
|
func (a *testAcquiree) requireBlocked() {
|
|
select {
|
|
case <-a.ec:
|
|
a.t.Fatal("AcquireJob should block")
|
|
default:
|
|
// OK
|
|
}
|
|
}
|
|
|
|
func (a *testAcquiree) requireCanceled(ctx context.Context) {
|
|
select {
|
|
case err := <-a.ec:
|
|
require.ErrorIs(a.t, err, context.Canceled)
|
|
case <-ctx.Done():
|
|
a.t.Fatal("timed out waiting for AcquireJob")
|
|
}
|
|
select {
|
|
case job := <-a.jc:
|
|
require.Equal(a.t, uuid.Nil, job.ID)
|
|
case <-ctx.Done():
|
|
a.t.Fatal("timed out waiting for AcquireJob")
|
|
}
|
|
}
|