From dfe4f91be38db299dafb68b85576b467f25e97e6 Mon Sep 17 00:00:00 2001 From: Irena Rindos Date: Wed, 22 Jan 2025 13:03:55 -0500 Subject: [PATCH] refactor(server): refactor worker tags (#5445) * refactor(server): refactor worker tags --- .../cluster/handlers/worker_service_test.go | 56 ++++++++----------- .../handlers/workers/worker_service.go | 8 +-- internal/server/repository_worker.go | 12 +++- internal/server/repository_worker_test.go | 9 ++- internal/server/worker.go | 23 +------- internal/server/worker_tag.go | 50 +++++++++++++---- internal/server/worker_tag_test.go | 48 ++++++++++++++-- internal/server/worker_tags_test.go | 14 ++--- internal/server/worker_test.go | 32 +++++------ 9 files changed, 148 insertions(+), 104 deletions(-) diff --git a/internal/daemon/cluster/handlers/worker_service_test.go b/internal/daemon/cluster/handlers/worker_service_test.go index d850cf123..46f222d9f 100644 --- a/internal/daemon/cluster/handlers/worker_service_test.go +++ b/internal/daemon/cluster/handlers/worker_service_test.go @@ -1316,17 +1316,14 @@ func TestRoutingInfo(t *testing.T) { require.NoError(t, err) assert.Equal(t, w1.Address, "2.3.4.5:8080") - expTags := []*server.Tag{ - {Key: "tag1", Value: "value1"}, - {Key: "tag2", Value: "value2"}, + expTags := server.Tags{ + "tag1": []string{"value1"}, + "tag2": []string{"value2"}, + } + assert.Equal(t, len(expTags), len(w1.ConfigTags)) + for k, v := range expTags { + assert.ElementsMatch(t, v, w1.ConfigTags[k]) } - assert.Empty( - t, - cmp.Diff( - w1.ConfigTags, - server.Tags(expTags), - ), - ) assert.Equal(t, w1.ReleaseVersion, "Boundary v0.18.0") assert.EqualValues(t, w1.OperationalState, server.ActiveOperationalState) assert.EqualValues(t, w1.LocalStorageState, server.AvailableLocalStorageState) @@ -1346,13 +1343,10 @@ func TestRoutingInfo(t *testing.T) { require.NoError(t, err) assert.Equal(t, w1.Address, "2.3.4.5:8080") - assert.Empty( - t, - cmp.Diff( - w1.ConfigTags, - server.Tags(expTags), - ), - ) + assert.Equal(t, len(expTags), len(w1.ConfigTags)) + for k, v := range expTags { + assert.ElementsMatch(t, v, w1.ConfigTags[k]) + } assert.Equal(t, w1.ReleaseVersion, "Boundary v0.18.0") assert.EqualValues(t, w1.OperationalState, server.ActiveOperationalState) assert.EqualValues(t, w1.LocalStorageState, server.AvailableLocalStorageState) @@ -1389,17 +1383,14 @@ func TestRoutingInfo(t *testing.T) { require.NoError(t, err) assert.Equal(t, w3.Address, "2.3.4.5:8080") - expTags := []*server.Tag{ - {Key: "tag1", Value: "value1"}, - {Key: "tag2", Value: "value2"}, + expTags := server.Tags{ + "tag1": []string{"value1"}, + "tag2": []string{"value2"}, + } + assert.Equal(t, len(expTags), len(w3.ConfigTags)) + for k, v := range expTags { + assert.ElementsMatch(t, v, w3.ConfigTags[k]) } - assert.Empty( - t, - cmp.Diff( - w3.ConfigTags, - server.Tags(expTags), - ), - ) assert.Equal(t, w3.ReleaseVersion, "Boundary v0.18.0") assert.EqualValues(t, w3.OperationalState, server.ActiveOperationalState) assert.EqualValues(t, w3.LocalStorageState, server.AvailableLocalStorageState) @@ -1419,13 +1410,10 @@ func TestRoutingInfo(t *testing.T) { require.NoError(t, err) assert.Equal(t, w3.Address, "2.3.4.5:8080") - assert.Empty( - t, - cmp.Diff( - w3.ConfigTags, - server.Tags(expTags), - ), - ) + assert.Equal(t, len(expTags), len(w3.ConfigTags)) + for k, v := range expTags { + assert.ElementsMatch(t, v, w3.ConfigTags[k]) + } assert.Equal(t, w3.ReleaseVersion, "Boundary v0.18.0") assert.EqualValues(t, w3.OperationalState, server.ActiveOperationalState) assert.EqualValues(t, w3.LocalStorageState, server.AvailableLocalStorageState) diff --git a/internal/daemon/controller/handlers/workers/worker_service.go b/internal/daemon/controller/handlers/workers/worker_service.go index 7d13185b2..db28d8582 100644 --- a/internal/daemon/controller/handlers/workers/worker_service.go +++ b/internal/daemon/controller/handlers/workers/worker_service.go @@ -905,16 +905,16 @@ func (s Service) toProto(ctx context.Context, in *server.Worker, opt ...handlers if outputFields.Has(globals.ControllerGeneratedActivationToken) && in.ControllerGeneratedActivationToken != "" { out.ControllerGeneratedActivationToken = &wrapperspb.StringValue{Value: in.ControllerGeneratedActivationToken} } - if outputFields.Has(globals.ConfigTagsField) && len(in.GetConfigTags()) > 0 { + if outputFields.Has(globals.ConfigTagsField) && len(in.ConfigTags) > 0 { var err error - out.ConfigTags, err = tagsToMapProto(in.GetConfigTags()) + out.ConfigTags, err = tagsToMapProto(in.ConfigTags) if err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing config tags proto")) } } - if outputFields.Has(globals.ApiTagsField) && len(in.GetApiTags()) > 0 { + if outputFields.Has(globals.ApiTagsField) && len(in.ApiTags) > 0 { var err error - out.ApiTags, err = tagsToMapProto(in.GetApiTags()) + out.ApiTags, err = tagsToMapProto(in.ApiTags) if err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing api tags proto")) } diff --git a/internal/server/repository_worker.go b/internal/server/repository_worker.go index f801eeaa7..435678b86 100644 --- a/internal/server/repository_worker.go +++ b/internal/server/repository_worker.go @@ -746,7 +746,7 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op // AddWorkerTags adds specified api tags to the repo worker and returns its new tags. // No options are currently supported. -func (r *Repository) AddWorkerTags(ctx context.Context, workerId string, workerVersion uint32, tags []*Tag, _ ...Option) ([]*Tag, error) { +func (r *Repository) AddWorkerTags(ctx context.Context, workerId string, workerVersion uint32, tags []*Tag, _ ...Option) (Tags, error) { const op = "server.(Repository).AddWorkerTags" switch { case workerId == "": @@ -765,7 +765,13 @@ func (r *Repository) AddWorkerTags(ctx context.Context, workerId string, workerV return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("no worker found with public id %s", workerId)) } - newTags := append(worker.ApiTags, tags...) + newTags := make(Tags) + for _, tag := range tags { + newTags[tag.Key] = append(newTags[tag.Key], tag.Value) + } + for k, v := range worker.ApiTags { + newTags[k] = append(newTags[k], v...) + } _, err = r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { worker := worker.clone() worker.PublicId = workerId @@ -777,7 +783,7 @@ func (r *Repository) AddWorkerTags(ctx context.Context, workerId string, workerV if rowsUpdated != 1 { return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("updated worker version and %d rows updated", rowsUpdated)) } - err = setWorkerApiTags(ctx, w, workerId, newTags) + err = setWorkerApiTags(ctx, w, workerId, newTags.convertToTag()) if err != nil { return errors.Wrap(ctx, err, op) } diff --git a/internal/server/repository_worker_test.go b/internal/server/repository_worker_test.go index 13a7815bc..f33e8491a 100644 --- a/internal/server/repository_worker_test.go +++ b/internal/server/repository_worker_test.go @@ -200,10 +200,13 @@ func TestLookupWorker(t *testing.T) { require.NoError(t, err) assert.Empty(t, cmp.Diff(w, got, protocmp.Transform())) assert.Equal(t, uint32(3), got.ActiveConnectionCount) - assert.Equal(t, map[string][]string{ + assert.Equal(t, server.Tags(map[string][]string{ "key": {"val"}, - }, got.CanonicalTags()) - assert.Equal(t, got.CanonicalTags(), got.GetConfigTags()) + }), got.CanonicalTags()) + assert.Equal(t, len(got.CanonicalTags()), len(got.ConfigTags)) + for k, v := range got.CanonicalTags() { + assert.ElementsMatch(t, v, got.ConfigTags[k]) + } }) t.Run("not found", func(t *testing.T) { got, err := repo.LookupWorker(ctx, "w_unknownid") diff --git a/internal/server/worker.go b/internal/server/worker.go index 17e6898c8..2432d46c7 100644 --- a/internal/server/worker.go +++ b/internal/server/worker.go @@ -148,7 +148,7 @@ func NewWorker(scopeId string, opt ...Option) *Worker { inputTags: opts.withWorkerTags, } if opts.withTestUseInputTagsAsApiTags { - worker.ApiTags = worker.inputTags + worker.ApiTags = convertToTags(worker.inputTags) } return worker } @@ -184,29 +184,10 @@ func (w *Worker) clone() *Worker { // CanonicalTags is the deduplicated set of tags contained on both the resource // set over the API as well as the tags reported by the worker itself. This // function is guaranteed to return a non-nil map. -func (w *Worker) CanonicalTags(opt ...Option) map[string][]string { +func (w *Worker) CanonicalTags() Tags { return compactTags(&w.ApiTags, &w.ConfigTags) } -// GetConfigTags returns the tags for this worker which has been set through -// the worker daemon's configuration file. -func (w *Worker) GetConfigTags() map[string][]string { - tags := make(map[string][]string) - for _, t := range w.ConfigTags { - tags[t.Key] = append(tags[t.Key], t.Value) - } - return tags -} - -// GetApiTags returns the api tags which have been set for this worker. -func (w *Worker) GetApiTags() map[string][]string { - tags := make(map[string][]string) - for _, t := range w.ApiTags { - tags[t.Key] = append(tags[t.Key], t.Value) - } - return tags -} - // GetLastStatusTime contains the last time the worker has reported to the // controller its connection status. If the worker has never reported to a // controller then nil is returned. diff --git a/internal/server/worker_tag.go b/internal/server/worker_tag.go index 7745a9a91..25994a89a 100644 --- a/internal/server/worker_tag.go +++ b/internal/server/worker_tag.go @@ -18,23 +18,36 @@ type Tag struct { Value string } +// convertToTags converts a slice of Tags to a map of tags +func convertToTags(tags []*Tag) Tags { + t := make(Tags) + for _, tag := range tags { + t[tag.Key] = append(t[tag.Key], tag.Value) + } + return t +} + // Tags allows us to scan a JSON array of worker tags from the database -type Tags []*Tag +type Tags map[string][]string func (t *Tags) clone() Tags { - newTags := make(Tags, 0, len(*t)) - for _, tag := range *t { - newTags = append(newTags, &Tag{Key: tag.Key, Value: tag.Value}) + newTags := make(map[string][]string) + for k, v := range *t { + for _, val := range v { + newTags[k] = append(newTags[k], val) + } } return newTags } -// compactTags takes a list of Tags and returns a map of deduplicated, compated tags +// compactTags takes a list of Tags and returns a map of deduplicated, compacted tags func compactTags(t ...*Tags) map[string][]string { compactedTags := make(map[Tag]struct{}) - for _, tags := range t { - for _, tag := range *tags { - compactedTags[*tag] = struct{}{} + for _, keys := range t { + for key, tags := range *keys { + for _, tag := range tags { + compactedTags[Tag{Key: key, Value: tag}] = struct{}{} + } } } @@ -45,17 +58,34 @@ func compactTags(t ...*Tags) map[string][]string { return tags } +// convertToTag converts a map of tags to a slice of Tags +func (t *Tags) convertToTag() []*Tag { + var tags []*Tag + for key, values := range *t { + for _, value := range values { + tags = append(tags, &Tag{Key: key, Value: value}) + } + } + return tags +} + // Scan scans value into Tags, and implements the sql.Scanner interface func (t *Tags) Scan(in any) error { var err error + var test []*Tag switch v := in.(type) { case string: - err = json.Unmarshal([]byte(v), &t) + err = json.Unmarshal([]byte(v), &test) case []byte: - err = json.Unmarshal(v, &t) + err = json.Unmarshal([]byte(v), &test) default: return fmt.Errorf("cannot scan type %T into tags", in) } + mapTags := make(map[string][]string) + for _, tag := range test { + mapTags[tag.Key] = append(mapTags[tag.Key], tag.Value) + } + *t = mapTags return err } diff --git a/internal/server/worker_tag_test.go b/internal/server/worker_tag_test.go index c17f2688f..e3f621a09 100644 --- a/internal/server/worker_tag_test.go +++ b/internal/server/worker_tag_test.go @@ -10,15 +10,14 @@ import ( "github.com/stretchr/testify/assert" ) -func TestDeduplicateTags(t *testing.T) { +func TestCompactTags(t *testing.T) { foo := 5 tags := make([]*Tags, 0, foo) for pippo := 0; pippo < foo; pippo++ { - tags = append(tags, &Tags{ - {Key: "key", Value: "shared"}, - {Key: "key", Value: fmt.Sprintf("unique-%d", pippo)}, - {Key: fmt.Sprintf("key-%d", pippo), Value: fmt.Sprintf("another-unique-%d", pippo)}, - }) + thisTag := make(Tags) + thisTag["key"] = []string{"shared", fmt.Sprintf("unique-%d", pippo)} + thisTag[fmt.Sprintf("key-%d", pippo)] = []string{fmt.Sprintf("another-unique-%d", pippo)} + tags = append(tags, &thisTag) } gotMap := compactTags(tags...) assert.Len(t, gotMap, 6) @@ -26,3 +25,40 @@ func TestDeduplicateTags(t *testing.T) { assert.ElementsMatch(t, gotMap["key-0"], []string{"another-unique-0"}) assert.ElementsMatch(t, gotMap["key-4"], []string{"another-unique-4"}) } + +func TestConvertToTag(t *testing.T) { + numTags := 5 + wantSlice := make([]*Tag, 0, numTags) + tags := make(Tags) + for i := 0; i < numTags; i++ { + thisTag := &Tag{ + Key: fmt.Sprintf("key-%d", i), + Value: fmt.Sprintf("unique-%d", i), + } + wantSlice = append(wantSlice, thisTag) + tags[fmt.Sprintf("key-%d", i)] = []string{fmt.Sprintf("unique-%d", i)} + } + gotTags := tags.convertToTag() + + assert.ElementsMatch(t, gotTags, wantSlice) +} + +func TestConvertToTags(t *testing.T) { + numTags := 5 + tagsSlice := make([]*Tag, 0, numTags) + wantTags := make(Tags) + for i := 0; i < numTags; i++ { + thisTag := &Tag{ + Key: fmt.Sprintf("key-%d", i), + Value: fmt.Sprintf("unique-%d", i), + } + tagsSlice = append(tagsSlice, thisTag) + wantTags[fmt.Sprintf("key-%d", i)] = []string{fmt.Sprintf("unique-%d", i)} + } + tags := convertToTags(tagsSlice) + + assert.Equal(t, len(tags), len(wantTags)) + for k, v := range tags { + assert.ElementsMatch(t, v, wantTags[k]) + } +} diff --git a/internal/server/worker_tags_test.go b/internal/server/worker_tags_test.go index 3f3c19e71..0e7676a93 100644 --- a/internal/server/worker_tags_test.go +++ b/internal/server/worker_tags_test.go @@ -259,7 +259,7 @@ func TestRepository_AddWorkerTags(t *testing.T) { return } assert.NoError(err) - assert.Equal(tt.want, got) + assert.ElementsMatch(tt.want, got.convertToTag()) repoWorker, err := repo.LookupWorker(context.Background(), tt.args.publicId) require.NoError(err) assert.Equal(tt.args.version+1, repoWorker.Version) @@ -604,7 +604,7 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) { } added, err := repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, manyTags) assert.NoError(err) - assert.Equal(manyTags, added) + assert.ElementsMatch(manyTags, added.convertToTag()) worker, err = repo.LookupWorker(context.Background(), worker.PublicId) require.NoError(err) assert.Equal(uint32(2), worker.Version) @@ -650,8 +650,8 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) { worker, err = repo.LookupWorker(context.Background(), worker.PublicId) require.NoError(err) assert.Equal(uint32(3), worker.Version) - assert.Contains(worker.ApiTags, &Tag{Key: "key", Value: "value"}) - assert.Contains(worker.ApiTags, &Tag{Key: "key2", Value: "value2"}) + assert.Contains(worker.ApiTags.convertToTag(), &Tag{Key: "key", Value: "value"}) + assert.Contains(worker.ApiTags.convertToTag(), &Tag{Key: "key2", Value: "value2"}) // Add another valid tag to worker _, err = repo.AddWorkerTags(context.Background(), worker.PublicId, worker.Version, []*Tag{ @@ -661,7 +661,7 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) { worker, err = repo.LookupWorker(context.Background(), worker.PublicId) require.NoError(err) assert.Equal(uint32(4), worker.Version) - assert.Contains(worker.ApiTags, &Tag{Key: "key!", Value: "value!"}) + assert.Contains(worker.ApiTags.convertToTag(), &Tag{Key: "key!", Value: "value!"}) assert.Equal(3, len(worker.ApiTags)) // Set all tags to nil @@ -681,7 +681,7 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) { {Key: "key3", Value: "value3"}, {Key: "key?", Value: "value?"}, } { - assert.Contains(worker.ConfigTags, ct) + assert.Contains(worker.ConfigTags.convertToTag(), ct) } assert.Equal(4, len(worker.ConfigTags)) @@ -693,6 +693,6 @@ func TestRepository_WorkerTagsConsequent(t *testing.T) { assert.Equal(uint32(6), worker.Version) assert.Equal(len(manyTags), len(worker.ApiTags)) for _, t := range manyTags { - assert.Contains(worker.ApiTags, t) + assert.Contains(worker.ApiTags.convertToTag(), t) } } diff --git a/internal/server/worker_test.go b/internal/server/worker_test.go index 45bae5406..c7d0b2594 100644 --- a/internal/server/worker_test.go +++ b/internal/server/worker_test.go @@ -24,16 +24,12 @@ import ( func TestWorkerCanonicalTags(t *testing.T) { w := NewWorker(scope.Global.String()) - w.ApiTags = []*Tag{ - {Key: "key", Value: "apis unique"}, - {Key: "key", Value: "shared"}, - {Key: "key2", Value: "apis key2 unique"}, - } - w.ConfigTags = []*Tag{ - {Key: "key", Value: "configs unique"}, - {Key: "key", Value: "shared"}, - {Key: "key3", Value: "configs key3 unique"}, - } + w.ApiTags = make(Tags) + w.ApiTags["key"] = []string{"apis unique", "shared"} + w.ApiTags["key2"] = []string{"apis key2 unique"} + w.ConfigTags = make(Tags) + w.ConfigTags["key"] = []string{"configs unique", "shared"} + w.ConfigTags["key3"] = []string{"configs key3 unique"} got := w.CanonicalTags() assert.Len(t, got, 3, "2 keys expected, 'key' and 'key2'") @@ -119,7 +115,9 @@ func TestWorkerAggregate(t *testing.T) { assert.NotNil(t, got.GetLastStatusTime()) assert.NotNil(t, got.GetReleaseVersion()) assert.Empty(t, got.ApiTags) - assert.Equal(t, got.ConfigTags, Tags{{Key: "key", Value: "val"}}) + wantTag := make(Tags) + wantTag["key"] = []string{"val"} + assert.Equal(t, got.ConfigTags, wantTag) }) t.Run("Worker with many config tag", func(t *testing.T) { @@ -150,7 +148,7 @@ func TestWorkerAggregate(t *testing.T) { got := getWorker(id) require.NotNil(t, got.GetLastStatusTime()) assert.Empty(t, got.ApiTags) - assert.ElementsMatch(t, got.ConfigTags, []*Tag{ + assert.ElementsMatch(t, got.ConfigTags.convertToTag(), []*Tag{ {Key: "key", Value: "val"}, {Key: "key", Value: "val2"}, {Key: "key2", Value: "val2"}, @@ -177,8 +175,10 @@ func TestWorkerAggregate(t *testing.T) { assert.Equal(t, id, got.GetPublicId()) assert.Equal(t, uint32(1), got.GetVersion()) assert.NotNil(t, got.GetLastStatusTime()) - assert.Empty(t, got.GetConfigTags()) - assert.Equal(t, got.ApiTags, Tags{{Key: "key", Value: "val"}}) + assert.Empty(t, got.ConfigTags) + wantTag := make(Tags) + wantTag["key"] = []string{"val"} + assert.Equal(t, got.ApiTags, wantTag) }) // Worker with mix of tag sources @@ -209,11 +209,11 @@ func TestWorkerAggregate(t *testing.T) { got := getWorker(id) require.NotNil(t, got.GetLastStatusTime()) - assert.ElementsMatch(t, got.ApiTags, []*Tag{ + assert.ElementsMatch(t, got.ApiTags.convertToTag(), []*Tag{ {Key: "key", Value: "val2"}, {Key: "key2", Value: "val2"}, }) - assert.ElementsMatch(t, got.ConfigTags, []*Tag{ + assert.ElementsMatch(t, got.ConfigTags.convertToTag(), []*Tag{ {Key: "key", Value: "val"}, }) })