frostfs-sdk-go/pool/tree/pool_test.go
Dmitrii Stepanov 56debcfa56
All checks were successful
DCO / DCO (pull_request) Successful in 56s
Tests and linters / Tests (1.21) (pull_request) Successful in 1m16s
Tests and linters / Tests (1.20) (pull_request) Successful in 1m13s
Tests and linters / Lint (pull_request) Successful in 2m20s
[#190] sdk-go: Gofumpt fixes
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
2023-11-22 19:21:20 +03:00

356 lines
7.9 KiB
Go

package tree
import (
"context"
"errors"
"testing"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/pool"
grpcService "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/pool/tree/service"
"github.com/stretchr/testify/require"
"go.uber.org/zap/zaptest"
)
type treeClientMock struct {
address string
err bool
}
func (t *treeClientMock) serviceClient() (grpcService.TreeServiceClient, error) {
if t.err {
return nil, errors.New("serviceClient() mock error")
}
return nil, nil
}
func (t *treeClientMock) endpoint() string {
return t.address
}
func (t *treeClientMock) isHealthy() bool {
return true
}
func (t *treeClientMock) setHealthy(bool) {
return
}
func (t *treeClientMock) dial(context.Context) error {
return nil
}
func (t *treeClientMock) redialIfNecessary(context.Context) (bool, error) {
if t.err {
return false, errors.New("redialIfNecessary() mock error")
}
return false, nil
}
func (t *treeClientMock) close() error {
return nil
}
func TestHandleError(t *testing.T) {
defaultError := errors.New("default error")
for _, tc := range []struct {
err error
expectedError error
}{
{
err: defaultError,
expectedError: defaultError,
},
{
err: errors.New("something not found"),
expectedError: ErrNodeNotFound,
},
{
err: errors.New("something is denied by some acl rule"),
expectedError: ErrNodeAccessDenied,
},
} {
t.Run("", func(t *testing.T) {
err := handleError("err message", tc.err)
require.True(t, errors.Is(err, tc.expectedError))
})
}
}
func TestRetry(t *testing.T) {
ctx := context.Background()
nodes := [][]string{
{"node00", "node01", "node02", "node03"},
{"node10", "node11", "node12", "node13"},
}
var lenNodes int
for i := range nodes {
lenNodes += len(nodes[i])
}
p := &Pool{
logger: zaptest.NewLogger(t),
innerPools: makeInnerPool(nodes),
maxRequestAttempts: lenNodes,
}
makeFn := func(client grpcService.TreeServiceClient) error {
return nil
}
t.Run("first ok", func(t *testing.T) {
err := p.requestWithRetry(ctx, makeFn)
require.NoError(t, err)
checkIndicesAndReset(t, p, 0, 0)
})
t.Run("first failed", func(t *testing.T) {
setErrors(p, "node00")
err := p.requestWithRetry(ctx, makeFn)
require.NoError(t, err)
checkIndicesAndReset(t, p, 0, 1)
})
t.Run("all failed", func(t *testing.T) {
setErrors(p, nodes[0]...)
setErrors(p, nodes[1]...)
err := p.requestWithRetry(ctx, makeFn)
require.Error(t, err)
checkIndicesAndReset(t, p, 0, 0)
})
t.Run("round", func(t *testing.T) {
setErrors(p, nodes[0][0], nodes[0][1])
setErrors(p, nodes[1]...)
err := p.requestWithRetry(ctx, makeFn)
require.NoError(t, err)
checkIndices(t, p, 0, 2)
resetClientsErrors(p)
setErrors(p, nodes[0][2], nodes[0][3])
err = p.requestWithRetry(ctx, makeFn)
require.NoError(t, err)
checkIndicesAndReset(t, p, 0, 0)
})
t.Run("group switch", func(t *testing.T) {
setErrors(p, nodes[0]...)
setErrors(p, nodes[1][0])
err := p.requestWithRetry(ctx, makeFn)
require.NoError(t, err)
checkIndicesAndReset(t, p, 1, 1)
})
t.Run("group round", func(t *testing.T) {
setErrors(p, nodes[0][1:]...)
err := p.requestWithRetry(ctx, makeFn)
require.NoError(t, err)
checkIndicesAndReset(t, p, 0, 0)
})
t.Run("group round switch", func(t *testing.T) {
setErrors(p, nodes[0]...)
p.setStartIndices(0, 1)
err := p.requestWithRetry(ctx, makeFn)
require.NoError(t, err)
checkIndicesAndReset(t, p, 1, 0)
})
t.Run("no panic group switch", func(t *testing.T) {
setErrors(p, nodes[1]...)
p.setStartIndices(1, 0)
err := p.requestWithRetry(ctx, makeFn)
require.NoError(t, err)
checkIndicesAndReset(t, p, 0, 0)
})
t.Run("error empty result", func(t *testing.T) {
errNodes, index := 2, 0
err := p.requestWithRetry(ctx, func(client grpcService.TreeServiceClient) error {
if index < errNodes {
index++
return errNodeEmptyResult
}
return nil
})
require.NoError(t, err)
checkIndicesAndReset(t, p, 0, errNodes)
})
t.Run("error not found", func(t *testing.T) {
errNodes, index := 2, 0
err := p.requestWithRetry(ctx, func(client grpcService.TreeServiceClient) error {
if index < errNodes {
index++
return ErrNodeNotFound
}
return nil
})
require.NoError(t, err)
checkIndicesAndReset(t, p, 0, errNodes)
})
t.Run("error access denied", func(t *testing.T) {
var index int
err := p.requestWithRetry(ctx, func(client grpcService.TreeServiceClient) error {
index++
return ErrNodeAccessDenied
})
require.ErrorIs(t, err, ErrNodeAccessDenied)
require.Equal(t, 1, index)
checkIndicesAndReset(t, p, 0, 0)
})
t.Run("limit attempts", func(t *testing.T) {
oldVal := p.maxRequestAttempts
p.maxRequestAttempts = 2
setErrors(p, nodes[0]...)
setErrors(p, nodes[1]...)
err := p.requestWithRetry(ctx, makeFn)
require.Error(t, err)
checkIndicesAndReset(t, p, 0, 2)
p.maxRequestAttempts = oldVal
})
}
func TestRebalance(t *testing.T) {
nodes := [][]string{
{"node00", "node01"},
{"node10", "node11"},
}
p := &Pool{
logger: zaptest.NewLogger(t),
innerPools: makeInnerPool(nodes),
rebalanceParams: rebalanceParameters{
nodesGroup: makeNodesGroup(nodes),
},
}
ctx := context.Background()
buffers := makeBuffer(p)
t.Run("check dirty buffers", func(t *testing.T) {
p.updateNodesHealth(ctx, buffers)
checkIndices(t, p, 0, 0)
setErrors(p, nodes[0][0])
p.updateNodesHealth(ctx, buffers)
checkIndices(t, p, 0, 1)
resetClients(p)
})
t.Run("don't change healthy status", func(t *testing.T) {
p.updateNodesHealth(ctx, buffers)
checkIndices(t, p, 0, 0)
resetClients(p)
})
t.Run("switch to second group", func(t *testing.T) {
setErrors(p, nodes[0][0], nodes[0][1])
p.updateNodesHealth(ctx, buffers)
checkIndices(t, p, 1, 0)
resetClients(p)
})
t.Run("switch back and forth", func(t *testing.T) {
setErrors(p, nodes[0][0], nodes[0][1])
p.updateNodesHealth(ctx, buffers)
checkIndices(t, p, 1, 0)
p.updateNodesHealth(ctx, buffers)
checkIndices(t, p, 1, 0)
setNoErrors(p, nodes[0][0])
p.updateNodesHealth(ctx, buffers)
checkIndices(t, p, 0, 0)
resetClients(p)
})
}
func makeInnerPool(nodes [][]string) []*innerPool {
res := make([]*innerPool, len(nodes))
for i, group := range nodes {
res[i] = &innerPool{clients: make([]client, len(group))}
for j, node := range group {
res[i].clients[j] = &treeClientMock{address: node}
}
}
return res
}
func makeNodesGroup(nodes [][]string) [][]pool.NodeParam {
res := make([][]pool.NodeParam, len(nodes))
for i, group := range nodes {
res[i] = make([]pool.NodeParam, len(group))
for j, node := range group {
res[i][j] = pool.NewNodeParam(1, node, 1)
}
}
return res
}
func makeBuffer(p *Pool) [][]bool {
buffers := make([][]bool, len(p.rebalanceParams.nodesGroup))
for i, nodes := range p.rebalanceParams.nodesGroup {
buffers[i] = make([]bool, len(nodes))
}
return buffers
}
func checkIndicesAndReset(t *testing.T, p *Pool, iExp, jExp int) {
checkIndices(t, p, iExp, jExp)
resetClients(p)
}
func checkIndices(t *testing.T, p *Pool, iExp, jExp int) {
i, j := p.getStartIndices()
require.Equal(t, [2]int{iExp, jExp}, [2]int{i, j})
}
func resetClients(p *Pool) {
resetClientsErrors(p)
p.setStartIndices(0, 0)
}
func resetClientsErrors(p *Pool) {
for _, group := range p.innerPools {
for _, cl := range group.clients {
node := cl.(*treeClientMock)
node.err = false
}
}
}
func setErrors(p *Pool, nodes ...string) {
setErrorsBase(p, true, nodes...)
}
func setNoErrors(p *Pool, nodes ...string) {
setErrorsBase(p, false, nodes...)
}
func setErrorsBase(p *Pool, err bool, nodes ...string) {
for _, group := range p.innerPools {
for _, cl := range group.clients {
node := cl.(*treeClientMock)
if containsStr(nodes, node.address) {
node.err = err
}
}
}
}
func containsStr(list []string, item string) bool {
for i := range list {
if list[i] == item {
return true
}
}
return false
}