CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(server): refactor worker tags (#5445)

Browse Source 
* refactor(server): refactor worker tags
This commit is contained in:
Irena Rindos authored and GitHub committed 2025-01-22 13:03:55 -05:00
1 parent e2102f65dd
commit dfe4f91be3
9 files changed
+148 -104
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/daemon/cluster/handlers/worker_service_test.go
+22 -34
View File
@@ -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)
internal/daemon/controller/handlers/workers/worker_service.go
+4 -4
View File
@@ -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"))
}
internal/server/repository_worker.go
+9 -3
View File
@@ -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)
}
internal/server/repository_worker_test.go
+6 -3
View File
@@ -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")
internal/server/worker.go
+2 -21
View File
@@ -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.
internal/server/worker_tag.go
+40 -10
View File
@@ -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
}
internal/server/worker_tag_test.go
+42 -6
View File
@@ -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])
}
}
internal/server/worker_tags_test.go
+7 -7
View File
@@ -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)
}
}
internal/server/worker_test.go
+16 -16
View File
@@ -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"},
})
})