diff --git a/.github/workflows/test-race.yml b/.github/workflows/test-race.yml new file mode 100644 index 000000000..efd1cfda9 --- /dev/null +++ b/.github/workflows/test-race.yml @@ -0,0 +1,180 @@ +name: test-race + +on: + push: + paths-ignore: + - 'website/**' + workflow_call: + workflow_dispatch: + +permissions: + contents: read + +env: + DOCKER_MIRROR: docker.mirror.hashicorp.services + +jobs: + setup: + outputs: + go-version: ${{ steps.get-go-version.outputs.go-version }} + cache-go-build: ${{ steps.go-cache-paths.outputs.go-build }} + cache-go-mod: ${{ steps.go-cache-paths.outputs.go-mod }} + cache-go-bin: ${{ steps.go-cache-paths.outputs.go-bin }} + go-cache-key: ${{ steps.go-cache-key.outputs.key }} + plugin-cache-path: ${{ steps.plugin-cache-paths.outputs.path }} + plugin-cache-key: ${{ steps.plugin-cache-key.outputs.key }} + runs-on: ${{ fromJSON(vars.RUNNER) }} + steps: + - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1 + with: + fetch-depth: '0' + - name: Determine Go version + id: get-go-version + # We use .go-version as our source of truth for current Go + # version, because "goenv" can react to it automatically. + run: | + echo "Building with Go $(cat .go-version)" + echo "go-version=$(cat .go-version)" >> "$GITHUB_OUTPUT" + - name: Set up Go + uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0 + with: + go-version: "${{ steps.get-go-version.outputs.go-version }}" + cache: false + - name: Determine go cache key + id: go-cache-key + run: | + echo "key=${{ runner.os }}-go-${{ hashFiles('**/go.sum', './Makefile', './tools/tools.go') }}" >> "$GITHUB_OUTPUT" + - name: Determine Go cache paths + id: go-cache-paths + run: | + echo "go-build=$(go env GOCACHE)" >> "$GITHUB_OUTPUT" + echo "go-mod=$(go env GOMODCACHE)" >> "$GITHUB_OUTPUT" + echo "go-bin=$(go env GOPATH)/bin" >> "$GITHUB_OUTPUT" + - name: Set up Go modules cache + uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0 + with: + path: | + ${{ steps.go-cache-paths.outputs.go-build }} + ${{ steps.go-cache-paths.outputs.go-mod }} + ${{ steps.go-cache-paths.outputs.go-bin }} + key: ${{ steps.go-cache-key.outputs.key }} + restore-keys: | + ${{ runner.os }}-go + - name: Install Tools + run: | + go mod download + make tools + - name: Determine plugin cache key + id: plugin-cache-key + run: | + echo "key=${{ runner.os }}-plugins-${{ hashFiles('plugins/**/*.go', 'plugins/**/go.sum', './Makefile', './scripts/plugins.sh') }}" >> "$GITHUB_OUTPUT" + - name: Determin plugin cache path + id: plugin-cache-paths + run: | + echo "path=plugins/**/assets/*.gz" >> "$GITHUB_OUTPUT" + - name: Set up plugin cache + id: plugin-cache + uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0 + with: + path: | + ${{ steps.plugin-cache-paths.outputs.path }} + key: ${{ steps.plugin-cache-key.outputs.key }} + restore-keys: | + ${{ runner.os }}-plugin + - name: Build Plugins + if: steps.plugin-cache.outputs.cache-hit != 'true' + run: | + make build-plugins + test-modules: + needs: + - setup + runs-on: ${{ fromJSON(vars.RUNNER) }} + strategy: + matrix: + module: ["api", "sdk"] + steps: + - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1 + - name: Set up go + uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0 + with: + go-version: "${{ needs.setup.outputs.go-version }}" + cache: false + - name: Set up Go modules cache + uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0 + with: + path: | + ${{ needs.setup.outputs.cache-go-build }} + ${{ needs.setup.outputs.cache-go-mod }} + ${{ needs.setup.outputs.cache-go-bin }} + key: ${{ needs.setup.outputs.go-cache-key }} + restore-keys: | + ${{ runner.os }}-go + fail-on-cache-miss: true + - name: Test ${{ matrix.module }} Module + run: | + make test-${{ matrix.module }} + + test: + needs: + - setup + runs-on: ${{ fromJSON(vars.RUNNER_LARGE) }} + steps: + - name: ulimit + run: | + echo "Soft limits" + ulimit -Sa + echo "Hard limits" + ulimit -Ha + - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1 + - name: Set up go + uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0 + with: + go-version: "${{ needs.setup.outputs.go-version }}" + cache: false + - name: Set up Go modules cache + uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0 + with: + path: | + ${{ needs.setup.outputs.cache-go-build }} + ${{ needs.setup.outputs.cache-go-mod }} + ${{ needs.setup.outputs.cache-go-bin }} + key: ${{ needs.setup.outputs.go-cache-key }} + restore-keys: | + ${{ runner.os }}-go + fail-on-cache-miss: true + - name: Set up plugin cache + id: plugin-cache + uses: actions/cache@13aacd865c20de90d75de3b17ebe84f7a17d57d2 # v4.0.0 + with: + path: | + ${{ needs.setup.outputs.plugin-cache-path }} + key: ${{ needs.setup.outputs.plugin-cache-key }} + restore-keys: | + ${{ runner.os }}-plugin + - name: GH fix for localhost resolution + if: github.repository == 'hashicorp/boundary' + run: | + cat /etc/hosts && echo "-----------" + sudo sed -i 's/::1 *localhost ip6-localhost ip6-loopback/::1 ip6 -localhost ip6-loopback/g' /etc/hosts + cat /etc/hosts + - name: Initialize Test Database + run: | + which pg_isready || sudo apt-get update && sudo apt-get install -y postgresql-client + make DOCKER_ARGS='-d' PG_OPTS='-c shared_buffers=256MB -c max_connections=200000' -C testing/dbtest/docker database-up + until pg_isready -h 127.0.0.1; do docker container inspect boundary-sql-tests &> /dev/null || exit 255; sleep 1; done + + - name: Test + uses: nick-fields/retry@14672906e672a08bd6eeb15720e9ed3ce869cdd4 # v2.9.0 + env: + TEST_PACKAGE: "./..." + TEST_TIMEOUT: 120m + TESTARGS: -race -v + with: + max_attempts: 1 + timeout_minutes: 120 + retry_on: error + command: make test + - name: Cleanup + if: always() + run: | + make -C testing/dbtest/docker clean diff --git a/go.mod b/go.mod index 1fb35abcc..a30d861e7 100644 --- a/go.mod +++ b/go.mod @@ -19,8 +19,8 @@ require ( github.com/hashicorp/boundary/api v0.0.45 github.com/hashicorp/boundary/sdk v0.0.42 github.com/hashicorp/cap v0.5.0 - github.com/hashicorp/dawdle v0.4.0 - github.com/hashicorp/eventlogger v0.2.8 + github.com/hashicorp/dawdle v0.5.0 + github.com/hashicorp/eventlogger v0.2.9 github.com/hashicorp/eventlogger/filters/encrypt v0.1.8-0.20231208142215-efdb51ec090d github.com/hashicorp/go-bexpr v0.1.13 github.com/hashicorp/go-cleanhttp v0.5.2 diff --git a/go.sum b/go.sum index b3466f597..3f5534f3e 100644 --- a/go.sum +++ b/go.sum @@ -194,15 +194,15 @@ github.com/hashicorp/cap v0.5.0 h1:YIlAYxdXXtx2IL1JDvP2OyEl55Ooi0yl573kSB9Orlw= github.com/hashicorp/cap v0.5.0/go.mod h1:IAy00Er+ZFpMo+5x6B4bkO2HgpzgrkfsuDWMmHAuKUE= github.com/hashicorp/cap/ldap v0.0.0-20240206183135-ed8f24513744 h1:6cU01ORaCUxRXRS0YYnmm4tMfAvFRjamhjgMrgmoWZs= github.com/hashicorp/cap/ldap v0.0.0-20240206183135-ed8f24513744/go.mod h1:SOJuo/OFD0qKQ7d8qOy+a2B9Tw0/97oHIo+qXxs7HZE= -github.com/hashicorp/dawdle v0.4.0 h1:nAE+aCoAmVulgCPPPrp6oGOykjPhxEe4VUNms/mTKeA= -github.com/hashicorp/dawdle v0.4.0/go.mod h1:gOBY8rFXpW4Rw39up1FIacw/1pG/fO779LfchCY3KUw= +github.com/hashicorp/dawdle v0.5.0 h1:cPKzDUfuq5JhhAcChg8cygWstV+WQ7MSn7FSP7AtPnI= +github.com/hashicorp/dawdle v0.5.0/go.mod h1:gOBY8rFXpW4Rw39up1FIacw/1pG/fO779LfchCY3KUw= github.com/hashicorp/dbassert v0.0.0-20231012105025-1bc1bd88e22b h1:rvg3eCucdlpfsnbYXM1Ett4f0reNmIuCruyJUF4iw/0= github.com/hashicorp/dbassert v0.0.0-20231012105025-1bc1bd88e22b/go.mod h1:GCYXHZsaYdDfPh/xD3A/t6754HPpumeLkA0vLpYVjZE= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= -github.com/hashicorp/eventlogger v0.2.8 h1:WITwN1QGMtxHQE7l69WBwLJIJA5vH5pcvX69peURU14= -github.com/hashicorp/eventlogger v0.2.8/go.mod h1://CHt6/j+Q2lc0NlUB5af4aS2M0c0aVBg9/JfcpAyhM= +github.com/hashicorp/eventlogger v0.2.9 h1:QO8hPXNngadMp72FVNTwIduLAG9fcLP7t59bSFd7gDY= +github.com/hashicorp/eventlogger v0.2.9/go.mod h1://CHt6/j+Q2lc0NlUB5af4aS2M0c0aVBg9/JfcpAyhM= github.com/hashicorp/eventlogger/filters/encrypt v0.1.8-0.20231208142215-efdb51ec090d h1:Zia/aoFQmhtzqY6NVR0DVcxmZowbavqbTeum1bRRZhs= github.com/hashicorp/eventlogger/filters/encrypt v0.1.8-0.20231208142215-efdb51ec090d/go.mod h1:tMywUTIvdB/FXhwm6HMTt61C8/eODY6gitCHhXtyojg= github.com/hashicorp/go-bexpr v0.1.13 h1:HNwp7vZrMpRq8VZXj8VF90LbZpRjQQpim1oJF0DgSwg= diff --git a/internal/auth/ldap/service_list_accounts_ext_test.go b/internal/auth/ldap/service_list_accounts_ext_test.go index 0d79332e8..7a6565be6 100644 --- a/internal/auth/ldap/service_list_accounts_ext_test.go +++ b/internal/auth/ldap/service_list_accounts_ext_test.go @@ -260,7 +260,7 @@ func TestService_ListAccounts(t *testing.T) { filterFunc := func(_ context.Context, a auth.Account) (bool, error) { return true, nil } - _, err = ldap.ListAccountsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) + _, err := ldap.ListAccountsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("nil repo", func(t *testing.T) { diff --git a/internal/auth/ldap/service_list_managed_groups_ext_test.go b/internal/auth/ldap/service_list_managed_groups_ext_test.go index cee551e0b..7ea6dad1f 100644 --- a/internal/auth/ldap/service_list_managed_groups_ext_test.go +++ b/internal/auth/ldap/service_list_managed_groups_ext_test.go @@ -175,7 +175,7 @@ func TestService_ListManagedGroups(t *testing.T) { filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) { return true, nil } - _, err = ldap.ListManagedGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) + _, err := ldap.ListManagedGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { @@ -263,7 +263,7 @@ func TestService_ListManagedGroups(t *testing.T) { filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) { return true, nil } - _, err = ldap.ListManagedGroupsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) + _, err := ldap.ListManagedGroupsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("nil repo", func(t *testing.T) { @@ -341,7 +341,7 @@ func TestService_ListManagedGroups(t *testing.T) { filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) { return true, nil } - _, err = ldap.ListManagedGroupsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) + _, err := ldap.ListManagedGroupsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/auth/oidc/service_list_accounts_ext_test.go b/internal/auth/oidc/service_list_accounts_ext_test.go index a54feedfe..af53740a6 100644 --- a/internal/auth/oidc/service_list_accounts_ext_test.go +++ b/internal/auth/oidc/service_list_accounts_ext_test.go @@ -266,7 +266,7 @@ func TestService_ListAccounts(t *testing.T) { filterFunc := func(_ context.Context, a auth.Account) (bool, error) { return true, nil } - _, err = oidc.ListAccountsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) + _, err := oidc.ListAccountsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("nil repo", func(t *testing.T) { diff --git a/internal/auth/oidc/service_list_managed_groups_ext_test.go b/internal/auth/oidc/service_list_managed_groups_ext_test.go index e1e12fac4..88ec611b0 100644 --- a/internal/auth/oidc/service_list_managed_groups_ext_test.go +++ b/internal/auth/oidc/service_list_managed_groups_ext_test.go @@ -181,7 +181,7 @@ func TestService_ListManagedGroups(t *testing.T) { filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) { return true, nil } - _, err = oidc.ListManagedGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) + _, err := oidc.ListManagedGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { @@ -347,7 +347,7 @@ func TestService_ListManagedGroups(t *testing.T) { filterFunc := func(_ context.Context, mg auth.ManagedGroup) (bool, error) { return true, nil } - _, err = oidc.ListManagedGroupsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) + _, err := oidc.ListManagedGroupsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/auth/password/service_list_accounts_ext_test.go b/internal/auth/password/service_list_accounts_ext_test.go index 31fb7ca16..d122ac654 100644 --- a/internal/auth/password/service_list_accounts_ext_test.go +++ b/internal/auth/password/service_list_accounts_ext_test.go @@ -162,7 +162,7 @@ func TestService_ListAccounts(t *testing.T) { filterFunc := func(_ context.Context, a auth.Account) (bool, error) { return true, nil } - _, err = password.ListAccountsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) + _, err := password.ListAccountsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, authMethod.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/authtoken/service_list_ext_test.go b/internal/authtoken/service_list_ext_test.go index dfa081fc1..e97bf1ffa 100644 --- a/internal/authtoken/service_list_ext_test.go +++ b/internal/authtoken/service_list_ext_test.go @@ -155,7 +155,7 @@ func TestService_List(t *testing.T) { filterFunc := func(_ context.Context, at *authtoken.AuthToken) (bool, error) { return true, nil } - _, err = authtoken.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()}) + _, err := authtoken.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()}) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { @@ -243,7 +243,7 @@ func TestService_List(t *testing.T) { filterFunc := func(_ context.Context, at *authtoken.AuthToken) (bool, error) { return true, nil } - _, err = authtoken.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()}) + _, err := authtoken.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()}) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/bsr/testing.go b/internal/bsr/testing.go index f25a07bce..b72ad418d 100644 --- a/internal/bsr/testing.go +++ b/internal/bsr/testing.go @@ -11,11 +11,7 @@ import ( "github.com/stretchr/testify/require" ) -var registerTestProtocol *sync.Once - -func init() { - registerTestProtocol = new(sync.Once) -} +var registerTestProtocol *sync.Once = new(sync.Once) // TestRegisterSummaryAllocFunc registers "TEST" as a protocol for all container types // The channel summary will include the following: diff --git a/internal/clientcache/internal/daemon/server.go b/internal/clientcache/internal/daemon/server.go index c25808c94..73af030a1 100644 --- a/internal/clientcache/internal/daemon/server.go +++ b/internal/clientcache/internal/daemon/server.go @@ -14,6 +14,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/hashicorp/boundary/api" @@ -62,7 +63,7 @@ type CacheServer struct { info map[string]string storeUrl string - store *db.DB + store atomic.Pointer[db.DB] tickerWg *sync.WaitGroup httpSrv *http.Server @@ -224,9 +225,13 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e s.info["Store debug"] = strconv.FormatBool(s.conf.StoreDebug) s.infoKeys = append(s.infoKeys, "Store debug") - if s.store, s.storeUrl, err = openStore(ctx, s.conf.DatabaseUrl, s.conf.StoreDebug); err != nil { + var store *db.DB + store, s.storeUrl, err = openStore(ctx, s.conf.DatabaseUrl, s.conf.StoreDebug) + if err != nil { return errors.Wrap(ctx, err, op) } + s.store.Store(store) + if s.storeUrl != "" { s.info["Database URL"] = s.storeUrl s.infoKeys = append(s.infoKeys, "Database URL") @@ -260,7 +265,7 @@ func (s *CacheServer) Serve(ctx context.Context, cmd Commander, opt ...Option) e s.printInfo(ctx) - repo, err := cache.NewRepository(ctx, s.store, &sync.Map{}, cmd.ReadTokenFromKeyring, opts.withBoundaryTokenReaderFunc) + repo, err := cache.NewRepository(ctx, s.store.Load(), &sync.Map{}, cmd.ReadTokenFromKeyring, opts.withBoundaryTokenReaderFunc) if err != nil { return errors.Wrap(ctx, err, op) } diff --git a/internal/clientcache/internal/daemon/testing.go b/internal/clientcache/internal/daemon/testing.go index 649bde1b6..f185e5bd1 100644 --- a/internal/clientcache/internal/daemon/testing.go +++ b/internal/clientcache/internal/daemon/testing.go @@ -71,7 +71,7 @@ func (s *TestServer) Serve(t *testing.T, opt ...Option) error { func (s *TestServer) AddKeyringToken(t *testing.T, address, keyring, tokenName, tokenId string, atReadFn cache.BoundaryTokenReaderFn) { t.Helper() ctx := context.Background() - r, err := cache.NewRepository(ctx, s.CacheServer.store, &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn) + r, err := cache.NewRepository(ctx, s.CacheServer.store.Load(), &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn) require.NoError(t, err) require.NoError(t, r.AddKeyringToken(ctx, address, cache.KeyringToken{ @@ -86,7 +86,7 @@ func (s *TestServer) AddKeyringToken(t *testing.T, address, keyring, tokenName, func (s *TestServer) AddResources(t *testing.T, p *authtokens.AuthToken, tars []*targets.Target, sess []*sessions.Session, atReadFn cache.BoundaryTokenReaderFn) { t.Helper() ctx := context.Background() - r, err := cache.NewRepository(ctx, s.CacheServer.store, &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn) + r, err := cache.NewRepository(ctx, s.CacheServer.store.Load(), &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn) require.NoError(t, err) tarFn := func(ctx context.Context, _, tok string, _ cache.RefreshTokenValue) ([]*targets.Target, []string, cache.RefreshTokenValue, error) { @@ -113,7 +113,7 @@ func (s *TestServer) AddResources(t *testing.T, p *authtokens.AuthToken, tars [] func (s *TestServer) AddUnsupportedCachingData(t *testing.T, p *authtokens.AuthToken, atReadFn cache.BoundaryTokenReaderFn) { t.Helper() ctx := context.Background() - r, err := cache.NewRepository(ctx, s.CacheServer.store, &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn) + r, err := cache.NewRepository(ctx, s.CacheServer.store.Load(), &sync.Map{}, s.cmd.ReadTokenFromKeyring, atReadFn) require.NoError(t, err) tarFn := func(ctx context.Context, _, tok string, _ cache.RefreshTokenValue) ([]*targets.Target, []string, cache.RefreshTokenValue, error) { diff --git a/internal/clientcache/internal/daemon/ticker.go b/internal/clientcache/internal/daemon/ticker.go index 005eb7118..8ac229d13 100644 --- a/internal/clientcache/internal/daemon/ticker.go +++ b/internal/clientcache/internal/daemon/ticker.go @@ -37,7 +37,6 @@ type refreshTicker struct { refreshInterval time.Duration recheckSupportInterval time.Duration randomizationFactor float64 - rand *rand.Rand refreshChan chan struct{} refresher refreshService } @@ -80,7 +79,6 @@ func newRefreshTicker(ctx context.Context, refresh refreshService, opt ...Option refreshInterval: refreshInterval, recheckSupportInterval: recheckSupportInterval, randomizationFactor: randomizationFactor, - rand: rand.New(rand.NewSource(time.Now().UnixNano())), refresher: refresh, // We make this channel size 1 so if something happens midway through the refresh @@ -151,12 +149,12 @@ func (rt *refreshTicker) nextIntervalWithRandomness(d time.Duration) time.Durati return d } - randFactor := rt.rand.Float64() + randFactor := rand.Float64() randFactor = randFactor * rt.randomizationFactor randDur := randFactor * float64(d) // Half a chance to be faster, not slower - if rt.rand.Float32() > 0.5 { + if rand.Float32() > 0.5 { randDur = -1 * randDur } return d + time.Duration(randDur) diff --git a/internal/cmd/base/logging.go b/internal/cmd/base/logging.go index b470ae3a9..8277e5bcb 100644 --- a/internal/cmd/base/logging.go +++ b/internal/cmd/base/logging.go @@ -7,9 +7,11 @@ import ( "fmt" "os" "strings" + "sync/atomic" "github.com/hashicorp/boundary/internal/cmd/base/logging" "github.com/hashicorp/go-hclog" + "google.golang.org/grpc/grpclog" ) func ProcessLogLevelAndFormat(flagLogLevel, flagLogFormat, configLogLevel, configLogFormat string) (hclog.Level, logging.LogFormat, error) { @@ -59,77 +61,140 @@ func ProcessLogLevelAndFormat(flagLogLevel, flagLogFormat, configLogLevel, confi return level, logFormat, nil } +type grpcLogFakerLogHolder struct { + logger hclog.Logger +} + +var grpcLogFaker *GRPCLogFaker = &GRPCLogFaker{ + loggingEnabled: new(atomic.Bool), + logger: new(atomic.Pointer[grpcLogFakerLogHolder]), +} + +func init() { + grpclog.SetLoggerV2(grpcLogFaker) +} + type GRPCLogFaker struct { - Logger hclog.Logger - Log bool + logger *atomic.Pointer[grpcLogFakerLogHolder] + loggingEnabled *atomic.Bool +} + +func (g *GRPCLogFaker) SetLogOnOff(on bool) { + g.loggingEnabled.Store(on) +} + +func (g *GRPCLogFaker) SetLogger(logger hclog.Logger) { + g.logger.Store(&grpcLogFakerLogHolder{logger: logger}) } func (g *GRPCLogFaker) Fatal(args ...any) { - g.Logger.Error(fmt.Sprint(args...)) + if l := g.logger.Load(); l != nil { + l.logger.Error(fmt.Sprint(args...)) + } os.Exit(1) } func (g *GRPCLogFaker) Fatalf(format string, args ...any) { - g.Logger.Error(fmt.Sprintf(format, args...)) + if l := g.logger.Load(); l != nil { + l.logger.Error(fmt.Sprintf(format, args...)) + } os.Exit(1) } func (g *GRPCLogFaker) Fatalln(args ...any) { - g.Logger.Error(fmt.Sprintln(args...)) + if l := g.logger.Load(); l != nil { + l.logger.Error(fmt.Sprintln(args...)) + } os.Exit(1) } func (g *GRPCLogFaker) Info(args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprint(args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprint(args...)) + } + } } } func (g *GRPCLogFaker) Infof(format string, args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprintf(format, args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprintf(format, args...)) + } + } } } func (g *GRPCLogFaker) Infoln(args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprintln(args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprintln(args...)) + } + } } } func (g *GRPCLogFaker) Warning(args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprint(args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprint(args...)) + } + } } } func (g *GRPCLogFaker) Warningf(format string, args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprintf(format, args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprintf(format, args...)) + } + } } } func (g *GRPCLogFaker) Warningln(args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprintln(args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprintln(args...)) + } + } } } func (g *GRPCLogFaker) Error(args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprint(args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprint(args...)) + } + } } } func (g *GRPCLogFaker) Errorf(format string, args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprintf(format, args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprintf(format, args...)) + } + } } } func (g *GRPCLogFaker) Errorln(args ...any) { - if g.Log && g.Logger.IsDebug() { - g.Logger.Debug(fmt.Sprintln(args...)) + if g.loggingEnabled.Load() { + if l := g.logger.Load(); l != nil { + if l.logger.IsDebug() { + l.logger.Debug(fmt.Sprintln(args...)) + } + } } } diff --git a/internal/cmd/base/servers.go b/internal/cmd/base/servers.go index 2cac17276..dc5f88df2 100644 --- a/internal/cmd/base/servers.go +++ b/internal/cmd/base/servers.go @@ -50,7 +50,6 @@ import ( "github.com/prometheus/client_golang/prometheus" "golang.org/x/text/cases" "golang.org/x/text/language" - "google.golang.org/grpc/grpclog" ) const ( @@ -285,10 +284,8 @@ func (b *Server) SetupLogging(flagLogLevel, flagLogFormat, configLogLevel, confi // create GRPC logger namedGRPCLogFaker := b.Logger.Named("grpclogfaker") - grpclog.SetLoggerV2(&GRPCLogFaker{ - Logger: namedGRPCLogFaker, - Log: os.Getenv("BOUNDARY_GRPC_LOGGING") != "", - }) + grpcLogFaker.SetLogOnOff(os.Getenv("BOUNDARY_GRPC_LOGGING") != "") + grpcLogFaker.SetLogger(namedGRPCLogFaker) b.Info["log level"] = logLevel.String() b.InfoKeys = append(b.InfoKeys, "log level") diff --git a/internal/credential/service_list_credentials_ext_test.go b/internal/credential/service_list_credentials_ext_test.go index cfd684023..b2cc0a326 100644 --- a/internal/credential/service_list_credentials_ext_test.go +++ b/internal/credential/service_list_credentials_ext_test.go @@ -182,7 +182,7 @@ func TestService_List(t *testing.T) { filterFunc := func(_ context.Context, c credential.Static) (bool, error) { return true, nil } - _, err = credential.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId()) + _, err := credential.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { @@ -270,7 +270,7 @@ func TestService_List(t *testing.T) { filterFunc := func(_ context.Context, c credential.Static) (bool, error) { return true, nil } - _, err = credential.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId()) + _, err := credential.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("nil repo", func(t *testing.T) { diff --git a/internal/credential/service_list_libraries_ext_test.go b/internal/credential/service_list_libraries_ext_test.go index ac759b4a2..27235f932 100644 --- a/internal/credential/service_list_libraries_ext_test.go +++ b/internal/credential/service_list_libraries_ext_test.go @@ -331,7 +331,7 @@ func TestLibraryService_List(t *testing.T) { filterFunc := func(_ context.Context, l credential.Library) (bool, error) { return true, nil } - _, err = credential.ListLibrariesRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId()) + _, err := credential.ListLibrariesRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, credStore.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/credential/service_list_stores_ext_test.go b/internal/credential/service_list_stores_ext_test.go index dc2c0ed00..86120489f 100644 --- a/internal/credential/service_list_stores_ext_test.go +++ b/internal/credential/service_list_stores_ext_test.go @@ -370,7 +370,7 @@ func TestStoreService_List(t *testing.T) { filterFunc := func(_ context.Context, s credential.Store) (bool, error) { return true, nil } - _, err = credential.ListStoresRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId}) + _, err := credential.ListStoresRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId}) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/daemon/cluster/handlers/testing.go b/internal/daemon/cluster/handlers/testing.go index ded05ae5d..4901d055d 100644 --- a/internal/daemon/cluster/handlers/testing.go +++ b/internal/daemon/cluster/handlers/testing.go @@ -87,12 +87,12 @@ func TestRegisterHandlerFn(t *testing.T, msgType pbs.MsgType, h UpstreamMessageH return func(t *testing.T) { t.Helper() testCtx := context.Background() - var cpHandlerRegistry sync.Map + cpHandlerRegistry := new(sync.Map) upstreamMessageHandler.Range(func(k, v interface{}) bool { cpHandlerRegistry.Store(k, v) return true }) - var cpTypeSpecifierRegister sync.Map + cpTypeSpecifierRegister := new(sync.Map) upstreamMessageTypeSpecifier.Range(func(k, v interface{}) bool { cpTypeSpecifierRegister.Store(k, v) return true diff --git a/internal/daemon/cluster/handlers/upstream_message_service_controller.go b/internal/daemon/cluster/handlers/upstream_message_service_controller.go index 9626ff079..111f78ed8 100644 --- a/internal/daemon/cluster/handlers/upstream_message_service_controller.go +++ b/internal/daemon/cluster/handlers/upstream_message_service_controller.go @@ -19,7 +19,7 @@ import ( "google.golang.org/protobuf/proto" ) -var upstreamMessageHandler sync.Map +var upstreamMessageHandler *sync.Map = new(sync.Map) // UpstreamMessageHandler defines a handler for an UpstreamMessageRequest(s). // diff --git a/internal/daemon/cluster/handlers/upstream_message_service_test.go b/internal/daemon/cluster/handlers/upstream_message_service_test.go index 34bb1382e..5d19c8335 100644 --- a/internal/daemon/cluster/handlers/upstream_message_service_test.go +++ b/internal/daemon/cluster/handlers/upstream_message_service_test.go @@ -65,7 +65,7 @@ func Test_RegisterUpstreamMessageHandler(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { assert, require := assert.New(t), require.New(t) - var cp sync.Map + cp := new(sync.Map) upstreamMessageHandler.Range(func(k, v interface{}) bool { cp.Store(k, v) return true @@ -468,7 +468,7 @@ func Test_RegisterUpstreamMessageTypeSpecifier(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { assert, require := assert.New(t), require.New(t) - var cp sync.Map + cp := new(sync.Map) upstreamMessageTypeSpecifier.Range(func(k, v interface{}) bool { cp.Store(k, v) return true diff --git a/internal/daemon/cluster/handlers/upstream_message_service_worker.go b/internal/daemon/cluster/handlers/upstream_message_service_worker.go index 809b90356..2b4631fe4 100644 --- a/internal/daemon/cluster/handlers/upstream_message_service_worker.go +++ b/internal/daemon/cluster/handlers/upstream_message_service_worker.go @@ -82,7 +82,7 @@ type UpstreamMessageTypeSpecifier interface { AllocResponse() proto.Message } -var upstreamMessageTypeSpecifier sync.Map +var upstreamMessageTypeSpecifier *sync.Map = new(sync.Map) // registerUpstreamMessageTypeSpecifier will register an // UpstreamMessageTypeSpecifier for the specified msg name. diff --git a/internal/daemon/worker/auth_rotation_test.go b/internal/daemon/worker/auth_rotation_test.go index 3f4deee90..2cf8e8d24 100644 --- a/internal/daemon/worker/auth_rotation_test.go +++ b/internal/daemon/worker/auth_rotation_test.go @@ -170,8 +170,8 @@ func TestRotationTicking(t *testing.T) { // Stop and start the client connections to ensure the new credentials // are valid; if not, we won't establish a new connection and rotation // will fail - require.NotNil(w.Worker().GrpcClientConn) - require.NoError(w.Worker().GrpcClientConn.Close()) + require.NotNil(w.Worker().GrpcClientConn.Load()) + require.NoError(w.Worker().GrpcClientConn.Load().Close()) require.NoError(w.Worker().StartControllerConnections()) rotationCount++ } diff --git a/internal/daemon/worker/controller_connection.go b/internal/daemon/worker/controller_connection.go index 9321b6f41..b3aeb4ee2 100644 --- a/internal/daemon/worker/controller_connection.go +++ b/internal/daemon/worker/controller_connection.go @@ -189,7 +189,7 @@ func (w *Worker) createClientConn(addr string) error { return fmt.Errorf("error dialing controller for worker auth: %w", err) } - w.GrpcClientConn = cc + w.GrpcClientConn.Store(cc) w.controllerMultihopConn.Store(multihop.NewMultihopServiceClient(cc)) var producer handlers.UpstreamMessageServiceClientProducer diff --git a/internal/daemon/worker/controller_connection_test.go b/internal/daemon/worker/controller_connection_test.go index 2baa7d302..d90e35b18 100644 --- a/internal/daemon/worker/controller_connection_test.go +++ b/internal/daemon/worker/controller_connection_test.go @@ -6,6 +6,7 @@ package worker import ( "context" "fmt" + "net" "strconv" "sync/atomic" "testing" @@ -119,8 +120,9 @@ func waitForConnectionStateCondition(upstreamConnectionState *atomic.Value, expe } type serverTestInfo struct { - srv *grpc.Server - address string + srv *grpc.Server + address string + listener net.Listener } func createTestServers(t *testing.T) ([]*serverTestInfo, error) { @@ -133,10 +135,12 @@ func createTestServers(t *testing.T) ([]*serverTestInfo, error) { return nil, err } srv := grpc.NewServer() - lInfo := &serverTestInfo{srv: srv, address: listener.Addr().String()} + lInfo := &serverTestInfo{srv: srv, listener: listener, address: listener.Addr().String()} servers = append(servers, lInfo) - go func(i int) { - if err := servers[i].srv.Serve(listener); err != nil { + } + for i := range servers { + go func(num int) { + if err := servers[num].srv.Serve(servers[num].listener); err != nil { t.Logf("error serving: %v", err) } }(i) diff --git a/internal/daemon/worker/countingconn_test.go b/internal/daemon/worker/countingconn_test.go index 729dd3e70..b53cdd02b 100644 --- a/internal/daemon/worker/countingconn_test.go +++ b/internal/daemon/worker/countingconn_test.go @@ -7,6 +7,7 @@ import ( "fmt" "net" "sync" + "sync/atomic" "testing" "github.com/stretchr/testify/require" @@ -20,54 +21,29 @@ func TestCountingConn(t *testing.T) { underlyingConn *testNetConn }{ { - name: "noErrors", - writeBytes: []byte("hello"), - underlyingConn: &testNetConn{ - bytesToRead: 100, - readErr: false, - writeErr: false, - closeErr: false, - }, + name: "noErrors", + writeBytes: []byte("hello"), + underlyingConn: newTestNetConn(100, false, false, false), }, { - name: "readErr", - writeBytes: []byte("hello"), - underlyingConn: &testNetConn{ - bytesToRead: 100, - readErr: true, - writeErr: false, - closeErr: false, - }, + name: "readErr", + writeBytes: []byte("hello"), + underlyingConn: newTestNetConn(100, true, false, false), }, { - name: "writeErr", - writeBytes: []byte("hello"), - underlyingConn: &testNetConn{ - bytesToRead: 100, - readErr: false, - writeErr: true, - closeErr: false, - }, + name: "writeErr", + writeBytes: []byte("hello"), + underlyingConn: newTestNetConn(100, false, true, false), }, { - name: "closeErr", - writeBytes: []byte("hello"), - underlyingConn: &testNetConn{ - bytesToRead: 100, - readErr: false, - writeErr: false, - closeErr: true, - }, + name: "closeErr", + writeBytes: []byte("hello"), + underlyingConn: newTestNetConn(100, false, false, true), }, { - name: "allErr", - writeBytes: []byte("hello"), - underlyingConn: &testNetConn{ - bytesToRead: 100, - readErr: true, - writeErr: true, - closeErr: true, - }, + name: "allErr", + writeBytes: []byte("hello"), + underlyingConn: newTestNetConn(100, true, true, true), }, } @@ -77,7 +53,7 @@ func TestCountingConn(t *testing.T) { readBytes := make([]byte, tt.underlyingConn.bytesToRead) read, err := conn.Read(readBytes) - require.True(t, tt.underlyingConn.readCalled) + require.True(t, tt.underlyingConn.readCalled.Load()) if tt.underlyingConn.readErr { require.Error(t, err) require.Equal(t, 1, read) @@ -90,7 +66,7 @@ func TestCountingConn(t *testing.T) { } written, err := conn.Write(tt.writeBytes) - require.True(t, tt.underlyingConn.writeCalled) + require.True(t, tt.underlyingConn.writeCalled.Load()) if tt.underlyingConn.writeErr { require.Error(t, err) require.Equal(t, 1, written) @@ -102,7 +78,7 @@ func TestCountingConn(t *testing.T) { } err = conn.Close() - require.True(t, tt.underlyingConn.closeCalled) + require.True(t, tt.underlyingConn.closeCalled.Load()) if tt.underlyingConn.closeErr { require.Error(t, err) } else { @@ -121,7 +97,7 @@ func TestCountingConnConcurrentCalls(t *testing.T) { concurrentReads := 1000 concurrentWrites := 1000 - conn := &countingConn{Conn: &testNetConn{bytesToRead: bytesToRead}} + conn := &countingConn{Conn: newTestNetConn(bytesToRead, false, false, false)} wg := sync.WaitGroup{} wg.Add(concurrentReads) @@ -164,13 +140,25 @@ type testNetConn struct { closeErr bool // Test results - readCalled bool - writeCalled bool - closeCalled bool + readCalled *atomic.Bool + writeCalled *atomic.Bool + closeCalled *atomic.Bool +} + +func newTestNetConn(bytesToRead int, readErr, writeErr, closeErr bool) *testNetConn { + return &testNetConn{ + bytesToRead: bytesToRead, + readErr: readErr, + writeErr: writeErr, + closeErr: closeErr, + readCalled: new(atomic.Bool), + writeCalled: new(atomic.Bool), + closeCalled: new(atomic.Bool), + } } func (t *testNetConn) Read(in []byte) (int, error) { - t.readCalled = true + t.readCalled.Store(true) if t.readErr { return 1, fmt.Errorf("oops, read error") } @@ -183,7 +171,7 @@ func (t *testNetConn) Read(in []byte) (int, error) { } func (t *testNetConn) Write(in []byte) (int, error) { - t.writeCalled = true + t.writeCalled.Store(true) if t.writeErr { return 1, fmt.Errorf("oops, write error") } @@ -192,7 +180,7 @@ func (t *testNetConn) Write(in []byte) (int, error) { } func (t *testNetConn) Close() error { - t.closeCalled = true + t.closeCalled.Store(true) if t.closeErr { return fmt.Errorf("oops, close error") } diff --git a/internal/daemon/worker/listeners_test.go b/internal/daemon/worker/listeners_test.go index 7a81fc41d..6a86914e9 100644 --- a/internal/daemon/worker/listeners_test.go +++ b/internal/daemon/worker/listeners_test.go @@ -89,7 +89,7 @@ func TestStartListeners(t *testing.T) { } w.controllerUpstreamMsgConn.Store(&dummyClientProducer) - manager, err := session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn)) + manager, err := session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn.Load())) require.NoError(t, err) err = w.startListeners(manager) if tt.expErr { diff --git a/internal/daemon/worker/proxy/proxy.go b/internal/daemon/worker/proxy/proxy.go index f6c17aa08..77dbca71c 100644 --- a/internal/daemon/worker/proxy/proxy.go +++ b/internal/daemon/worker/proxy/proxy.go @@ -17,7 +17,7 @@ var ( TcpHandlerName = "tcp" // handlers is the map of registered handlers - handlers sync.Map + handlers *sync.Map = new(sync.Map) // ErrUnknownProtocol specifies the provided protocol has no registered handler ErrUnknownProtocol = errors.New("proxy: handler not found for protocol") diff --git a/internal/daemon/worker/proxy/proxy_conn_tracker.go b/internal/daemon/worker/proxy/proxy_conn_tracker.go index f6965adb7..417b460ea 100644 --- a/internal/daemon/worker/proxy/proxy_conn_tracker.go +++ b/internal/daemon/worker/proxy/proxy_conn_tracker.go @@ -9,10 +9,10 @@ import ( ) // ProxyState contains the current state of proxies in this process. -var ProxyState proxyState +var ProxyState proxyState = proxyState{proxyCount: new(atomic.Int64)} type proxyState struct { - proxyCount atomic.Int64 + proxyCount *atomic.Int64 } // CurrentProxiedConnections returns the current number of ongoing proxied diff --git a/internal/daemon/worker/proxy/proxy_conn_tracker_test.go b/internal/daemon/worker/proxy/proxy_conn_tracker_test.go index bb01c1292..11a1ec60f 100644 --- a/internal/daemon/worker/proxy/proxy_conn_tracker_test.go +++ b/internal/daemon/worker/proxy/proxy_conn_tracker_test.go @@ -6,6 +6,7 @@ package proxy import ( "net/http" "net/http/httptest" + "sync/atomic" "testing" "github.com/stretchr/testify/assert" @@ -14,7 +15,7 @@ import ( func TestProxyStateHelpers(t *testing.T) { old := ProxyState - ProxyState = proxyState{} + ProxyState = proxyState{proxyCount: new(atomic.Int64)} t.Cleanup(func() { ProxyState = old }) diff --git a/internal/daemon/worker/proxy/proxy_test.go b/internal/daemon/worker/proxy/proxy_test.go index 9159f594c..5de35018f 100644 --- a/internal/daemon/worker/proxy/proxy_test.go +++ b/internal/daemon/worker/proxy/proxy_test.go @@ -24,7 +24,7 @@ func TestRegisterHandler(t *testing.T) { t.Cleanup(func() { handlers = oldHandler }) - handlers = sync.Map{} + handlers = new(sync.Map) err := RegisterHandler("protocol", fn) require.NoError(err) @@ -47,7 +47,7 @@ func TestAlwaysTcpGetHandler(t *testing.T) { t.Cleanup(func() { handlers = oldHandler }) - handlers = sync.Map{} + handlers = new(sync.Map) _, err := tcpOnly("wid", nil) assert.ErrorIs(err, ErrUnknownProtocol) diff --git a/internal/daemon/worker/rpc_registration.go b/internal/daemon/worker/rpc_registration.go index cdad537d1..61c387e9f 100644 --- a/internal/daemon/worker/rpc_registration.go +++ b/internal/daemon/worker/rpc_registration.go @@ -37,7 +37,7 @@ func registerWorkerStatusSessionService(ctx context.Context, w *Worker, server * return fmt.Errorf("%s: server is nil", op) } - statusSessionService := NewWorkerProxyServiceServer(w.GrpcClientConn) + statusSessionService := NewWorkerProxyServiceServer(w.GrpcClientConn.Load()) pbs.RegisterServerCoordinationServiceServer(server, statusSessionService) pbs.RegisterSessionServiceServer(server, statusSessionService) return nil diff --git a/internal/daemon/worker/session/manager.go b/internal/daemon/worker/session/manager.go index 81b415d6f..ba356af7e 100644 --- a/internal/daemon/worker/session/manager.go +++ b/internal/daemon/worker/session/manager.go @@ -61,7 +61,7 @@ type Manager interface { type manager struct { controllerSessionConn pbs.SessionServiceClient - sessionMap sync.Map + sessionMap *sync.Map } var _ Manager = (*manager)(nil) @@ -74,6 +74,7 @@ func NewManager(client pbs.SessionServiceClient) (*manager, error) { } return &manager{ controllerSessionConn: client, + sessionMap: new(sync.Map), }, nil } diff --git a/internal/daemon/worker/session/manager_test.go b/internal/daemon/worker/session/manager_test.go index e42359641..720872fbc 100644 --- a/internal/daemon/worker/session/manager_test.go +++ b/internal/daemon/worker/session/manager_test.go @@ -273,7 +273,7 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) { cases := []struct { name string sessionCloseInfo map[string][]*pbs.CloseConnectionResponseData - sessionInfoMap func() sync.Map + sessionInfoMap func() *sync.Map expected []string expectedClosed map[string]struct{} expectedErr []error @@ -288,8 +288,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) { {ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED}, }, }, - sessionInfoMap: func() sync.Map { - var m sync.Map + sessionInfoMap: func() *sync.Map { + m := new(sync.Map) m.Store("one", &sess{ resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{ SessionId: "one", @@ -333,8 +333,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) { {ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED}, }, }, - sessionInfoMap: func() sync.Map { - var m sync.Map + sessionInfoMap: func() *sync.Map { + m := new(sync.Map) m.Store("one", &sess{ resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{ SessionId: "one", @@ -369,8 +369,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) { {ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED}, }, }, - sessionInfoMap: func() sync.Map { - var m sync.Map + sessionInfoMap: func() *sync.Map { + m := new(sync.Map) m.Store("one", &sess{ resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{ SessionId: "one", @@ -400,8 +400,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) { {ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED}, }, }, - sessionInfoMap: func() sync.Map { - var m sync.Map + sessionInfoMap: func() *sync.Map { + m := new(sync.Map) m.Store("one", &sess{ resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{ SessionId: "one", @@ -431,8 +431,8 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) { { name: "empty", sessionCloseInfo: make(map[string][]*pbs.CloseConnectionResponseData), - sessionInfoMap: func() sync.Map { - var m sync.Map + sessionInfoMap: func() *sync.Map { + m := new(sync.Map) m.Store("one", &sess{ resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{ SessionId: "one", @@ -453,7 +453,7 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) { tc := tc t.Run(tc.name, func(t *testing.T) { require := require.New(t) - manager := &manager{} + manager := &manager{sessionMap: new(sync.Map)} manager.sessionMap = tc.sessionInfoMap() actual, actualErr := setCloseTimeForResponse(manager, tc.sessionCloseInfo) diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index ae9c86143..7274823fd 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -181,7 +181,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess }) // Send status information - client := pbs.NewServerCoordinationServiceClient(w.GrpcClientConn) + client := pbs.NewServerCoordinationServiceClient(w.GrpcClientConn.Load()) var tags []*pb.TagPair // If we're not going to request a tag update, no reason to have these // marshaled on every status call. @@ -242,7 +242,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess // In the case that the control plane has gone down and come up with different IPs, // append initial upstreams/ cluster addr to the resolver to try - if w.GrpcClientConn.GetState() == connectivity.TransientFailure { + if w.GrpcClientConn.Load().GetState() == connectivity.TransientFailure { lastStatus := w.lastStatusSuccess.Load().(*LastStatusInformation) if lastStatus != nil && lastStatus.LastCalculatedUpstreams != nil { addrs := lastStatus.LastCalculatedUpstreams diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index f32aca490..093004a60 100644 --- a/internal/daemon/worker/worker.go +++ b/internal/daemon/worker/worker.go @@ -116,10 +116,11 @@ type Worker struct { tickerWg sync.WaitGroup - // grpc.ClientConns are thread safe. - // See https://github.com/grpc/grpc-go/blob/master/Documentation/concurrency.md#clients - // This is exported for tests. - GrpcClientConn *grpc.ClientConn + // grpc.ClientConns are thread safe. See + // https://github.com/grpc/grpc-go/blob/master/Documentation/concurrency.md#clients + // However this is an atomic because we sometimes swap this pointer out + // (mostly in tests) - which isn't thread safe. This is exported for tests. + GrpcClientConn atomic.Pointer[grpc.ClientConn] // receives address updates and contains the grpc resolver. addressReceivers []addressReceiver @@ -511,7 +512,7 @@ func (w *Worker) Start() error { return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error making controller connections")) } - w.sessionManager, err = session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn)) + w.sessionManager, err = session.NewManager(pbs.NewSessionServiceClient(w.GrpcClientConn.Load())) if err != nil { return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error creating session manager")) } diff --git a/internal/daemon/worker/workerdisconnect_test.go b/internal/daemon/worker/workerdisconnect_test.go index 7891fdfaf..7d39d1d01 100644 --- a/internal/daemon/worker/workerdisconnect_test.go +++ b/internal/daemon/worker/workerdisconnect_test.go @@ -82,7 +82,7 @@ func TestDeleteConnectedWorkers(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - prevState := tc.testWorker.Worker().GrpcClientConn.GetState() + prevState := tc.testWorker.Worker().GrpcClientConn.Load().GetState() require.NotEqual(t, connectivity.TransientFailure, prevState) require.NotEqual(t, connectivity.Shutdown, prevState) _, err = serverRepo.DeleteWorker(ctx, tc.workerId) @@ -90,12 +90,12 @@ func TestDeleteConnectedWorkers(t *testing.T) { stateChangeCtx, cancel := context.WithTimeout(ctx, 4*time.Second) defer cancel() for { - tc.testWorker.Worker().GrpcClientConn.ResetConnectBackoff() - if !tc.testWorker.Worker().GrpcClientConn.WaitForStateChange(stateChangeCtx, prevState) { + tc.testWorker.Worker().GrpcClientConn.Load().ResetConnectBackoff() + if !tc.testWorker.Worker().GrpcClientConn.Load().WaitForStateChange(stateChangeCtx, prevState) { assert.Fail(t, "State didn't change before context timed out") break } - newState := tc.testWorker.Worker().GrpcClientConn.GetState() + newState := tc.testWorker.Worker().GrpcClientConn.Load().GetState() t.Logf("Changed from previous state: %s to new state: %s", prevState, newState) if newState == connectivity.Shutdown || newState == connectivity.TransientFailure { break diff --git a/internal/event/eventer.go b/internal/event/eventer.go index 55c87e073..f4761480e 100644 --- a/internal/event/eventer.go +++ b/internal/event/eventer.go @@ -118,7 +118,8 @@ var ( ) func defaultLogger() hclog.Logger { - opts := hclog.DefaultOptions + opts := new(hclog.LoggerOptions) + *opts = *hclog.DefaultOptions opts.JSONFormat = true return hclog.New(opts) } diff --git a/internal/host/plugin/service_list_host_sets_ext_test.go b/internal/host/plugin/service_list_host_sets_ext_test.go index 9cbba357a..4367a0b1c 100644 --- a/internal/host/plugin/service_list_host_sets_ext_test.go +++ b/internal/host/plugin/service_list_host_sets_ext_test.go @@ -169,7 +169,7 @@ func TestService_ListHostSets(t *testing.T) { filterFunc := func(_ context.Context, s host.Set, plg *plugin.Plugin) (bool, error) { return true, nil } - _, _, err = hostplugin.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId()) + _, _, err := hostplugin.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId()) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/host/service_list_catalogs_ext_test.go b/internal/host/service_list_catalogs_ext_test.go index 2332a35cf..2e5760521 100644 --- a/internal/host/service_list_catalogs_ext_test.go +++ b/internal/host/service_list_catalogs_ext_test.go @@ -209,7 +209,7 @@ func TestCatalogService_List(t *testing.T) { filterFunc := func(_ context.Context, c host.Catalog, plgs map[string]*plugin.Plugin) (bool, error) { return true, nil } - _, _, err = host.ListCatalogsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId}) + _, _, err := host.ListCatalogsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId}) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { @@ -297,7 +297,7 @@ func TestCatalogService_List(t *testing.T) { filterFunc := func(_ context.Context, c host.Catalog, plgs map[string]*plugin.Plugin) (bool, error) { return true, nil } - _, _, err = host.ListCatalogsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId}) + _, _, err := host.ListCatalogsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{prj.PublicId}) require.ErrorContains(t, err, "missing token") }) t.Run("missing repo", func(t *testing.T) { diff --git a/internal/iam/service_list_ext_test.go b/internal/iam/service_list_ext_test.go index 8e268dfa6..83ed3814c 100644 --- a/internal/iam/service_list_ext_test.go +++ b/internal/iam/service_list_ext_test.go @@ -839,7 +839,7 @@ func TestService_ListUsers(t *testing.T) { filterFunc := func(_ context.Context, r *iam.User) (bool, error) { return true, nil } - _, err = iam.ListUsersRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()}) + _, err := iam.ListUsersRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, []string{org.GetPublicId()}) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { @@ -1368,7 +1368,7 @@ func TestService_ListGroups(t *testing.T) { filterFunc := func(_ context.Context, r *iam.Group) (bool, error) { return true, nil } - _, err = iam.ListGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, relevantScopes) + _, err := iam.ListGroupsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, relevantScopes) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { @@ -1456,7 +1456,7 @@ func TestService_ListGroups(t *testing.T) { filterFunc := func(_ context.Context, r *iam.Group) (bool, error) { return true, nil } - _, err = iam.ListGroupsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, relevantScopes) + _, err := iam.ListGroupsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, relevantScopes) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/perms/grants.go b/internal/perms/grants.go index 56ad5f372..b4f6ebbed 100644 --- a/internal/perms/grants.go +++ b/internal/perms/grants.go @@ -178,7 +178,7 @@ func (g Grant) CanonicalString() string { } // MarshalJSON provides a custom marshaller for grants -func (g Grant) MarshalJSON(ctx context.Context) ([]byte, error) { +func (g Grant) MarshalJSON() ([]byte, error) { const op = "perms.(Grant).MarshalJSON" res := make(map[string]any, 4) if g.id != "" { @@ -203,7 +203,7 @@ func (g Grant) MarshalJSON(ctx context.Context) ([]byte, error) { } b, err := json.Marshal(res) if err != nil { - return nil, errors.Wrap(ctx, err, op, errors.WithCode(errors.Encode)) + return nil, fmt.Errorf("%s: error marshaling grant: %w", op, err) } return b, nil } diff --git a/internal/perms/grants_test.go b/internal/perms/grants_test.go index e5dbe13df..40c6b4178 100644 --- a/internal/perms/grants_test.go +++ b/internal/perms/grants_test.go @@ -132,8 +132,6 @@ func Test_ValidateType(t *testing.T) { func Test_MarshalingAndCloning(t *testing.T) { t.Parallel() - ctx := context.Background() - type input struct { name string input Grant @@ -278,7 +276,7 @@ func Test_MarshalingAndCloning(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - output, err := test.input.MarshalJSON(ctx) + output, err := test.input.MarshalJSON() require.NoError(t, err) assert.Equal(t, test.jsonOutput, string(output)) assert.Equal(t, test.canonicalString, test.input.CanonicalString()) @@ -1342,7 +1340,7 @@ func FuzzParse(f *testing.F) { if g.CanonicalString() != g2.CanonicalString() { t.Errorf("grant roundtrip failed, input %q, output %q", g.CanonicalString(), g2.CanonicalString()) } - jsonBytes, err := g.MarshalJSON(ctx) + jsonBytes, err := g.MarshalJSON() if err != nil { t.Error("Failed to marshal JSON:", err) } diff --git a/internal/plugin/loopback/testing_grpc_stream_test.go b/internal/plugin/loopback/testing_grpc_stream_test.go index 7d9be4aff..21b46b04d 100644 --- a/internal/plugin/loopback/testing_grpc_stream_test.go +++ b/internal/plugin/loopback/testing_grpc_stream_test.go @@ -415,7 +415,7 @@ func Test_GetObjectStream(t *testing.T) { }, time.Second*10, time.Millisecond*50) require.Eventually(func() bool { - return stream.streamClosed && stream.IsStreamClosed() + return stream.IsStreamClosed() }, time.Second*10, time.Millisecond*50) }) } diff --git a/internal/session/service_list_ext_test.go b/internal/session/service_list_ext_test.go index e1033313f..84a1bcf67 100644 --- a/internal/session/service_list_ext_test.go +++ b/internal/session/service_list_ext_test.go @@ -155,7 +155,7 @@ func TestService_List(t *testing.T) { filterFunc := func(_ context.Context, s *session.Session) (bool, error) { return true, nil } - _, err = session.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true) + _, err := session.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, true) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { diff --git a/internal/target/service_list_ext_test.go b/internal/target/service_list_ext_test.go index d84c210d4..83d8ac952 100644 --- a/internal/target/service_list_ext_test.go +++ b/internal/target/service_list_ext_test.go @@ -160,7 +160,7 @@ func TestService_List(t *testing.T) { filterFunc := func(_ context.Context, t target.Target) (bool, error) { return true, nil } - _, err = target.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo) + _, err := target.ListPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) { @@ -238,7 +238,7 @@ func TestService_List(t *testing.T) { filterFunc := func(_ context.Context, t target.Target) (bool, error) { return true, nil } - _, err = target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo) + _, err := target.ListRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo) require.ErrorContains(t, err, "missing token") }) t.Run("wrong token type", func(t *testing.T) {