[etcd] Expose cached etcd clients in cluster etcd config service

src/cluster/client/etcd/client.go

@@ -26,6 +26,7 @@ import (
 	"math/rand"
 	"os"
 	"path/filepath"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -55,12 +56,23 @@ const (
 
 var errInvalidNamespace = errors.New("invalid namespace")
 
+// make sure m3cluster and etcd client interfaces are implemented, and that
+// Client is a superset of cluster.Client.
+var _ client.Client = Client((*csclient)(nil))
+
 type newClientFn func(cluster Cluster) (*clientv3.Client, error)
 
 type cacheFileForZoneFn func(zone string) etcdkv.CacheFileFn
 
-// NewConfigServiceClient returns a ConfigServiceClient.
-func NewConfigServiceClient(opts Options) (client.Client, error) {
+// ZoneClient is a cached etcd client for a zone.
+type ZoneClient struct {
+	Client *clientv3.Client
+	Zone   string
+}
+
+// NewEtcdConfigServiceClient returns a new etcd-backed cluster client.
+//nolint:golint
+func NewEtcdConfigServiceClient(opts Options) (*csclient, error) {
 	if err := opts.Validate(); err != nil {
 		return nil, err
 	}
@@ -83,6 +95,11 @@ func NewConfigServiceClient(opts Options) (client.Client, error) {
 	}, nil
 }
 
+// NewConfigServiceClient returns a ConfigServiceClient.
+func NewConfigServiceClient(opts Options) (client.Client, error) {
+	return NewEtcdConfigServiceClient(opts)
+}
+
 type csclient struct {
 	sync.RWMutex
 	clis map[string]*clientv3.Client
@@ -276,6 +293,29 @@ func (c *csclient) etcdClientGen(zone string) (*clientv3.Client, error) {
 	return cli, nil
 }
 
+// Clients returns all currently cached etcd clients.
+func (c *csclient) Clients() []ZoneClient {
+	c.Lock()
+	defer c.Unlock()
+
+	var (
+		zones   = make([]string, 0, len(c.clis))
+		clients = make([]ZoneClient, 0, len(c.clis))
+	)
+
+	for k := range c.clis {
+		zones = append(zones, k)
+	}
+
+	sort.Strings(zones)
+
+	for _, zone := range zones {
+		clients = append(clients, ZoneClient{Zone: zone, Client: c.clis[zone]})
+	}
+
+	return clients
+}
+
 func newClient(cluster Cluster) (*clientv3.Client, error) {
 	tls, err := cluster.TLSOptions().Config()
 	if err != nil {

src/cluster/client/etcd/client_test.go

@@ -27,9 +27,10 @@ import (
 	"github.com/m3db/m3/src/cluster/kv"
 	"github.com/m3db/m3/src/cluster/services"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.etcd.io/etcd/clientv3"
 	"go.etcd.io/etcd/integration"
-	"github.com/stretchr/testify/require"
 )
 
 func TestETCDClientGen(t *testing.T) {
@@ -297,6 +298,29 @@ func TestReuseKVStore(t *testing.T) {
 	client.storeLock.Unlock()
 }
 
+func TestGetEtcdClients(t *testing.T) {
+	opts := testOptions()
+	c, err := NewEtcdConfigServiceClient(opts)
+	require.NoError(t, err)
+
+	c1, err := c.etcdClientGen("zone2")
+	require.NoError(t, err)
+	require.Equal(t, 1, len(c.clis))
+
+	c2, err := c.etcdClientGen("zone1")
+	require.NoError(t, err)
+	require.Equal(t, 2, len(c.clis))
+	require.False(t, c1 == c2)
+
+	clients := c.Clients()
+	require.Len(t, clients, 2)
+
+	assert.Equal(t, clients[0].Zone, "zone1")
+	assert.Equal(t, clients[0].Client, c2)
+	assert.Equal(t, clients[1].Zone, "zone2")
+	assert.Equal(t, clients[1].Client, c1)
+}
+
 func TestValidateNamespace(t *testing.T) {
 	inputs := []struct {
 		ns        string

src/cluster/client/etcd/types.go

@@ -25,6 +25,7 @@ import (
 	"os"
 	"time"
 
+	"github.com/m3db/m3/src/cluster/client"
 	"github.com/m3db/m3/src/cluster/services"
 	"github.com/m3db/m3/src/x/instrument"
 	"github.com/m3db/m3/src/x/retry"
@@ -135,3 +136,10 @@ type Cluster interface {
 	DialTimeout() time.Duration
 	SetDialTimeout(value time.Duration) Cluster
 }
+
+// Client is an etcd-backed m3cluster client.
+type Client interface {
+	client.Client
+
+	Clients() []ZoneClient
+}