Files
threshold/internal/server/worker_list_test.go
T
846c849589 Worker Domain Refactor (#5338)
* feat(server): Refactor LookupWorkerByName to be a test method

* feat(server): Split worker tags table

* refact(worker): Export api and config tags

* refact(server): Return worker id from UpsertWorkerStatus

* refact(server): Create ListWorkerStorageBucketUpdates domain function

* refact(server): Use dedicated HCPb-worker list func

* refact(server): Create ConnectionRoute domain function

* refact(handlers): Move worker selection logic to internal/server package

This commit moves all the logic to select a worker to handle a session
to the internal/server package. Most functions were moved to this
package with no changes, however, there are a couple of notable
exceptions:

- Given that we don't use GRPC errors in repos, all errors were adjusted
  to use the internal/errors package instead.
- StorageBucketFilterCredIdFn is now used to prevent the internal/server
  package from importing internal/storage/plugin, which would cause a
  dep cycle. Callers of SelectSessionWorkers must implement this func
  which is responsible for returning a storage bucket's worker filter as
  well as its associated credential id.
- authorizeSessionWithWorkerFilter now receives an
  internal/server.(*Repository) object when called. This is transparent
  to SelectSessionWorkers callers as it is handled internally.

* feat(server): Add FilteredWithFeatures fn and new server options

* refact(server): create new list and lookup worker queries

* test(server): Add WorkerList unit tests

* feat(server): Implement WorkerList randomness using crypto/rand pkg

* chore(db): add index to server_worker table

* refact(server): prefix test method with Test

* refact(server): rename Downstreamers interface to Graph

* chore(server): unexport and rename method

* chore: add doc string to WorkerAddress (#5420)

---------

Co-authored-by: Irena Rindos <irenarindos@users.noreply.github.com>
Co-authored-by: irenarindos <irena.rindos@gmail.com>
2025-01-08 15:18:03 +00:00

339 lines
8.7 KiB
Go

// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1
package server
import (
"math/rand/v2"
"strconv"
"testing"
"github.com/hashicorp/boundary/sdk/pbs/plugin"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestWorkerList_FilterWorkersByLocalStorageState(t *testing.T) {
t.Run("list of workers with unhealthy workers", func(t *testing.T) {
workers := WorkerList([]*Worker{
NewWorker("global",
WithName("worker1"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(AvailableLocalStorageState.String())),
NewWorker("global",
WithName("worker2"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(LowStorageLocalStorageState.String())),
NewWorker("global",
WithName("worker3"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(CriticallyLowStorageLocalStorageState.String())),
})
healthyWorkers := FilterWorkersByLocalStorageState(workers)
assert.Equal(t, 1, len(healthyWorkers))
})
t.Run("list of workers with some unsupported worker versions", func(t *testing.T) {
workers := WorkerList([]*Worker{
NewWorker("global",
WithName("worker1"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(AvailableLocalStorageState.String())),
NewWorker("global",
WithName("worker2"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(UnknownLocalStorageState.String())),
NewWorker("global",
WithName("worker3"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(CriticallyLowStorageLocalStorageState.String())),
NewWorker("global",
WithName("worker4"),
WithReleaseVersion("Boundary v0.15.0"),
WithLocalStorageState(CriticallyLowStorageLocalStorageState.String())),
})
healthyWorkers := FilterWorkersByLocalStorageState(workers)
assert.Equal(t, 2, len(healthyWorkers))
})
t.Run("list of workers with no healthy workers", func(t *testing.T) {
workers := WorkerList([]*Worker{
NewWorker("global",
WithName("worker1"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(LowStorageLocalStorageState.String())),
NewWorker("global",
WithName("worker2"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(OutOfStorageLocalStorageState.String())),
NewWorker("global",
WithName("worker3"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(CriticallyLowStorageLocalStorageState.String())),
})
healthyWorkers := FilterWorkersByLocalStorageState(workers)
assert.Equal(t, 0, len(healthyWorkers))
})
t.Run("list of workers with all unknown local storage state", func(t *testing.T) {
workers := WorkerList([]*Worker{
NewWorker("global",
WithName("worker1"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(UnknownLocalStorageState.String())),
NewWorker("global",
WithName("worker2"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(UnknownLocalStorageState.String())),
NewWorker("global",
WithName("worker3"),
WithReleaseVersion("Boundary v0.16.0"),
WithLocalStorageState(UnknownLocalStorageState.String())),
})
healthyWorkers := FilterWorkersByLocalStorageState(workers)
assert.Equal(t, 3, len(healthyWorkers))
})
}
func Test_FilterStorageBucketCredentialByWriteAccess(t *testing.T) {
t.Parallel()
cases := []struct {
name string
sbcState *plugin.StorageBucketCredentialState
expectedFilter bool
}{
{
name: "nil sbcState",
expectedFilter: true,
},
{
name: "nil underlying state",
sbcState: &plugin.StorageBucketCredentialState{},
expectedFilter: true,
},
{
name: "nil write permission",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{},
},
expectedFilter: true,
},
{
name: "ok state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Write: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_OK,
},
},
},
expectedFilter: true,
},
{
name: "unknown state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Write: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_UNKNOWN,
},
},
},
expectedFilter: true,
},
{
name: "error state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Write: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_ERROR,
},
},
},
expectedFilter: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
assert := assert.New(t)
assert.Equal(tc.expectedFilter, FilterStorageBucketCredentialByWriteAccess(tc.sbcState))
})
}
}
func Test_FilterStorageBucketCredentialByReadAccess(t *testing.T) {
t.Parallel()
cases := []struct {
name string
sbcState *plugin.StorageBucketCredentialState
expectedFilter bool
}{
{
name: "nil sbcState",
expectedFilter: true,
},
{
name: "nil underlying state",
sbcState: &plugin.StorageBucketCredentialState{},
expectedFilter: true,
},
{
name: "nil read permission",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{},
},
expectedFilter: true,
},
{
name: "ok state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Read: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_OK,
},
},
},
expectedFilter: true,
},
{
name: "unknown state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Read: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_UNKNOWN,
},
},
},
expectedFilter: true,
},
{
name: "error state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Read: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_ERROR,
},
},
},
expectedFilter: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
assert := assert.New(t)
assert.Equal(tc.expectedFilter, FilterStorageBucketCredentialByReadAccess(tc.sbcState))
})
}
}
func Test_FilterStorageBucketCredentialByDeleteAccess(t *testing.T) {
t.Parallel()
cases := []struct {
name string
sbcState *plugin.StorageBucketCredentialState
expectedFilter bool
}{
{
name: "nil sbcState",
expectedFilter: true,
},
{
name: "nil underlying state",
sbcState: &plugin.StorageBucketCredentialState{},
expectedFilter: true,
},
{
name: "nil delete permission",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{},
},
expectedFilter: true,
},
{
name: "ok state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Delete: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_OK,
},
},
},
expectedFilter: true,
},
{
name: "unknown state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Delete: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_UNKNOWN,
},
},
},
expectedFilter: true,
},
{
name: "error state",
sbcState: &plugin.StorageBucketCredentialState{
State: &plugin.Permissions{
Delete: &plugin.Permission{
State: plugin.StateType_STATE_TYPE_ERROR,
},
},
},
expectedFilter: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
assert := assert.New(t)
assert.Equal(tc.expectedFilter, FilterStorageBucketCredentialByDeleteAccess(tc.sbcState))
})
}
}
func TestShuffle(t *testing.T) {
t.Parallel()
t.Run("noElements", func(t *testing.T) {
in := WorkerList{}
out, err := in.Shuffle()
require.NoError(t, err)
require.Equal(t, in, out)
})
t.Run("oneElement", func(t *testing.T) {
in := WorkerList{NewWorker("test_scope", WithName("1"))}
out, err := in.Shuffle()
require.NoError(t, err)
require.Equal(t, in, out)
})
t.Run("multipleElements", func(t *testing.T) {
// We need a large amount of minimum workers here to statistically
// mitigate against the case where Shuffle just-so-happens to shuffle
// the elements into the same order they were in before.
n := rand.IntN(1000-99) + 100 // [100, 1000]
inOrder := make([]int, 0, n)
in := make(WorkerList, 0, n)
for i := 0; i < n; i++ {
inOrder = append(inOrder, i)
in = append(in, NewWorker("test", WithName(strconv.Itoa(i))))
}
out, err := in.Shuffle()
require.NoError(t, err)
require.ElementsMatch(t, in, out)
outOrder := make([]int, 0)
for _, w := range out {
i, _ := strconv.Atoi(w.Name)
outOrder = append(outOrder, i)
}
require.NotEqual(t, inOrder, outOrder)
})
}