diff --git a/internal/kv/etcd/etcd_kv.go b/internal/kv/etcd/etcd_kv.go index e0efd29be6..b2008b8be9 100644 --- a/internal/kv/etcd/etcd_kv.go +++ b/internal/kv/etcd/etcd_kv.go @@ -27,6 +27,7 @@ const ( RequestTimeout = 10 * time.Second ) +// EtcdKV implments TxnKv interface, it support to process multiple kvs in a transactions type EtcdKV struct { client *clientv3.Client rootPath string @@ -59,8 +60,8 @@ func (kv *EtcdKV) GetPath(key string) string { } func (kv *EtcdKV) LoadWithPrefix(key string) ([]string, []string, error) { + start := time.Now() key = path.Join(kv.rootPath, key) - log.Debug("LoadWithPrefix ", zap.String("prefix", key)) ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() resp, err := kv.client.Get(ctx, key, clientv3.WithPrefix(), @@ -74,12 +75,13 @@ func (kv *EtcdKV) LoadWithPrefix(key string) ([]string, []string, error) { keys = append(keys, string(kv.Key)) values = append(values, string(kv.Value)) } + CheckElapseAndWarn(start, "Slow etcd operation load with prefix") return keys, values, nil } func (kv *EtcdKV) LoadWithPrefix2(key string) ([]string, []string, []int64, error) { + start := time.Now() key = path.Join(kv.rootPath, key) - log.Debug("LoadWithPrefix ", zap.String("prefix", key)) ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() resp, err := kv.client.Get(ctx, key, clientv3.WithPrefix(), @@ -95,10 +97,12 @@ func (kv *EtcdKV) LoadWithPrefix2(key string) ([]string, []string, []int64, erro values = append(values, string(kv.Value)) versions = append(versions, kv.Version) } + CheckElapseAndWarn(start, "Slow etcd operation load with prefix2") return keys, values, versions, nil } func (kv *EtcdKV) Load(key string) (string, error) { + start := time.Now() key = path.Join(kv.rootPath, key) ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() @@ -109,11 +113,12 @@ func (kv *EtcdKV) Load(key string) (string, error) { if resp.Count <= 0 { return "", fmt.Errorf("there is no value on key = %s", key) } - + CheckElapseAndWarn(start, "Slow etcd operation load") return string(resp.Kvs[0].Value), nil } func (kv *EtcdKV) MultiLoad(keys []string) ([]string, error) { + start := time.Now() ops := make([]clientv3.Op, 0, len(keys)) for _, keyLoad := range keys { ops = append(ops, clientv3.OpGet(path.Join(kv.rootPath, keyLoad))) @@ -134,23 +139,21 @@ func (kv *EtcdKV) MultiLoad(keys []string) ([]string, error) { result = append(result, "") } for _, ev := range rp.GetResponseRange().Kvs { - log.Debug("MultiLoad", zap.ByteString("key", ev.Key), - zap.ByteString("value", ev.Value)) result = append(result, string(ev.Value)) } } if len(invalid) != 0 { - log.Debug("MultiLoad: there are invalid keys", - zap.Strings("keys", invalid)) + log.Warn("MultiLoad: there are invalid keys", zap.Strings("keys", invalid)) err = fmt.Errorf("there are invalid keys: %s", invalid) return result, err } + CheckElapseAndWarn(start, "Slow etcd operation multi load") return result, nil } func (kv *EtcdKV) LoadWithRevision(key string) ([]string, []string, int64, error) { + start := time.Now() key = path.Join(kv.rootPath, key) - log.Debug("LoadWithPrefix ", zap.String("prefix", key)) ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() resp, err := kv.client.Get(ctx, key, clientv3.WithPrefix(), @@ -164,27 +167,33 @@ func (kv *EtcdKV) LoadWithRevision(key string) ([]string, []string, int64, error keys = append(keys, string(kv.Key)) values = append(values, string(kv.Value)) } + CheckElapseAndWarn(start, "Slow etcd operation load with revision") return keys, values, resp.Header.Revision, nil } func (kv *EtcdKV) Save(key, value string) error { + start := time.Now() key = path.Join(kv.rootPath, key) ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() _, err := kv.client.Put(ctx, key, value) + CheckElapseAndWarn(start, "Slow etcd operation save") return err } // SaveWithLease is a function to put value in etcd with etcd lease options. func (kv *EtcdKV) SaveWithLease(key, value string, id clientv3.LeaseID) error { + start := time.Now() key = path.Join(kv.rootPath, key) ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() _, err := kv.client.Put(ctx, key, value, clientv3.WithLease(id)) + CheckElapseAndWarn(start, "Slow etcd operation save with lease") return err } func (kv *EtcdKV) MultiSave(kvs map[string]string) error { + start := time.Now() ops := make([]clientv3.Op, 0, len(kvs)) for key, value := range kvs { ops = append(ops, clientv3.OpPut(path.Join(kv.rootPath, key), value)) @@ -194,28 +203,34 @@ func (kv *EtcdKV) MultiSave(kvs map[string]string) error { defer cancel() _, err := kv.client.Txn(ctx).If().Then(ops...).Commit() + CheckElapseAndWarn(start, "Slow etcd operation multi save") return err } func (kv *EtcdKV) RemoveWithPrefix(prefix string) error { + start := time.Now() key := path.Join(kv.rootPath, prefix) ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() _, err := kv.client.Delete(ctx, key, clientv3.WithPrefix()) + CheckElapseAndWarn(start, "Slow etcd operation remove with prefix") return err } func (kv *EtcdKV) Remove(key string) error { + start := time.Now() key = path.Join(kv.rootPath, key) ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() _, err := kv.client.Delete(ctx, key) + CheckElapseAndWarn(start, "Slow etcd operation remove") return err } func (kv *EtcdKV) MultiRemove(keys []string) error { + start := time.Now() ops := make([]clientv3.Op, 0, len(keys)) for _, key := range keys { ops = append(ops, clientv3.OpDelete(path.Join(kv.rootPath, key))) @@ -225,10 +240,12 @@ func (kv *EtcdKV) MultiRemove(keys []string) error { defer cancel() _, err := kv.client.Txn(ctx).If().Then(ops...).Commit() + CheckElapseAndWarn(start, "Slow etcd operation multi remove") return err } func (kv *EtcdKV) MultiSaveAndRemove(saves map[string]string, removals []string) error { + start := time.Now() ops := make([]clientv3.Op, 0, len(saves)+len(removals)) for key, value := range saves { ops = append(ops, clientv3.OpPut(path.Join(kv.rootPath, key), value)) @@ -238,47 +255,55 @@ func (kv *EtcdKV) MultiSaveAndRemove(saves map[string]string, removals []string) ops = append(ops, clientv3.OpDelete(path.Join(kv.rootPath, keyDelete))) } - log.Debug("MultiSaveAndRemove") ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() _, err := kv.client.Txn(ctx).If().Then(ops...).Commit() + CheckElapseAndWarn(start, "Slow etcd operation multi save and remove") return err } func (kv *EtcdKV) Watch(key string) clientv3.WatchChan { + start := time.Now() key = path.Join(kv.rootPath, key) rch := kv.client.Watch(context.Background(), key, clientv3.WithCreatedNotify()) + CheckElapseAndWarn(start, "Slow etcd operation watch") return rch } func (kv *EtcdKV) WatchWithPrefix(key string) clientv3.WatchChan { + start := time.Now() key = path.Join(kv.rootPath, key) rch := kv.client.Watch(context.Background(), key, clientv3.WithPrefix(), clientv3.WithCreatedNotify()) + CheckElapseAndWarn(start, "Slow etcd operation watch with prefix") return rch } func (kv *EtcdKV) WatchWithRevision(key string, revision int64) clientv3.WatchChan { + start := time.Now() key = path.Join(kv.rootPath, key) rch := kv.client.Watch(context.Background(), key, clientv3.WithPrefix(), clientv3.WithPrevKV(), clientv3.WithRev(revision)) + CheckElapseAndWarn(start, "Slow etcd operation watch with revision") return rch } func (kv *EtcdKV) MultiRemoveWithPrefix(keys []string) error { + start := time.Now() ops := make([]clientv3.Op, 0, len(keys)) for _, k := range keys { op := clientv3.OpDelete(path.Join(kv.rootPath, k), clientv3.WithPrefix()) ops = append(ops, op) } - log.Debug("MultiRemoveWithPrefix") ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() _, err := kv.client.Txn(ctx).If().Then(ops...).Commit() + CheckElapseAndWarn(start, "Slow etcd operation multi remove with prefix") return err } func (kv *EtcdKV) MultiSaveAndRemoveWithPrefix(saves map[string]string, removals []string) error { + start := time.Now() ops := make([]clientv3.Op, 0, len(saves)) for key, value := range saves { ops = append(ops, clientv3.OpPut(path.Join(kv.rootPath, key), value)) @@ -288,33 +313,38 @@ func (kv *EtcdKV) MultiSaveAndRemoveWithPrefix(saves map[string]string, removals ops = append(ops, clientv3.OpDelete(path.Join(kv.rootPath, keyDelete), clientv3.WithPrefix())) } - log.Debug("MultiSaveAndRemove") ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() _, err := kv.client.Txn(ctx).If().Then(ops...).Commit() + CheckElapseAndWarn(start, "Slow etcd operation multi save and move with prefix") return err } // Grant creates a new lease implemented in etcd grant interface. func (kv *EtcdKV) Grant(ttl int64) (id clientv3.LeaseID, err error) { + start := time.Now() resp, err := kv.client.Grant(context.Background(), ttl) + CheckElapseAndWarn(start, "Slow etcd operation grant") return resp.ID, err } // KeepAlive keeps the lease alive forever with leaseID. // Implemented in etcd interface. func (kv *EtcdKV) KeepAlive(id clientv3.LeaseID) (<-chan *clientv3.LeaseKeepAliveResponse, error) { + start := time.Now() ch, err := kv.client.KeepAlive(context.Background(), id) if err != nil { return nil, err } + CheckElapseAndWarn(start, "Slow etcd operation keepAlive") return ch, nil } // CompareValueAndSwap compares the existing value with compare, and if they are // equal, the target is stored in etcd. func (kv *EtcdKV) CompareValueAndSwap(key, value, target string, opts ...clientv3.OpOption) error { + start := time.Now() ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() resp, err := kv.client.Txn(ctx).If( @@ -329,13 +359,14 @@ func (kv *EtcdKV) CompareValueAndSwap(key, value, target string, opts ...clientv if !resp.Succeeded { return fmt.Errorf("function CompareAndSwap error for compare is false for key: %s", key) } - + CheckElapseAndWarn(start, "Slow etcd operation compare value and swap") return nil } // CompareVersionAndSwap compares the existing key-value's version with version, and if // they are equal, the target is stored in etcd. func (kv *EtcdKV) CompareVersionAndSwap(key string, version int64, target string, opts ...clientv3.OpOption) error { + start := time.Now() ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout) defer cancel() resp, err := kv.client.Txn(ctx).If( @@ -350,6 +381,15 @@ func (kv *EtcdKV) CompareVersionAndSwap(key string, version int64, target string if !resp.Succeeded { return fmt.Errorf("function CompareAndSwap error for compare is false for key: %s", key) } - + CheckElapseAndWarn(start, "Slow etcd operation compare version and swap") return nil } + +func CheckElapseAndWarn(start time.Time, message string) bool { + elapsed := time.Since(start) + if elapsed.Milliseconds() > 2000 { + log.Warn(message, zap.String("time spent", elapsed.String())) + return true + } + return false +} diff --git a/internal/kv/etcd/etcd_kv_test.go b/internal/kv/etcd/etcd_kv_test.go index ba2dfb10d3..b881f0f7d1 100644 --- a/internal/kv/etcd/etcd_kv_test.go +++ b/internal/kv/etcd/etcd_kv_test.go @@ -15,6 +15,7 @@ import ( "os" "strings" "testing" + "time" etcdkv "github.com/milvus-io/milvus/internal/kv/etcd" "github.com/milvus-io/milvus/internal/util/paramtable" @@ -488,3 +489,13 @@ func TestEtcdKV_Load(te *testing.T) { }) } + +func TestElapse(t *testing.T) { + start := time.Now() + isElapse := etcdkv.CheckElapseAndWarn(start, "err message") + assert.Equal(t, isElapse, false) + + time.Sleep(time.Duration(3) * time.Second) + isElapse = etcdkv.CheckElapseAndWarn(start, "err message") + assert.Equal(t, isElapse, true) +}