Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

feat(coderd/database): support exact tags match in AcquireProvisionerJob query#12244

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Closed
johnstcn wants to merge3 commits intomainfromcj/provisionerd_tag_policy
Closed
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletionscoderd/database/dbfake/dbfake.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -192,8 +192,9 @@ func (b WorkspaceBuildBuilder) Do() WorkspaceResponse {
UUID: uuid.New(),
Valid: true,
},
Types: []database.ProvisionerType{database.ProvisionerTypeEcho},
Tags: []byte(`{"scope": "organization"}`),
Types: []database.ProvisionerType{database.ProvisionerTypeEcho},
Tags: []byte(`{"scope": "organization"}`),
ExactTagMatch: false,
})
require.NoError(b.t, err, "acquire starting job")
if j.ID == job.ID {
Expand Down
9 changes: 5 additions & 4 deletionscoderd/database/dbgen/dbgen.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -417,10 +417,11 @@ func ProvisionerJob(t testing.TB, db database.Store, ps pubsub.Pubsub, orig data
}
if !orig.StartedAt.Time.IsZero() {
job, err = db.AcquireProvisionerJob(genCtx, database.AcquireProvisionerJobParams{
StartedAt: orig.StartedAt,
Types: []database.ProvisionerType{database.ProvisionerTypeEcho},
Tags: must(json.Marshal(orig.Tags)),
WorkerID: uuid.NullUUID{},
StartedAt: orig.StartedAt,
Types: []database.ProvisionerType{database.ProvisionerTypeEcho},
Tags: must(json.Marshal(orig.Tags)),
WorkerID: uuid.NullUUID{},
ExactTagMatch: false,
})
require.NoError(t, err)
// There is no easy way to make sure we acquire the correct job.
Expand Down
32 changes: 20 additions & 12 deletionscoderd/database/dbmem/dbmem.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -748,6 +748,22 @@ var deletedUserLinkError = &pq.Error{
Routine: "exec_stmt_raise",
}

// m1 and m2 are equal iff |m1| = |m2| ^ m2 ⊆ m1
func tagsEqual(m1, m2 map[string]string) bool {
return len(m1) == len(m2) && tagsSubset(m1, m2)
}

// m2 is a subset of m1 if each key in m1 exists in m2
// with the same value
func tagsSubset(m1, m2 map[string]string) bool {
for k, v1 := range m1 {
if v2, found := m2[k]; !found || v1 != v2 {
return false
}
}
return true
}

func (*FakeQuerier) AcquireLock(_ context.Context, _ int64) error {
return xerrors.New("AcquireLock must only be called within a transaction")
}
Expand DownExpand Up@@ -783,19 +799,11 @@ func (q *FakeQuerier) AcquireProvisionerJob(_ context.Context, arg database.Acqu
}
}

missing := false
for key, value := range provisionerJob.Tags {
provided, found := tags[key]
if !found {
missing = true
break
}
if provided != value {
missing = true
break
}
matchFunc := tagsSubset
if arg.ExactTagMatch {
matchFunc = tagsEqual
}
ifmissing {
if!matchFunc(provisionerJob.Tags, tags) {
continue
}
provisionerJob.StartedAt = arg.StartedAt
Expand Down
19 changes: 13 additions & 6 deletionscoderd/database/queries.sql.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more abouthow customized files appear on GitHub.

9 changes: 7 additions & 2 deletionscoderd/database/queries/provisionerjobs.sql
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,8 +21,13 @@ WHERE
nested.started_at IS NULL
-- Ensure the caller has the correct provisioner.
AND nested.provisioner = ANY(@types :: provisioner_type [ ])
-- Ensure the caller satisfies all job tags.
AND nested.tags <@ @tags :: jsonb
-- Ensure the caller satisfies all job tags if requested,
AND CASE
WHEN @exact_tag_match :: boolean THEN nested.tags = @tags :: jsonb
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

Are both guaranteed to be sorted arrays? Might be a good idea to allow unsorted equality.

Copy link
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

I think they'remap[string]string so we want to completely ignore order in comparisons.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others.Learn more.

Ah right, then ajsonb tojsonb comparison should be fine 👍🏻

-- Otherwise, ensure caller satisfies a subset of tags.
ELSE
nested.tags <@ @tags :: jsonb
END
ORDER BY
nested.created_at
FOR UPDATE
Expand Down
17 changes: 14 additions & 3 deletionscoderd/provisionerdserver/acquirer.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -49,6 +49,8 @@ type Acquirer struct {
mu sync.Mutex
q map[dKey]domain

exactTagMatch bool

// testing only
backupPollDuration time.Duration
}
Expand All@@ -61,6 +63,12 @@ func TestingBackupPollDuration(dur time.Duration) AcquirerOption {
}
}

func WithExactTagMatch() AcquirerOption {
return func(a *Acquirer) {
a.exactTagMatch = true
}
}

// AcquirerStore is the subset of database.Store that the Acquirer needs
type AcquirerStore interface {
AcquireProvisionerJob(context.Context, database.AcquireProvisionerJobParams) (database.ProvisionerJob, error)
Expand All@@ -76,6 +84,7 @@ func NewAcquirer(ctx context.Context, logger slog.Logger, store AcquirerStore, p
ps: ps,
q: make(map[dKey]domain),
backupPollDuration: backupPollDuration,
exactTagMatch: false,
}
for _, opt := range opts {
opt(a)
Expand All@@ -96,7 +105,8 @@ func (a *Acquirer) AcquireJob(
logger := a.logger.With(
slog.F("worker_id", worker),
slog.F("provisioner_types", pt),
slog.F("tags", tags))
slog.F("tags", tags),
slog.F("exact_tag_match", a.exactTagMatch))
logger.Debug(ctx, "acquiring job")
dk := domainKey(pt, tags)
dbTags, err := tags.ToJSON()
Expand DownExpand Up@@ -128,8 +138,9 @@ func (a *Acquirer) AcquireJob(
UUID: worker,
Valid: true,
},
Types: pt,
Tags: dbTags,
Types: pt,
Tags: dbTags,
ExactTagMatch: a.exactTagMatch,
})
if xerrors.Is(err, sql.ErrNoRows) {
logger.Debug(ctx, "no job available")
Expand Down
88 changes: 88 additions & 0 deletionscoderd/provisionerdserver/acquirer_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -9,6 +9,7 @@ import (
"time"

"github.com/google/uuid"
"github.com/sqlc-dev/pqtype"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/goleak"
Expand All@@ -18,6 +19,8 @@ import (
"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"
Expand DownExpand Up@@ -315,6 +318,91 @@ func TestAcquirer_UnblockOnCancel(t *testing.T) {
require.Equal(t, jobID, job.ID)
}

func TestAcquirer_ExactTagMatch(t *testing.T) {
t.Parallel()
if testing.Short() {
t.Skip("skipping this test due to -short")
}

for _, tt := range []struct {
name string
provisionerJobTags map[string]string
acquireJobTags map[string]string
expectAcquire bool
}{
{
name: "match",
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "foo": "bar"},
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "foo": "bar"},
expectAcquire: true,
},
{
name: "subset",
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "foo": "bar"},
acquireJobTags: map[string]string{"scope": "organization", "owner": ""},
expectAcquire: false,
},
{
name: "key mismatch",
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "fop": "bar"},
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "foo": "bar"},
expectAcquire: false,
},
{
name: "value mismatch",
provisionerJobTags: map[string]string{"scope": "organization", "owner": "", "foo": "baz"},
acquireJobTags: map[string]string{"scope": "organization", "owner": "", "foo": "bar"},
expectAcquire: false,
},
} {
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}
opts := []provisionerdserver.AcquirerOption{
provisionerdserver.WithExactTagMatch(),
}
acq := provisionerdserver.NewAcquirer(ctx, log, db, ps, opts...)
aj, err := acq.AcquireJob(ctx, uuid.New(), ptypes, tt.acquireJobTags)
if tt.expectAcquire {
require.NoError(t, err)
require.Equal(t, pj.ID, aj.ID)
} else {
require.ErrorIs(t, err, context.DeadlineExceeded, "should have timed out")
require.Empty(t, aj, "should not have acquired job")
}
})
}
}

