func (c *csclient) txnGen(
	opts kv.OverrideOptions,
	cacheFileFn cacheFileForZoneFn,
) (kv.TxnStore, error) {
	cli, err := c.etcdClientGen(opts.Zone())
	if err != nil {
		return nil, err
	}
	c.storeLock.Lock()
	defer c.storeLock.Unlock()
	key := kvStoreCacheKey(opts.Zone(), opts.Namespace(), opts.Environment())
	store, ok := c.stores[key]
	if ok {
		return store, nil
	}
	if store, err = etcdkv.NewStore(
		cli.KV,
		cli.Watcher,
		c.newkvOptions(opts, cacheFileFn),
	); err != nil {
		return nil, err
	}
	c.stores[key] = store
	return store, nil
}
txnGen assumes the caller has validated the options passed if they are
user-supplied (as opposed to constructed ourselves).
package etcd
import (
	"errors"
	"fmt"
	"math/rand"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
	"github.com/m3db/m3cluster/client"
	"github.com/m3db/m3cluster/kv"
	etcdkv "github.com/m3db/m3cluster/kv/etcd"
	"github.com/m3db/m3cluster/services"
	etcdheartbeat "github.com/m3db/m3cluster/services/heartbeat/etcd"
	"github.com/m3db/m3cluster/services/leader"
	"github.com/m3db/m3x/instrument"
	"github.com/m3db/m3x/log"
	"github.com/m3db/m3x/retry"
	"github.com/coreos/etcd/clientv3"
	"github.com/uber-go/tally"
)
const (
	hierarchySeparator = "/"
	internalPrefix     = "_"
	cacheFileSeparator = "_"
	cacheFileSuffix    = ".json"
	kvPrefix = "_kv"
)
var errInvalidNamespace = errors.New("invalid namespace")
type newClientFn func(cluster Cluster) (*clientv3.Client, error)
type cacheFileForZoneFn func(zone string) etcdkv.CacheFileFn
func NewConfigServiceClient(opts Options) (client.Client, error) {
	if err := opts.Validate(); err != nil {
		return nil, err
	}
	scope := opts.InstrumentOptions().
		MetricsScope().
		Tagged(map[string]string{"service": opts.Service()})
	return &csclient{
		opts:    opts,
		sdOpts:  opts.ServicesOptions(),
		kvScope: scope.Tagged(map[string]string{"config_service": "kv"}),
		sdScope: scope.Tagged(map[string]string{"config_service": "sd"}),
		hbScope: scope.Tagged(map[string]string{"config_service": "hb"}),
		clis:    make(map[string]*clientv3.Client),
		logger:  opts.InstrumentOptions().Logger(),
		newFn:   newClient,
		retrier: retry.NewRetrier(opts.RetryOptions()),
		stores:  make(map[string]kv.TxnStore),
	}, nil
}
type csclient struct {
	sync.RWMutex
	clis map[string]*clientv3.Client
	opts    Options
	sdOpts  services.Options
	kvScope tally.Scope
	sdScope tally.Scope
	hbScope tally.Scope
	logger  log.Logger
	newFn   newClientFn
	retrier retry.Retrier
	storeLock sync.Mutex
	stores    map[string]kv.TxnStore
}
func (c *csclient) Services(opts services.OverrideOptions) (services.Services, error) {
	if opts == nil {
		opts = services.NewOverrideOptions()
	}
	return c.createServices(opts)
}
func (c *csclient) KV() (kv.Store, error) {
	return c.Txn()
}
func (c *csclient) Txn() (kv.TxnStore, error) {
	return c.TxnStore(kv.NewOverrideOptions())
}
func (c *csclient) Store(opts kv.OverrideOptions) (kv.Store, error) {
	return c.TxnStore(opts)
}
func (c *csclient) TxnStore(opts kv.OverrideOptions) (kv.TxnStore, error) {
	opts, err := c.sanitizeOptions(opts)
	if err != nil {
		return nil, err
	}
	return c.createTxnStore(opts)
}
func (c *csclient) createServices(opts services.OverrideOptions) (services.Services, error) {
	nOpts := opts.NamespaceOptions()
	cacheFileExtraFields := []string{nOpts.PlacementNamespace(), nOpts.MetadataNamespace()}
	return services.NewServices(c.sdOpts.
		SetHeartbeatGen(c.heartbeatGen()).
		SetKVGen(c.kvGen(c.cacheFileFn(cacheFileExtraFields...))).
		SetLeaderGen(c.leaderGen()).
		SetNamespaceOptions(nOpts).
		SetInstrumentsOptions(instrument.NewOptions().
			SetLogger(c.logger).
			SetMetricsScope(c.sdScope),
		),
	)
}
func (c *csclient) createTxnStore(opts kv.OverrideOptions) (kv.TxnStore, error) {
	if err := opts.Validate(); err != nil {
		return nil, err
	}
	return c.txnGen(opts, c.cacheFileFn())
}
func (c *csclient) kvGen(fn cacheFileForZoneFn) services.KVGen {
	return services.KVGen(func(zone string) (kv.Store, error) {
		opts := kv.NewOverrideOptions().SetZone(zone)
		return c.txnGen(opts, fn)
	})
}
func (c *csclient) newkvOptions(
	opts kv.OverrideOptions,
	cacheFileFn cacheFileForZoneFn,
) etcdkv.Options {
	kvOpts := etcdkv.NewOptions().
		SetInstrumentsOptions(instrument.NewOptions().
			SetLogger(c.logger).
			SetMetricsScope(c.kvScope)).
		SetCacheFileFn(cacheFileFn(opts.Zone())).
		SetWatchWithRevision(c.opts.WatchWithRevision())
	if ns := opts.Namespace(); ns != "" {
		kvOpts = kvOpts.SetPrefix(kvOpts.ApplyPrefix(ns))
	}
	if env := opts.Environment(); env != "" {
		kvOpts = kvOpts.SetPrefix(kvOpts.ApplyPrefix(env))
	}
	return kvOpts
}