milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refine KV errors (#27588) 

Signed-off-by: yah01 <yah2er0ne@outlook.com>
pull/27657/head
yah01 2023-10-13 17:15:34 +08:00 committed by GitHub
parent d9431266bb
commit 15746b5b5f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 35 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
internal/kv/etcd/embed_etcd_kv.go
View File

@ -31,7 +31,6 @@ import (
"github.com/milvus-io/milvus/internal/kv"
"github.com/milvus-io/milvus/internal/kv/predicates"
"github.com/milvus-io/milvus/pkg/common"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/util/merr"
)
@ -131,6 +130,7 @@ func (kv *EmbedEtcdKV) LoadWithPrefix(key string) ([]string, []string, error) {
if err != nil {
return nil, nil, err
}
keys := make([]string, 0, resp.Count)
values := make([]string, 0, resp.Count)
for _, kv := range resp.Kvs {
@ -219,7 +219,7 @@ func (kv *EmbedEtcdKV) Load(key string) (string, error) {
return "", err
}
if resp.Count <= 0 {
return "", common.NewKeyNotExistError(key)
return "", merr.WrapErrIoKeyNotFound(key)
}
return string(resp.Kvs[0].Value), nil
@ -235,7 +235,7 @@ func (kv *EmbedEtcdKV) LoadBytes(key string) ([]byte, error) {
return nil, err
}
if resp.Count <= 0 {
return nil, common.NewKeyNotExistError(key)
return nil, merr.WrapErrIoKeyNotFound(key)
}
return resp.Kvs[0].Value, nil

7
internal/kv/etcd/etcd_kv.go
View File

@ -27,7 +27,6 @@ import (
"go.uber.org/zap"
"github.com/milvus-io/milvus/internal/kv/predicates"
"github.com/milvus-io/milvus/pkg/common"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/metrics"
"github.com/milvus-io/milvus/pkg/util/merr"
@ -207,7 +206,7 @@ func (kv *etcdKV) Load(key string) (string, error) {
return "", err
}
if resp.Count <= 0 {
return "", common.NewKeyNotExistError(key)
return "", merr.WrapErrIoKeyNotFound(key)
}
CheckElapseAndWarn(start, "Slow etcd operation load", zap.String("key", key))
return string(resp.Kvs[0].Value), nil
@ -221,10 +220,10 @@ func (kv *etcdKV) LoadBytes(key string) ([]byte, error) {
defer cancel()
resp, err := kv.getEtcdMeta(ctx, key)
if err != nil {
return []byte{}, err
return nil, err
}
if resp.Count <= 0 {
return []byte{}, common.NewKeyNotExistError(key)
return nil, merr.WrapErrIoKeyNotFound(key)
}
CheckElapseAndWarn(start, "Slow etcd operation load", zap.String("key", key))
return resp.Kvs[0].Value, nil

5
internal/kv/mem/mem_kv.go
View File

@ -23,7 +23,6 @@ import (
"github.com/google/btree"
"github.com/milvus-io/milvus/internal/kv/predicates"
"github.com/milvus-io/milvus/pkg/common"
"github.com/milvus-io/milvus/pkg/util/merr"
)
@ -84,7 +83,7 @@ func (kv *MemoryKV) Load(key string) (string, error) {
defer kv.RUnlock()
item := kv.tree.Get(memoryKVItem{key: key})
if item == nil {
return "", common.NewKeyNotExistError(key)
return "", merr.WrapErrIoKeyNotFound(key)
}
return item.(memoryKVItem).value.String(), nil
}
@ -95,7 +94,7 @@ func (kv *MemoryKV) LoadBytes(key string) ([]byte, error) {
defer kv.RUnlock()
item := kv.tree.Get(memoryKVItem{key: key})
if item == nil {
return []byte{}, common.NewKeyNotExistError(key)
return nil, merr.WrapErrIoKeyNotFound(key)
}
return item.(memoryKVItem).value.ByteSlice(), nil
}

7
internal/kv/tikv/txn_tikv.go
View File

@ -34,7 +34,6 @@ import (
"github.com/milvus-io/milvus/internal/kv"
"github.com/milvus-io/milvus/internal/kv/predicates"
"github.com/milvus-io/milvus/pkg/common"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/metrics"
"github.com/milvus-io/milvus/pkg/util/merr"
@ -141,7 +140,7 @@ func (kv *txnTiKV) Has(key string) (bool, error) {
_, err := kv.getTiKVMeta(ctx, key)
if err != nil {
// Dont error out if not present unless failed call to tikv
if common.IsKeyNotExistError(err) {
if errors.Is(err, merr.ErrIoKeyNotFound) {
return false, nil
}
loggingErr = errors.Wrap(err, fmt.Sprintf("Failed to read key: %s", key))
@ -199,7 +198,7 @@ func (kv *txnTiKV) Load(key string) (string, error) {
val, err := kv.getTiKVMeta(ctx, key)
if err != nil {
if common.IsKeyNotExistError(err) {
if errors.Is(err, merr.ErrIoKeyNotFound) {
loggingErr = err
} else {
loggingErr = errors.Wrap(err, fmt.Sprintf("Failed to read key %s", key))
@ -649,7 +648,7 @@ func (kv *txnTiKV) getTiKVMeta(ctx context.Context, key string) (string, error)
metrics.MetaOpCounter.WithLabelValues(metrics.MetaGetLabel, metrics.FailLabel).Inc()
if err == tikverr.ErrNotExist {
// If key is missing
return "", common.NewKeyNotExistError(key)
return "", merr.WrapErrIoKeyNotFound(key)
}
// If call to tikv fails
return "", errors.Wrap(err, fmt.Sprintf("Failed to get value for key %s in getTiKVMeta", key))

19
internal/metastore/kv/rootcoord/kv_catalog.go
View File

@ -5,6 +5,7 @@ import (
"encoding/json"
"fmt"
"github.com/cockroachdb/errors"
"github.com/golang/protobuf/proto"
"go.uber.org/zap"
@ -383,7 +384,7 @@ func (kc *Catalog) GetCredential(ctx context.Context, username string) (*model.C
k := fmt.Sprintf("%s/%s", CredentialPrefix, username)
v, err := kc.Txn.Load(k)
if err != nil {
if common.IsKeyNotExistError(err) {
if errors.Is(err, merr.ErrIoKeyNotFound) {
log.Debug("not found the user", zap.String("key", k))
} else {
log.Warn("get credential meta fail", zap.String("key", k), zap.Error(err))
@ -543,7 +544,7 @@ func (kc *Catalog) DropPartition(ctx context.Context, dbID int64, collectionID t
func (kc *Catalog) DropCredential(ctx context.Context, username string) error {
k := fmt.Sprintf("%s/%s", CredentialPrefix, username)
userResults, err := kc.ListUser(ctx, util.DefaultTenant, &milvuspb.UserEntity{Name: username}, true)
if err != nil && !common.IsKeyNotExistError(err) {
if err != nil && !errors.Is(err, merr.ErrIoKeyNotFound) {
log.Warn("fail to list user", zap.String("key", k), zap.Error(err))
return err
}
@ -720,7 +721,7 @@ func (kc *Catalog) ListCredentials(ctx context.Context) ([]string, error) {
func (kc *Catalog) save(k string) error {
var err error
if _, err = kc.Txn.Load(k); err != nil && !common.IsKeyNotExistError(err) {
if _, err = kc.Txn.Load(k); err != nil && !errors.Is(err, merr.ErrIoKeyNotFound) {
return err
}
if err == nil {
@ -732,10 +733,10 @@ func (kc *Catalog) save(k string) error {
func (kc *Catalog) remove(k string) error {
var err error
if _, err = kc.Txn.Load(k); err != nil && !common.IsKeyNotExistError(err) {
if _, err = kc.Txn.Load(k); err != nil && !errors.Is(err, merr.ErrIoKeyNotFound) {
return err
}
if err != nil && common.IsKeyNotExistError(err) {
if err != nil && errors.Is(err, merr.ErrIoKeyNotFound) {
return common.NewIgnorableError(fmt.Errorf("the key[%s] isn't existed", k))
}
return kc.Txn.Remove(k)
@ -753,7 +754,7 @@ func (kc *Catalog) CreateRole(ctx context.Context, tenant string, entity *milvus
func (kc *Catalog) DropRole(ctx context.Context, tenant string, roleName string) error {
k := funcutil.HandleTenantForEtcdKey(RolePrefix, tenant, roleName)
roleResults, err := kc.ListRole(ctx, tenant, &milvuspb.RoleEntity{Name: roleName}, true)
if err != nil && !common.IsKeyNotExistError(err) {
if err != nil && !errors.Is(err, merr.ErrIoKeyNotFound) {
log.Warn("fail to list role", zap.String("key", k), zap.Error(err))
return err
}
@ -962,12 +963,12 @@ func (kc *Catalog) AlterGrant(ctx context.Context, tenant string, entity *milvus
} else {
log.Warn("fail to load grant privilege entity", zap.String("key", k), zap.Any("type", operateType), zap.Error(err))
if funcutil.IsRevoke(operateType) {
if common.IsKeyNotExistError(err) {
if errors.Is(err, merr.ErrIoKeyNotFound) {
return common.NewIgnorableError(fmt.Errorf("the grant[%s] isn't existed", k))
}
return err
}
if !common.IsKeyNotExistError(err) {
if !errors.Is(err, merr.ErrIoKeyNotFound) {
return err
}
@ -983,7 +984,7 @@ func (kc *Catalog) AlterGrant(ctx context.Context, tenant string, entity *milvus
_, err = kc.Txn.Load(k)
if err != nil {
log.Warn("fail to load the grantee id", zap.String("key", k), zap.Error(err))
if !common.IsKeyNotExistError(err) {
if !errors.Is(err, merr.ErrIoKeyNotFound) {
log.Warn("fail to load the grantee id", zap.String("key", k), zap.Error(err))
return err
}

21
internal/metastore/kv/rootcoord/kv_catalog_test.go
View File

@ -29,6 +29,7 @@ import (
"github.com/milvus-io/milvus/pkg/util"
"github.com/milvus-io/milvus/pkg/util/crypto"
"github.com/milvus-io/milvus/pkg/util/funcutil"
"github.com/milvus-io/milvus/pkg/util/merr"
"github.com/milvus-io/milvus/pkg/util/typeutil"
)
@ -1481,7 +1482,7 @@ func TestRBAC_Role(t *testing.T) {
otherError = fmt.Errorf("mock load error")
)
kvmock.EXPECT().Load(notExistKey).Return("", common.NewKeyNotExistError(notExistKey)).Once()
kvmock.EXPECT().Load(notExistKey).Return("", merr.WrapErrIoKeyNotFound(notExistKey)).Once()
kvmock.EXPECT().Load(errorKey).Return("", otherError).Once()
kvmock.EXPECT().Load(mock.Anything).Return("", nil).Once()
kvmock.EXPECT().Remove(mock.Anything).Call.Return(nil).Once()
@ -1525,7 +1526,7 @@ func TestRBAC_Role(t *testing.T) {
otherError = fmt.Errorf("mock load error")
)
kvmock.EXPECT().Load(notExistKey).Return("", common.NewKeyNotExistError(notExistKey)).Once()
kvmock.EXPECT().Load(notExistKey).Return("", merr.WrapErrIoKeyNotFound(notExistKey)).Once()
kvmock.EXPECT().Load(errorKey).Return("", otherError).Once()
kvmock.EXPECT().Load(mock.Anything).Return("", nil).Once()
kvmock.EXPECT().Save(mock.Anything, mock.Anything).Call.Return(nil).Once()
@ -1572,7 +1573,7 @@ func TestRBAC_Role(t *testing.T) {
otherError = fmt.Errorf("mock load error")
)
kvmock.EXPECT().Load(notExistPath).Return("", common.NewKeyNotExistError(notExistName)).Once()
kvmock.EXPECT().Load(notExistPath).Return("", merr.WrapErrIoKeyNotFound(notExistName)).Once()
kvmock.EXPECT().Load(errorPath).Return("", otherError).Once()
kvmock.EXPECT().Load(mock.Anything).Return("", nil).Once()
kvmock.EXPECT().Save(mock.Anything, mock.Anything).Call.Return(nil).Once()
@ -1679,7 +1680,7 @@ func TestRBAC_Role(t *testing.T) {
kvmock.EXPECT().Load(errorRoleSavepath).Return("", nil)
// Catalog.save() returns nil
kvmock.EXPECT().Load(noErrorRoleSavepath).Return("", common.NewKeyNotExistError(noErrorRoleSavepath))
kvmock.EXPECT().Load(noErrorRoleSavepath).Return("", merr.WrapErrIoKeyNotFound(noErrorRoleSavepath))
// Catalog.remove() returns error
kvmock.EXPECT().Load(errorRoleRemovepath).Return("", errors.New("not exists"))
@ -2096,19 +2097,19 @@ func TestRBAC_Grant(t *testing.T) {
})
kvmock.EXPECT().Load(keyNotExistRoleKey).Call.
Return("", func(key string) error {
return common.NewKeyNotExistError(key)
return merr.WrapErrIoKeyNotFound(key)
})
kvmock.EXPECT().Load(keyNotExistRoleKeyWithDb).Call.
Return("", func(key string) error {
return common.NewKeyNotExistError(key)
return merr.WrapErrIoKeyNotFound(key)
})
kvmock.EXPECT().Load(errorSaveRoleKey).Call.
Return("", func(key string) error {
return common.NewKeyNotExistError(key)
return merr.WrapErrIoKeyNotFound(key)
})
kvmock.EXPECT().Load(errorSaveRoleKeyWithDb).Call.
Return("", func(key string) error {
return common.NewKeyNotExistError(key)
return merr.WrapErrIoKeyNotFound(key)
})
kvmock.EXPECT().Save(keyNotExistRoleKeyWithDb, mock.Anything).Return(nil)
kvmock.EXPECT().Save(errorSaveRoleKeyWithDb, mock.Anything).Return(errors.New("mock save error role"))
@ -2126,11 +2127,11 @@ func TestRBAC_Grant(t *testing.T) {
})
kvmock.EXPECT().Load(keyNotExistPrivilegeKey).Call.
Return("", func(key string) error {
return common.NewKeyNotExistError(key)
return merr.WrapErrIoKeyNotFound(key)
})
kvmock.EXPECT().Load(keyNotExistPrivilegeKey2WithDb).Call.
Return("", func(key string) error {
return common.NewKeyNotExistError(key)
return merr.WrapErrIoKeyNotFound(key)
})
kvmock.EXPECT().Load(mock.Anything).Call.Return("", nil)

6
internal/rootcoord/root_coord.go
View File

@ -2420,7 +2420,7 @@ func (c *Core) SelectRole(ctx context.Context, in *milvuspb.SelectRoleRequest) (
if in.Role != nil {
if _, err := c.meta.SelectRole(util.DefaultTenant, &milvuspb.RoleEntity{Name: in.Role.Name}, false); err != nil {
if common.IsKeyNotExistError(err) {
if errors.Is(err, merr.ErrIoKeyNotFound) {
return &milvuspb.SelectRoleResponse{
Status: merr.Success(),
}, nil
@ -2467,7 +2467,7 @@ func (c *Core) SelectUser(ctx context.Context, in *milvuspb.SelectUserRequest) (
if in.User != nil {
if _, err := c.meta.SelectUser(util.DefaultTenant, &milvuspb.UserEntity{Name: in.User.Name}, false); err != nil {
if common.IsKeyNotExistError(err) {
if errors.Is(err, merr.ErrIoKeyNotFound) {
return &milvuspb.SelectUserResponse{
Status: merr.Success(),
}, nil
@ -2688,7 +2688,7 @@ func (c *Core) SelectGrant(ctx context.Context, in *milvuspb.SelectGrantRequest)
}
grantEntities, err := c.meta.SelectGrant(util.DefaultTenant, in.Entity)
if common.IsKeyNotExistError(err) {
if errors.Is(err, merr.ErrIoKeyNotFound) {
return &milvuspb.SelectGrantResponse{
Status: merr.Success(),
}, nil

19
pkg/common/error.go
View File

@ -53,22 +53,3 @@ func IsIgnorableError(err error) bool {
_, ok := err.(*IgnorableError)
return ok
}
var _ error = &KeyNotExistError{}
func NewKeyNotExistError(key string) error {
return &KeyNotExistError{key: key}
}
func IsKeyNotExistError(err error) bool {
_, ok := err.(*KeyNotExistError)
return ok
}
type KeyNotExistError struct {
key string
}
func (k *KeyNotExistError) Error() string {
return fmt.Sprintf("there is no value on key = %s", k.key)
}

6
pkg/common/error_test.go
View File

@ -29,9 +29,3 @@ func TestIgnorableError(t *testing.T) {
assert.True(t, IsIgnorableError(iErr))
assert.False(t, IsIgnorableError(err))
}
func TestNotExistError(t *testing.T) {
err := errors.New("err")
assert.Equal(t, false, IsKeyNotExistError(err))
assert.Equal(t, true, IsKeyNotExistError(NewKeyNotExistError("foo")))
}