func postJob(t *testing.T, ps pubsub.Pubsub, pt database.ProvisionerType, tags provisionerdserver.Tags) {
t.Helper()
msg, err := json.Marshal(provisionerjobs.JobPosting{
Expand Down
3 changes: 2 additions & 1 deletioncoderd/provisionerdserver/provisionerdserver_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -573,7 +573,8 @@ func TestUpdateJob(t *testing.T) {
UUID: srvID,
Valid: true,
},
Types: []database.ProvisionerType{database.ProvisionerTypeEcho},
Types: []database.ProvisionerType{database.ProvisionerTypeEcho},
ExactTagMatch: false,
})
require.NoError(t, err)
return job.ID
Expand Down
5 changes: 3 additions & 2 deletionsenterprise/coderd/schedule/template_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -181,8 +181,9 @@ func TestTemplateUpdateBuildDeadlines(t *testing.T) {
UUID: uuid.New(),
Valid: true,
},
Types: []database.ProvisionerType{database.ProvisionerTypeEcho},
Tags: json.RawMessage(fmt.Sprintf(`{%q: "yeah"}`, c.name)),
Types: []database.ProvisionerType{database.ProvisionerTypeEcho},
Tags: json.RawMessage(fmt.Sprintf(`{%q: "yeah"}`, c.name)),
ExactTagMatch: false,
})
require.NoError(t, err)
require.Equal(t, job.ID, acquiredJob.ID)
Expand Down
[8]ページ先頭
©2009-2025 Movatter.jp