remoting/etcdv3/client.go - dubbo-go - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package etcdv3

 import (
 	"context"
 	"sync"
 	"time"
 )

 import (
 	"github.com/coreos/etcd/clientv3"
 	"github.com/coreos/etcd/clientv3/concurrency"
 	perrors "github.com/pkg/errors"
 	"google.golang.org/grpc"
 )

 import (
 	"github.com/apache/dubbo-go/common/logger"
 )

 const (
 	// ConnDelay connection delay
 	ConnDelay = 3
 	// MaxFailTimes max failure times
 	MaxFailTimes = 15
 	// RegistryETCDV3Client client name
 	RegistryETCDV3Client = "etcd registry"
 	// metadataETCDV3Client client name
 	MetadataETCDV3Client = "etcd metadata"
 )

 var (
 	// Defines related errors
 	ErrNilETCDV3Client = perrors.New("etcd raw client is nil") // full describe the ERR
 	ErrKVPairNotFound  = perrors.New("k/v pair not found")
 )

 // nolint
 type Options struct {
 	name      string
 	endpoints []string
 	client    *Client
 	timeout   time.Duration
 	heartbeat int // heartbeat second
 }

 // Option will define a function of handling Options
 type Option func(*Options)

 // WithEndpoints sets etcd client endpoints
 func WithEndpoints(endpoints ...string) Option {
 	return func(opt *Options) {
 		opt.endpoints = endpoints
 	}
 }

 // WithName sets etcd client name
 func WithName(name string) Option {
 	return func(opt *Options) {
 		opt.name = name
 	}
 }

 // WithTimeout sets etcd client timeout
 func WithTimeout(timeout time.Duration) Option {
 	return func(opt *Options) {
 		opt.timeout = timeout
 	}
 }

 // WithHeartbeat sets etcd client heartbeat
 func WithHeartbeat(heartbeat int) Option {
 	return func(opt *Options) {
 		opt.heartbeat = heartbeat
 	}
 }

 // ValidateClient validates client and sets options
 func ValidateClient(container clientFacade, opts ...Option) error {
 	options := &Options{
 		heartbeat: 1, // default heartbeat
 	}
 	for _, opt := range opts {
 		opt(options)
 	}

 	lock := container.ClientLock()
 	lock.Lock()
 	defer lock.Unlock()

 	// new Client
 	if container.Client() == nil {
 		newClient, err := NewClient(options.name, options.endpoints, options.timeout, options.heartbeat)
 		if err != nil {
 			logger.Warnf("new etcd client (name{%s}, etcd addresses{%v}, timeout{%d}) = error{%v}",
 				options.name, options.endpoints, options.timeout, err)
 			return perrors.WithMessagef(err, "new client (address:%+v)", options.endpoints)
 		}
 		container.SetClient(newClient)
 	}

 	// Client lose connection with etcd server
 	if container.Client().rawClient == nil {
 		newClient, err := NewClient(options.name, options.endpoints, options.timeout, options.heartbeat)
 		if err != nil {
 			logger.Warnf("new etcd client (name{%s}, etcd addresses{%v}, timeout{%d}) = error{%v}",
 				options.name, options.endpoints, options.timeout, err)
 			return perrors.WithMessagef(err, "new client (address:%+v)", options.endpoints)
 		}
 		container.SetClient(newClient)
 	}

 	return nil
 }

 //  nolint
 func NewServiceDiscoveryClient(opts ...Option) *Client {
 	options := &Options{
 		heartbeat: 1, // default heartbeat
 	}
 	for _, opt := range opts {
 		opt(options)
 	}

 	newClient, err := NewClient(options.name, options.endpoints, options.timeout, options.heartbeat)
 	if err != nil {
 		logger.Errorf("new etcd client (name{%s}, etcd addresses{%v}, timeout{%d}) = error{%v}",
 			options.name, options.endpoints, options.timeout, err)
 	}
 	return newClient
 }

 // Client represents etcd client Configuration
 type Client struct {
 	lock sync.RWMutex

 	// these properties are only set once when they are started.
 	name      string
 	endpoints []string
 	timeout   time.Duration
 	heartbeat int

 	ctx       context.Context    // if etcd server connection lose, the ctx.Done will be sent msg
 	cancel    context.CancelFunc // cancel the ctx, all watcher will stopped
 	rawClient *clientv3.Client

 	exit chan struct{}
 	Wait sync.WaitGroup
 }

 // nolint
 func NewClient(name string, endpoints []string, timeout time.Duration, heartbeat int) (*Client, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	rawClient, err := clientv3.New(clientv3.Config{
 		Context:     ctx,
 		Endpoints:   endpoints,
 		DialTimeout: timeout,
 		DialOptions: []grpc.DialOption{grpc.WithBlock()},
 	})
 	if err != nil {
 		return nil, perrors.WithMessage(err, "new raw client block connect to server")
 	}

 	c := &Client{
 		name:      name,
 		timeout:   timeout,
 		endpoints: endpoints,
 		heartbeat: heartbeat,

 		ctx:       ctx,
 		cancel:    cancel,
 		rawClient: rawClient,

 		exit: make(chan struct{}),
 	}

 	if err := c.maintenanceStatus(); err != nil {
 		return nil, perrors.WithMessage(err, "client maintenance status")
 	}
 	return c, nil
 }

 // NOTICE: need to get the lock before calling this method
 func (c *Client) clean() {
 	// close raw client
 	c.rawClient.Close()

 	// cancel ctx for raw client
 	c.cancel()

 	// clean raw client
 	c.rawClient = nil
 }

 func (c *Client) stop() bool {
 	select {
 	case <-c.exit:
 		return true
 	default:
 		close(c.exit)
 	}
 	return false
 }

 // nolint
 func (c *Client) Close() {
 	if c == nil {
 		return
 	}

 	// stop the client
 	c.stop()

 	// wait client maintenance status stop
 	c.Wait.Wait()

 	c.lock.Lock()
 	defer c.lock.Unlock()
 	if c.rawClient != nil {
 		c.clean()
 	}
 	logger.Warnf("etcd client{name:%s, endpoints:%s} exit now.", c.name, c.endpoints)
 }

 func (c *Client) maintenanceStatus() error {
 	s, err := concurrency.NewSession(c.rawClient, concurrency.WithTTL(c.heartbeat))
 	if err != nil {
 		return perrors.WithMessage(err, "new session with server")
 	}

 	// must add wg before go maintenance status goroutine
 	c.Wait.Add(1)
 	go c.maintenanceStatusLoop(s)
 	return nil
 }

 func (c *Client) maintenanceStatusLoop(s *concurrency.Session) {
 	defer func() {
 		c.Wait.Done()
 		logger.Infof("etcd client {endpoints:%v, name:%s} maintenance goroutine game over.", c.endpoints, c.name)
 	}()

 	for {
 		select {
 		case <-c.Done():
 			// Client be stopped, will clean the client hold resources
 			return
 		case <-s.Done():
 			logger.Warn("etcd server stopped")
 			c.lock.Lock()
 			// when etcd server stopped, cancel ctx, stop all watchers
 			c.clean()
 			// when connection lose, stop client, trigger reconnect to etcd
 			c.stop()
 			c.lock.Unlock()
 			return
 		}
 	}
 }

 // if k not exist will put k/v in etcd, otherwise return nil
 func (c *Client) put(k string, v string, opts ...clientv3.OpOption) error {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return ErrNilETCDV3Client
 	}

 	_, err := c.rawClient.Txn(c.ctx).
 		If(clientv3.Compare(clientv3.Version(k), "<", 1)).
 		Then(clientv3.OpPut(k, v, opts...)).
 		Commit()
 	return err
 }

 // if k not exist will put k/v in etcd
 // if k is already exist in etcd, replace it
 func (c *Client) update(k string, v string, opts ...clientv3.OpOption) error {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return ErrNilETCDV3Client
 	}

 	_, err := c.rawClient.Txn(c.ctx).
 		If(clientv3.Compare(clientv3.Version(k), "!=", -1)).
 		Then(clientv3.OpPut(k, v, opts...)).
 		Commit()
 	return err
 }

 func (c *Client) delete(k string) error {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return ErrNilETCDV3Client
 	}

 	_, err := c.rawClient.Delete(c.ctx, k)
 	return err
 }

 func (c *Client) get(k string) (string, error) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return "", ErrNilETCDV3Client
 	}

 	resp, err := c.rawClient.Get(c.ctx, k)
 	if err != nil {
 		return "", err
 	}

 	if len(resp.Kvs) == 0 {
 		return "", ErrKVPairNotFound
 	}

 	return string(resp.Kvs[0].Value), nil
 }

 // nolint
 func (c *Client) CleanKV() error {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return ErrNilETCDV3Client
 	}

 	_, err := c.rawClient.Delete(c.ctx, "", clientv3.WithPrefix())
 	return err
 }

 func (c *Client) getChildren(k string) ([]string, []string, error) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return nil, nil, ErrNilETCDV3Client
 	}

 	resp, err := c.rawClient.Get(c.ctx, k, clientv3.WithPrefix())
 	if err != nil {
 		return nil, nil, err
 	}

 	if len(resp.Kvs) == 0 {
 		return nil, nil, ErrKVPairNotFound
 	}

 	kList := make([]string, 0, len(resp.Kvs))
 	vList := make([]string, 0, len(resp.Kvs))
 	for _, kv := range resp.Kvs {
 		kList = append(kList, string(kv.Key))
 		vList = append(vList, string(kv.Value))
 	}
 	return kList, vList, nil
 }

 func (c *Client) watchWithPrefix(prefix string) (clientv3.WatchChan, error) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return nil, ErrNilETCDV3Client
 	}

 	return c.rawClient.Watch(c.ctx, prefix, clientv3.WithPrefix()), nil
 }

 func (c *Client) watch(k string) (clientv3.WatchChan, error) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return nil, ErrNilETCDV3Client
 	}

 	return c.rawClient.Watch(c.ctx, k), nil
 }

 func (c *Client) keepAliveKV(k string, v string) error {
 	c.lock.RLock()
 	defer c.lock.RUnlock()

 	if c.rawClient == nil {
 		return ErrNilETCDV3Client
 	}

 	// make lease time longer, since 1 second is too short
 	lease, err := c.rawClient.Grant(c.ctx, int64(30*time.Second.Seconds()))
 	if err != nil {
 		return perrors.WithMessage(err, "grant lease")
 	}

 	keepAlive, err := c.rawClient.KeepAlive(c.ctx, lease.ID)
 	if err != nil || keepAlive == nil {
 		c.rawClient.Revoke(c.ctx, lease.ID)
 		if err != nil {
 			return perrors.WithMessage(err, "keep alive lease")
 		} else {
 			return perrors.New("keep alive lease")
 		}
 	}

 	_, err = c.rawClient.Put(c.ctx, k, v, clientv3.WithLease(lease.ID))
 	return perrors.WithMessage(err, "put k/v with lease")
 }

 // nolint
 func (c *Client) Done() <-chan struct{} {
 	return c.exit
 }

 // nolint
 func (c *Client) Valid() bool {
 	select {
 	case <-c.exit:
 		return false
 	default:
 	}

 	c.lock.RLock()
 	defer c.lock.RUnlock()
 	return c.rawClient != nil
 }

 // nolint
 func (c *Client) Create(k string, v string) error {
 	err := c.put(k, v)
 	return perrors.WithMessagef(err, "put k/v (key: %s value %s)", k, v)
 }

 // Update key value ...
 func (c *Client) Update(k, v string) error {
 	err := c.update(k, v)
 	return perrors.WithMessagef(err, "Update k/v (key: %s value %s)", k, v)
 }

 // nolint
 func (c *Client) Delete(k string) error {
 	err := c.delete(k)
 	return perrors.WithMessagef(err, "delete k/v (key %s)", k)
 }

 // RegisterTemp registers a temporary node
 func (c *Client) RegisterTemp(k, v string) error {
 	err := c.keepAliveKV(k, v)
 	return perrors.WithMessagef(err, "keepalive kv (key %s)", k)
 }

 // GetChildrenKVList gets children kv list by @k
 func (c *Client) GetChildrenKVList(k string) ([]string, []string, error) {
 	kList, vList, err := c.getChildren(k)
 	return kList, vList, perrors.WithMessagef(err, "get key children (key %s)", k)
 }

 // Get gets value by @k
 func (c *Client) Get(k string) (string, error) {
 	v, err := c.get(k)
 	return v, perrors.WithMessagef(err, "get key value (key %s)", k)
 }

 // Watch watches on spec key
 func (c *Client) Watch(k string) (clientv3.WatchChan, error) {
 	wc, err := c.watch(k)
 	return wc, perrors.WithMessagef(err, "watch prefix (key %s)", k)
 }

 // WatchWithPrefix watches on spec prefix
 func (c *Client) WatchWithPrefix(prefix string) (clientv3.WatchChan, error) {
 	wc, err := c.watchWithPrefix(prefix)
 	return wc, perrors.WithMessagef(err, "watch prefix (key %s)", prefix)
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package etcdv3

	import (
	"context"
	"sync"
	"time"
	)

	import (
	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/clientv3/concurrency"
	perrors "github.com/pkg/errors"
	"google.golang.org/grpc"
	)

	import (
	"github.com/apache/dubbo-go/common/logger"
	)

	const (
	// ConnDelay connection delay
	ConnDelay = 3
	// MaxFailTimes max failure times
	MaxFailTimes = 15
	// RegistryETCDV3Client client name
	RegistryETCDV3Client = "etcd registry"
	// metadataETCDV3Client client name
	MetadataETCDV3Client = "etcd metadata"
	)

	var (
	// Defines related errors
	ErrNilETCDV3Client = perrors.New("etcd raw client is nil") // full describe the ERR
	ErrKVPairNotFound = perrors.New("k/v pair not found")
	)

	// nolint
	type Options struct {
	name string
	endpoints []string
	client *Client
	timeout time.Duration
	heartbeat int // heartbeat second
	}

	// Option will define a function of handling Options
	type Option func(*Options)

	// WithEndpoints sets etcd client endpoints
	func WithEndpoints(endpoints ...string) Option {
	return func(opt *Options) {
	opt.endpoints = endpoints
	}
	}

	// WithName sets etcd client name
	func WithName(name string) Option {
	return func(opt *Options) {
	opt.name = name
	}
	}

	// WithTimeout sets etcd client timeout
	func WithTimeout(timeout time.Duration) Option {
	return func(opt *Options) {
	opt.timeout = timeout
	}
	}

	// WithHeartbeat sets etcd client heartbeat
	func WithHeartbeat(heartbeat int) Option {
	return func(opt *Options) {
	opt.heartbeat = heartbeat
	}
	}

	// ValidateClient validates client and sets options
	func ValidateClient(container clientFacade, opts ...Option) error {
	options := &Options{
	heartbeat: 1, // default heartbeat
	}
	for _, opt := range opts {
	opt(options)
	}

	lock := container.ClientLock()
	lock.Lock()
	defer lock.Unlock()

	// new Client
	if container.Client() == nil {
	newClient, err := NewClient(options.name, options.endpoints, options.timeout, options.heartbeat)
	if err != nil {
	logger.Warnf("new etcd client (name{%s}, etcd addresses{%v}, timeout{%d}) = error{%v}",
	options.name, options.endpoints, options.timeout, err)
	return perrors.WithMessagef(err, "new client (address:%+v)", options.endpoints)
	}
	container.SetClient(newClient)
	}

	// Client lose connection with etcd server
	if container.Client().rawClient == nil {
	newClient, err := NewClient(options.name, options.endpoints, options.timeout, options.heartbeat)
	if err != nil {
	logger.Warnf("new etcd client (name{%s}, etcd addresses{%v}, timeout{%d}) = error{%v}",
	options.name, options.endpoints, options.timeout, err)
	return perrors.WithMessagef(err, "new client (address:%+v)", options.endpoints)
	}
	container.SetClient(newClient)
	}

	return nil
	}

	// nolint
	func NewServiceDiscoveryClient(opts ...Option) *Client {
	options := &Options{
	heartbeat: 1, // default heartbeat
	}
	for _, opt := range opts {
	opt(options)
	}

	newClient, err := NewClient(options.name, options.endpoints, options.timeout, options.heartbeat)
	if err != nil {
	logger.Errorf("new etcd client (name{%s}, etcd addresses{%v}, timeout{%d}) = error{%v}",
	options.name, options.endpoints, options.timeout, err)
	}
	return newClient
	}

	// Client represents etcd client Configuration
	type Client struct {
	lock sync.RWMutex

	// these properties are only set once when they are started.
	name string
	endpoints []string
	timeout time.Duration
	heartbeat int

	ctx context.Context // if etcd server connection lose, the ctx.Done will be sent msg
	cancel context.CancelFunc // cancel the ctx, all watcher will stopped
	rawClient *clientv3.Client

	exit chan struct{}
	Wait sync.WaitGroup
	}

	// nolint
	func NewClient(name string, endpoints []string, timeout time.Duration, heartbeat int) (*Client, error) {
	ctx, cancel := context.WithCancel(context.Background())
	rawClient, err := clientv3.New(clientv3.Config{
	Context: ctx,
	Endpoints: endpoints,
	DialTimeout: timeout,
	DialOptions: []grpc.DialOption{grpc.WithBlock()},
	})
	if err != nil {
	return nil, perrors.WithMessage(err, "new raw client block connect to server")
	}

	c := &Client{
	name: name,
	timeout: timeout,
	endpoints: endpoints,
	heartbeat: heartbeat,

	ctx: ctx,
	cancel: cancel,
	rawClient: rawClient,

	exit: make(chan struct{}),
	}

	if err := c.maintenanceStatus(); err != nil {
	return nil, perrors.WithMessage(err, "client maintenance status")
	}
	return c, nil
	}

	// NOTICE: need to get the lock before calling this method
	func (c *Client) clean() {
	// close raw client
	c.rawClient.Close()

	// cancel ctx for raw client
	c.cancel()

	// clean raw client
	c.rawClient = nil
	}

	func (c *Client) stop() bool {
	select {
	case <-c.exit:
	return true
	default:
	close(c.exit)
	}
	return false
	}

	// nolint
	func (c *Client) Close() {
	if c == nil {
	return
	}

	// stop the client
	c.stop()

	// wait client maintenance status stop
	c.Wait.Wait()

	c.lock.Lock()
	defer c.lock.Unlock()
	if c.rawClient != nil {
	c.clean()
	}
	logger.Warnf("etcd client{name:%s, endpoints:%s} exit now.", c.name, c.endpoints)
	}

	func (c *Client) maintenanceStatus() error {
	s, err := concurrency.NewSession(c.rawClient, concurrency.WithTTL(c.heartbeat))
	if err != nil {
	return perrors.WithMessage(err, "new session with server")
	}

	// must add wg before go maintenance status goroutine
	c.Wait.Add(1)
	go c.maintenanceStatusLoop(s)
	return nil
	}

	func (c Client) maintenanceStatusLoop(s concurrency.Session) {
	defer func() {
	c.Wait.Done()
	logger.Infof("etcd client {endpoints:%v, name:%s} maintenance goroutine game over.", c.endpoints, c.name)
	}()

	for {
	select {
	case <-c.Done():
	// Client be stopped, will clean the client hold resources
	return
	case <-s.Done():
	logger.Warn("etcd server stopped")
	c.lock.Lock()
	// when etcd server stopped, cancel ctx, stop all watchers
	c.clean()
	// when connection lose, stop client, trigger reconnect to etcd
	c.stop()
	c.lock.Unlock()
	return
	}
	}
	}

	// if k not exist will put k/v in etcd, otherwise return nil
	func (c *Client) put(k string, v string, opts ...clientv3.OpOption) error {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return ErrNilETCDV3Client
	}

	_, err := c.rawClient.Txn(c.ctx).
	If(clientv3.Compare(clientv3.Version(k), "<", 1)).
	Then(clientv3.OpPut(k, v, opts...)).
	Commit()
	return err
	}

	// if k not exist will put k/v in etcd
	// if k is already exist in etcd, replace it
	func (c *Client) update(k string, v string, opts ...clientv3.OpOption) error {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return ErrNilETCDV3Client
	}

	_, err := c.rawClient.Txn(c.ctx).
	If(clientv3.Compare(clientv3.Version(k), "!=", -1)).
	Then(clientv3.OpPut(k, v, opts...)).
	Commit()
	return err
	}

	func (c *Client) delete(k string) error {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return ErrNilETCDV3Client
	}

	_, err := c.rawClient.Delete(c.ctx, k)
	return err
	}

	func (c *Client) get(k string) (string, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return "", ErrNilETCDV3Client
	}

	resp, err := c.rawClient.Get(c.ctx, k)
	if err != nil {
	return "", err
	}

	if len(resp.Kvs) == 0 {
	return "", ErrKVPairNotFound
	}

	return string(resp.Kvs[0].Value), nil
	}

	// nolint
	func (c *Client) CleanKV() error {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return ErrNilETCDV3Client
	}

	_, err := c.rawClient.Delete(c.ctx, "", clientv3.WithPrefix())
	return err
	}

	func (c *Client) getChildren(k string) ([]string, []string, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return nil, nil, ErrNilETCDV3Client
	}

	resp, err := c.rawClient.Get(c.ctx, k, clientv3.WithPrefix())
	if err != nil {
	return nil, nil, err
	}

	if len(resp.Kvs) == 0 {
	return nil, nil, ErrKVPairNotFound
	}

	kList := make([]string, 0, len(resp.Kvs))
	vList := make([]string, 0, len(resp.Kvs))
	for _, kv := range resp.Kvs {
	kList = append(kList, string(kv.Key))
	vList = append(vList, string(kv.Value))
	}
	return kList, vList, nil
	}

	func (c *Client) watchWithPrefix(prefix string) (clientv3.WatchChan, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return nil, ErrNilETCDV3Client
	}

	return c.rawClient.Watch(c.ctx, prefix, clientv3.WithPrefix()), nil
	}

	func (c *Client) watch(k string) (clientv3.WatchChan, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return nil, ErrNilETCDV3Client
	}

	return c.rawClient.Watch(c.ctx, k), nil
	}

	func (c *Client) keepAliveKV(k string, v string) error {
	c.lock.RLock()
	defer c.lock.RUnlock()

	if c.rawClient == nil {
	return ErrNilETCDV3Client
	}

	// make lease time longer, since 1 second is too short
	lease, err := c.rawClient.Grant(c.ctx, int64(30*time.Second.Seconds()))
	if err != nil {
	return perrors.WithMessage(err, "grant lease")
	}

	keepAlive, err := c.rawClient.KeepAlive(c.ctx, lease.ID)
	if err != nil \|\| keepAlive == nil {
	c.rawClient.Revoke(c.ctx, lease.ID)
	if err != nil {
	return perrors.WithMessage(err, "keep alive lease")
	} else {
	return perrors.New("keep alive lease")
	}
	}

	_, err = c.rawClient.Put(c.ctx, k, v, clientv3.WithLease(lease.ID))
	return perrors.WithMessage(err, "put k/v with lease")
	}

	// nolint
	func (c *Client) Done() <-chan struct{} {
	return c.exit
	}

	// nolint
	func (c *Client) Valid() bool {
	select {
	case <-c.exit:
	return false
	default:
	}

	c.lock.RLock()
	defer c.lock.RUnlock()
	return c.rawClient != nil
	}

	// nolint
	func (c *Client) Create(k string, v string) error {
	err := c.put(k, v)
	return perrors.WithMessagef(err, "put k/v (key: %s value %s)", k, v)
	}

	// Update key value ...
	func (c *Client) Update(k, v string) error {
	err := c.update(k, v)
	return perrors.WithMessagef(err, "Update k/v (key: %s value %s)", k, v)
	}

	// nolint
	func (c *Client) Delete(k string) error {
	err := c.delete(k)
	return perrors.WithMessagef(err, "delete k/v (key %s)", k)
	}

	// RegisterTemp registers a temporary node
	func (c *Client) RegisterTemp(k, v string) error {
	err := c.keepAliveKV(k, v)
	return perrors.WithMessagef(err, "keepalive kv (key %s)", k)
	}

	// GetChildrenKVList gets children kv list by @k
	func (c *Client) GetChildrenKVList(k string) ([]string, []string, error) {
	kList, vList, err := c.getChildren(k)
	return kList, vList, perrors.WithMessagef(err, "get key children (key %s)", k)
	}

	// Get gets value by @k
	func (c *Client) Get(k string) (string, error) {
	v, err := c.get(k)
	return v, perrors.WithMessagef(err, "get key value (key %s)", k)
	}

	// Watch watches on spec key
	func (c *Client) Watch(k string) (clientv3.WatchChan, error) {
	wc, err := c.watch(k)
	return wc, perrors.WithMessagef(err, "watch prefix (key %s)", k)
	}

	// WatchWithPrefix watches on spec prefix
	func (c *Client) WatchWithPrefix(prefix string) (clientv3.WatchChan, error) {
	wc, err := c.watchWithPrefix(prefix)
	return wc, perrors.WithMessagef(err, "watch prefix (key %s)", prefix)
	}