func (c *EtcdLock) isHeld() (bool, error) {
	if c.semaphoreKey == "" {
		return false, nil
	}
	currentSemaphoreKey, _, _, err := c.getSemaphoreKey()
	if err != nil {
		return false, err
	}
	return c.semaphoreKey == currentSemaphoreKey, nil
}
isHeld determines if we are the current holders of the lock.
package physical
import (
	"encoding/base64"
	"errors"
	"fmt"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"
	log "github.com/mgutz/logxi/v1"
	"github.com/armon/go-metrics"
	"github.com/coreos/etcd/client"
	"github.com/coreos/etcd/pkg/transport"
	"golang.org/x/net/context"
)
const (
	EtcdNodeFilePrefix = "."
	EtcdNodeLockPrefix = "_"
	EtcdMachineDelimiter = ","
	EtcdLockTTL = 15 * time.Second
	EtcdLockRenewInterval = 5 * time.Second
	EtcdWatchRetryInterval = time.Second
	EtcdWatchRetryMax = 5
)
var (
	EtcdSyncConfigError          = errors.New("client setup failed: unable to parse etcd sync field in config")
	EtcdSyncClusterError         = errors.New("client setup failed: unable to sync etcd cluster")
	EtcdAddressError             = errors.New("client setup failed: address must be valid URL (ex. 'scheme://host:port')")
	EtcdSemaphoreKeysEmptyError  = errors.New("lock queue is empty")
	EtcdLockHeldError            = errors.New("lock already held")
	EtcdLockNotHeldError         = errors.New("lock not held")
	EtcdSemaphoreKeyRemovedError = errors.New("semaphore key removed before lock aquisition")
)
func errorIsMissingKey(err error) bool {
	etcdErr, ok := err.(client.Error)
	return ok && etcdErr.Code == client.ErrorCodeKeyNotFound
}
type EtcdBackend struct {
	path       string
	kAPI       client.KeysAPI
	permitPool *PermitPool
	logger     log.Logger
	haEnabled  bool
}
func newEtcdBackend(conf map[string]string, logger log.Logger) (Backend, error) {
	path, ok := conf["path"]
	if !ok {
		path = "/vault"
	}
	if !strings.HasPrefix(path, "/") {
		path = "/" + path
	}
	machines := "http://127.0.0.1:2379"
	if address, ok := conf["address"]; ok {
		machines = address
	}
	machinesEnv := os.Getenv("ETCD_ADDR")
	if machinesEnv != "" {
		machines = machinesEnv
	}
	machinesParsed := strings.Split(machines, EtcdMachineDelimiter)
	for _, machine := range machinesParsed {
		u, urlErr := url.Parse(machine)
		if urlErr != nil || u.Scheme == "" {
			return nil, EtcdAddressError
		}
	}
	haEnabled := os.Getenv("ETCD_HA_ENABLED")
	if haEnabled == "" {
		haEnabled = conf["ha_enabled"]
	}
	haEnabledBool, _ := strconv.ParseBool(haEnabled)
	var cTransport client.CancelableTransport
	cert, hasCert := conf["tls_cert_file"]
	key, hasKey := conf["tls_key_file"]
	ca, hasCa := conf["tls_ca_file"]
	if (hasCert && hasKey) || hasCa {
		var transportErr error
		tls := transport.TLSInfo{
			CAFile:   ca,
			CertFile: cert,
			KeyFile:  key,
		}
		cTransport, transportErr = transport.NewTransport(tls, 30*time.Second)
		if transportErr != nil {
			return nil, transportErr
		}
	} else {
		cTransport = client.DefaultTransport
	}
	cfg := client.Config{
		Endpoints: machinesParsed,
		Transport: cTransport,
	}
	username := os.Getenv("ETCD_USERNAME")
	if username == "" {
		username, _ = conf["username"]
	}
	password := os.Getenv("ETCD_PASSWORD")
	if password == "" {
		password, _ = conf["password"]
	}
	if username != "" && password != "" {
		cfg.Username = username
		cfg.Password = password
	}
	c, err := client.New(cfg)
	if err != nil {
		return nil, err
	}
	sync, ok := conf["sync"]
	if !ok {
		sync = "yes"
	}
	switch sync {
	case "yes", "true", "y", "1":
		ctx, cancel := context.WithTimeout(context.Background(), client.DefaultRequestTimeout)
		syncErr := c.Sync(ctx)
		cancel()
		if syncErr != nil {
			return nil, fmt.Errorf("%s: %s", EtcdSyncClusterError, syncErr)
		}
	case "no", "false", "n", "0":
	default:
		return nil, fmt.Errorf("value of 'sync' could not be understood")
	}
	kAPI := client.NewKeysAPI(c)
	return &EtcdBackend{
		path:       path,
		kAPI:       kAPI,
		permitPool: NewPermitPool(DefaultParallelOperations),
		logger:     logger,
		haEnabled:  haEnabledBool,
	}, nil
}
func (c *EtcdBackend) Put(entry *Entry) error {
	defer metrics.MeasureSince([]string{"etcd", "put"}, time.Now())
	value := base64.StdEncoding.EncodeToString(entry.Value)
	c.permitPool.Acquire()
	defer c.permitPool.Release()
	_, err := c.kAPI.Set(context.Background(), c.nodePath(entry.Key), value, nil)
	return err
}
func (c *EtcdBackend) Get(key string) (*Entry, error) {
	defer metrics.MeasureSince([]string{"etcd", "get"}, time.Now())
	c.permitPool.Acquire()
	defer c.permitPool.Release()
	getOpts := &client.GetOptions{
		Recursive: false,
		Sort:      false,
	}
	response, err := c.kAPI.Get(context.Background(), c.nodePath(key), getOpts)
	if err != nil {
		if errorIsMissingKey(err) {
			return nil, nil
		}
		return nil, err
	}
	value, err := base64.StdEncoding.DecodeString(response.Node.Value)
	if err != nil {
		return nil, err
	}
	return &Entry{
		Key:   key,
		Value: value,
	}, nil
}
func (c *EtcdBackend) Delete(key string) error {
	defer metrics.MeasureSince([]string{"etcd", "delete"}, time.Now())
	c.permitPool.Acquire()
	defer c.permitPool.Release()
	delOpts := &client.DeleteOptions{
		Recursive: false,
	}
	_, err := c.kAPI.Delete(context.Background(), c.nodePath(key), delOpts)
	if err != nil && !errorIsMissingKey(err) {
		return err
	}
	return nil
}
func (c *EtcdBackend) List(prefix string) ([]string, error) {
	defer metrics.MeasureSince([]string{"etcd", "list"}, time.Now())
	path := c.nodePathDir(prefix)
	c.permitPool.Acquire()
	defer c.permitPool.Release()
	getOpts := &client.GetOptions{
		Recursive: false,
		Sort:      true,
	}
	response, err := c.kAPI.Get(context.Background(), path, getOpts)
	if err != nil {
		if errorIsMissingKey(err) {
			return []string{}, nil
		}
		return nil, err
	}
	out := make([]string, len(response.Node.Nodes))
	for i, node := range response.Node.Nodes {
		name := strings.TrimPrefix(node.Key, path)
		if node.Dir {
			out[i] = name + "/"
		} else {
			out[i] = name[1:]
		}
	}
	return out, nil
}
func (b *EtcdBackend) nodePath(key string) string {
	return filepath.Join(b.path, filepath.Dir(key), EtcdNodeFilePrefix+filepath.Base(key))
}
func (b *EtcdBackend) nodePathDir(key string) string {
	return filepath.Join(b.path, key) + "/"
}
func (b *EtcdBackend) nodePathLock(key string) string {
	return filepath.Join(b.path, filepath.Dir(key), EtcdNodeLockPrefix+filepath.Base(key)+"/")
}
func (c *EtcdBackend) LockWith(key, value string) (Lock, error) {
	return &EtcdLock{
		kAPI:            c.kAPI,
		value:           value,
		semaphoreDirKey: c.nodePathLock(key),
	}, nil
}
func (e *EtcdBackend) HAEnabled() bool {
	return e.haEnabled
}
type EtcdLock struct {
	kAPI                                 client.KeysAPI
	value, semaphoreDirKey, semaphoreKey string
	lock                                 sync.Mutex
}
func (c *EtcdLock) addSemaphoreKey() (string, uint64, error) {
	opts := &client.CreateInOrderOptions{
		TTL: EtcdLockTTL,
	}
	response, err := c.kAPI.CreateInOrder(context.Background(), c.semaphoreDirKey, c.value, opts)
	if err != nil {
		return "", 0, err
	}
	return response.Node.Key, response.Index, nil
}
func (c *EtcdLock) renewSemaphoreKey() (string, uint64, error) {
	setOpts := &client.SetOptions{
		TTL:       EtcdLockTTL,
		PrevExist: client.PrevExist,
	}
	response, err := c.kAPI.Set(context.Background(), c.semaphoreKey, c.value, setOpts)
	if err != nil {
		return "", 0, err
	}
	return response.Node.Key, response.Index, nil
}
func (c *EtcdLock) getSemaphoreKey() (string, string, uint64, error) {
	getOpts := &client.GetOptions{
		Recursive: false,
		Sort:      true,
	}
	response, err := c.kAPI.Get(context.Background(), c.semaphoreDirKey, getOpts)
	if err != nil {
		return "", "", 0, err
	}
	if response.Node.Nodes.Len() == 0 {
		return "", "", response.Index, nil
	}
	return response.Node.Nodes[0].Key, response.Node.Nodes[0].Value, response.Index, nil
}