func (s *EtcdV3) DeleteTree(directory string) error {
	ctx, cancel := context.WithTimeout(context.Background(), etcdDefaultTimeout)
	resp, err := s.client.KV.Delete(ctx, s.normalize(directory), etcd.WithPrefix())
	cancel()
	if err != nil {
		return err
	}
	if resp.Deleted == 0 {
		return store.ErrKeyNotFound
	}
	return nil
}
DeleteTree deletes a range of keys under a given directory
package etcdv3
import (
	"context"
	"crypto/tls"
	"strings"
	"sync"
	"time"
	"github.com/abronan/valkeyrie"
	"github.com/abronan/valkeyrie/store"
	etcd "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
)
const (
	defaultLockTTL     = 20 * time.Second
	etcdDefaultTimeout = 5 * time.Second
	lockSuffix         = "___lock"
)
type EtcdV3 struct {
	client *etcd.Client
}
type etcdLock struct {
	lock  sync.Mutex
	store *EtcdV3
	mutex   *concurrency.Mutex
	session *concurrency.Session
	mutexKey       string  
	writeKey       string  
	value          string
	ttl            time.Duration
	deleteOnUnlock bool
}
func Register() {
	valkeyrie.AddStore(store.ETCDV3, New)
}
func New(addrs []string, options *store.Config) (store.Store, error) {
	s := &EtcdV3{}
	var (
		entries []string
		err     error
	)
	entries = store.CreateEndpoints(addrs, "http")
	cfg := &etcd.Config{
		Endpoints: entries,
	}
	if options != nil {
		if options.TLS != nil {
			setTLS(cfg, options.TLS, addrs)
		}
		if options.ConnectionTimeout != 0 {
			setTimeout(cfg, options.ConnectionTimeout)
		}
		if options.Username != "" {
			setCredentials(cfg, options.Username, options.Password)
		}
		if options.SyncPeriod != 0 {
			cfg.AutoSyncInterval = options.SyncPeriod
		}
	}
	s.client, err = etcd.New(*cfg)
	if err != nil {
		return nil, err
	}
	return s, nil
}
func setTLS(cfg *etcd.Config, tls *tls.Config, addrs []string) {
	entries := store.CreateEndpoints(addrs, "https")
	cfg.Endpoints = entries
	cfg.TLS = tls
}
func setTimeout(cfg *etcd.Config, time time.Duration) {
	cfg.DialTimeout = time
}
func setCredentials(cfg *etcd.Config, username, password string) {
	cfg.Username = username
	cfg.Password = password
}
func (s *EtcdV3) normalize(key string) string {
	key = store.Normalize(key)
	return strings.TrimPrefix(key, "/")
}
func (s *EtcdV3) Get(key string, opts *store.ReadOptions) (pair *store.KVPair, err error) {
	ctx, cancel := context.WithTimeout(context.Background(), etcdDefaultTimeout)
	var result *etcd.GetResponse
	if opts != nil && !opts.Consistent {
		result, err = s.client.KV.Get(ctx, s.normalize(key), etcd.WithSerializable())
	} else {
		result, err = s.client.KV.Get(ctx, s.normalize(key))
	}
	cancel()
	if err != nil {
		return nil, err
	}
	if result.Count == 0 {
		return nil, store.ErrKeyNotFound
	}
	kvs := []*store.KVPair{}
	for _, pair := range result.Kvs {
		kvs = append(kvs, &store.KVPair{
			Key:       string(pair.Key),
			Value:     pair.Value,
			LastIndex: uint64(pair.ModRevision),
		})
	}
	return kvs[0], nil
}
func (s *EtcdV3) Put(key string, value []byte, opts *store.WriteOptions) (err error) {
	ctx, cancel := context.WithTimeout(context.Background(), etcdDefaultTimeout)
	pr := s.client.Txn(ctx)
	if opts != nil && opts.TTL > 0 {
		lease := etcd.NewLease(s.client)
		grant, err := lease.Grant(context.Background(), int64(opts.TTL/time.Second))
		if err != nil {
			cancel()
			return err
		}
		if opts.KeepAlive {
			_, err = lease.KeepAlive(context.Background(), grant.ID)
			if err != nil {
				cancel()
				return err
			}
		}
		pr.Then(etcd.OpPut(key, string(value), etcd.WithLease(grant.ID)))
	} else {
		pr.Then(etcd.OpPut(key, string(value)))
	}
	_, err = pr.Commit()
	cancel()
	if err != nil {
		return err
	}
	return nil
}
func (s *EtcdV3) Delete(key string) error {
	resp, err := s.client.KV.Delete(context.Background(), s.normalize(key))
	if resp != nil && resp.Deleted == 0 {
		return store.ErrKeyNotFound
	}
	return err
}
func (s *EtcdV3) Exists(key string, opts *store.ReadOptions) (bool, error) {
	_, err := s.Get(key, opts)
	if err != nil {
		if err == store.ErrKeyNotFound {
			return false, nil
		}
		return false, err
	}
	return true, nil
}
func (s *EtcdV3) Watch(key string, stopCh <-chan struct{}, opts *store.ReadOptions) (<-chan *store.KVPair, error) {
	wc := etcd.NewWatcher(s.client)
	respCh := make(chan *store.KVPair)
	pair, err := s.Get(key, opts)
	if err != nil {
		return nil, err
	}
	go func() {
		defer wc.Close()
		defer close(respCh)
		respCh <- pair
		watchCh := wc.Watch(context.Background(), s.normalize(key))
		for resp := range watchCh {
			select {
			case <-stopCh:
				return
			default:
			}
			for _, ev := range resp.Events {
				respCh <- &store.KVPair{
					Key:       key,
					Value:     ev.Kv.Value,
					LastIndex: uint64(ev.Kv.ModRevision),
				}
			}
		}
	}()
	return respCh, nil
}
func (s *EtcdV3) WatchTree(directory string, stopCh <-chan struct{}, opts *store.ReadOptions) (<-chan []*store.KVPair, error) {
	wc := etcd.NewWatcher(s.client)
	respCh := make(chan []*store.KVPair)
	rev, pairs, err := s.list(directory, opts)
	if err != nil {
		return nil, err
	}
	go func() {
		defer wc.Close()
		defer close(respCh)
		respCh <- pairs
		rev++
		watchCh := wc.Watch(context.Background(), s.normalize(directory), etcd.WithPrefix(), etcd.WithRev(rev))
		for resp := range watchCh {
			select {
			case <-stopCh:
				return
			default:
			}
			list := make([]*store.KVPair, len(resp.Events))
			for i, ev := range resp.Events {
				list[i] = &store.KVPair{
					Key:       string(ev.Kv.Key),
					Value:     ev.Kv.Value,
					LastIndex: uint64(ev.Kv.ModRevision),
				}
			}
			respCh <- list
		}
	}()
	return respCh, nil
}
func (s *EtcdV3) AtomicPut(key string, value []byte, previous *store.KVPair, opts *store.WriteOptions) (bool, *store.KVPair, error) {
	var cmp etcd.Cmp
	var testIndex bool
	if previous != nil {
		testIndex = true
		cmp = etcd.Compare(etcd.ModRevision(key), "=", int64(previous.LastIndex))
	} else {
		testIndex = false
		cmp = etcd.Compare(etcd.CreateRevision(key), "=", 0)
	}
	ctx, cancel := context.WithTimeout(context.Background(), etcdDefaultTimeout)
	pr := s.client.Txn(ctx).If(cmp)
	if opts != nil && opts.TTL > 0 {
		lease := etcd.NewLease(s.client)
		resp, err := lease.Grant(context.Background(), int64(opts.TTL/time.Second))
		if err != nil {
			cancel()
			return false, nil, err
		}
		pr.Then(etcd.OpPut(key, string(value), etcd.WithLease(resp.ID)))
	} else {
		pr.Then(etcd.OpPut(key, string(value)))
	}
	txn, err := pr.Commit()
	cancel()
	if err != nil {
		return false, nil, err
	}
	if !txn.Succeeded {
		if testIndex {
			return false, nil, store.ErrKeyModified
		}
		return false, nil, store.ErrKeyExists
	}
	updated := &store.KVPair{
		Key:       key,
		Value:     value,
		LastIndex: uint64(txn.Header.Revision),
	}
	return true, updated, nil
}
func (s *EtcdV3) AtomicDelete(key string, previous *store.KVPair) (bool, error) {
	if previous == nil {
		return false, store.ErrPreviousNotSpecified
	}
	cmp := etcd.Compare(etcd.ModRevision(key), "=", int64(previous.LastIndex))
	ctx, cancel := context.WithTimeout(context.Background(), etcdDefaultTimeout)
	txn, err := s.client.Txn(ctx).
		If(cmp).
		Then(etcd.OpDelete(key)).
		Commit()
	cancel()
	if err != nil {
		return false, err
	}
	if len(txn.Responses) == 0 {
		return false, store.ErrKeyNotFound
	}
	if !txn.Succeeded {
		return false, store.ErrKeyModified
	}
	return true, nil
}
func (s *EtcdV3) List(directory string, opts *store.ReadOptions) ([]*store.KVPair, error) {
	_, kv, err := s.list(directory, opts)
	return kv, err
}