etcd_docker 3: Incorporate docker based etcd integration package into… (

#4147)

PR 3 for #4144

High level approach is as described in #4144 .

This PR incorporates the new test package into our unittests. Usage is via the `etcdintegration` package, which
makes it transparent to the test code; it simply gets an etcd server started via different means.

One piece of weirdness to call out here: the package currently relies on autosync being *disabled* on the client side.
This is because the advertise client URL (aka what etcd tells clients to connect to) isn't correct for the open port on the host.

That is we have:

- etcd: listen on container port 0.0.0.0:2379, advertise 0.0.0.0:2379
- docker: expose etcd port 2379 to 0.0.0.0:0 on host machine (random free port)
- client: connect to etcd via host machine.

We could probably make this better.

commit-id:263fed13

go.mod

@@ -70,7 +70,6 @@ require (
 	go.etcd.io/etcd/client/pkg/v3 v3.6.0-alpha.0
 	go.etcd.io/etcd/client/v3 v3.6.0-alpha.0
 	go.etcd.io/etcd/server/v3 v3.6.0-alpha.0
-	go.etcd.io/etcd/tests/v3 v3.6.0-alpha.0
 	go.opentelemetry.io/collector v0.45.0
 	go.opentelemetry.io/otel v1.4.1
 	go.opentelemetry.io/otel/bridge/opentracing v1.4.1
@@ -121,7 +120,6 @@ require (
 	github.com/go-playground/locales v0.13.0 // indirect
 	github.com/go-playground/universal-translator v0.17.0 // indirect
 	github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
-	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/google/btree v1.0.1 // indirect
 	github.com/gorilla/handlers v1.5.1 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect

go.sum

@@ -1592,8 +1592,6 @@ go.etcd.io/etcd/raft/v3 v3.6.0-alpha.0 h1:BQ6CnNP4pIpy5rusFlTBxAacDgPXhuiHFwoTsB
 go.etcd.io/etcd/raft/v3 v3.6.0-alpha.0/go.mod h1:/kZdrBXlc5fUgYXfIEQ0B5sb7ejXPKbtF4jWzF1exiQ=
 go.etcd.io/etcd/server/v3 v3.6.0-alpha.0 h1:BQUVqBqNFZZyrRbfydrRLzq9hYvCcRj97SsX1YwD7CA=
 go.etcd.io/etcd/server/v3 v3.6.0-alpha.0/go.mod h1:3QM2rLq3B3hSXmVEvgVt3vEEbG/AumSs0Is7EgrlKzU=
-go.etcd.io/etcd/tests/v3 v3.6.0-alpha.0 h1:3qrZ3p/E7CxdV1kKtAU75hHOcUoXcSTwC7ELKWyzMJo=
-go.etcd.io/etcd/tests/v3 v3.6.0-alpha.0/go.mod h1:hFQkP/cTsZIXXvUv+BsGHZ3TK+76XZMi5GToYA94iac=
 go.mongodb.org/mongo-driver v1.0.3/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
 go.mongodb.org/mongo-driver v1.1.1/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
 go.mongodb.org/mongo-driver v1.1.2/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=

src/aggregator/integration/custom_aggregations_test.go

@@ -1,4 +1,4 @@
-// +build integration
+//go:build integration
 
 // Copyright (c) 2016 Uber Technologies, Inc.
 //
@@ -68,7 +68,6 @@ func testCustomAggregations(t *testing.T, metadataFns [4]metadataFn) {
 	if testing.Short() {
 		t.SkipNow()
 	}
-
 	aggTypesOpts := aggregation.NewTypesOptions().
 		SetCounterTypeStringTransformFn(aggregation.SuffixTransform).
 		SetTimerTypeStringTransformFn(aggregation.SuffixTransform).
@@ -179,7 +178,7 @@ func testCustomAggregations(t *testing.T, metadataFns [4]metadataFn) {
 	// must be the longer than the lowest resolution across all policies.
 	finalTime := end.Add(6 * time.Second)
 	clock.SetNow(finalTime)
-	time.Sleep(6 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	require.NoError(t, client.close())
 

src/aggregator/integration/election.go

@@ -26,9 +26,9 @@ import (
 	"github.com/m3db/m3/src/cluster/services"
 	"github.com/m3db/m3/src/cluster/services/leader"
 
+	integration "github.com/m3db/m3/src/integration/resources/docker/dockerexternal/etcdintegration"
 	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.etcd.io/etcd/tests/v3/framework/integration"
 )
 
 var (
@@ -40,27 +40,38 @@ var (
 )
 
 type testCluster struct {
-	t       *testing.T
-	cluster *integration.Cluster
+	t             *testing.T
+	cluster       *integration.Cluster
+	leaderService services.LeaderService
 }
 
 func newTestCluster(t *testing.T) *testCluster {
 	integration.BeforeTestExternal(t)
-	return &testCluster{
+	cluster := &testCluster{
 		t: t,
 		cluster: integration.NewCluster(t, &integration.ClusterConfig{
 			Size: testClusterSize,
 		}),
 	}
+	return cluster
 }
 
 func (tc *testCluster) LeaderService() services.LeaderService {
+	if tc.leaderService != nil {
+		return tc.leaderService
+	}
+
 	svc, err := leader.NewService(tc.etcdClient(), tc.options())
 	require.NoError(tc.t, err)
-	return svc
+	tc.leaderService = svc
+	return tc.leaderService
 }
 
 func (tc *testCluster) Close() {
+	if tc.leaderService != nil {
+		// amainsd: check error here!
+		_ = tc.leaderService.Close()
+	}
 	tc.cluster.Terminate(tc.t)
 }
 

src/aggregator/integration/metadata_change_test.go

@@ -1,4 +1,4 @@
-// +build integration
+//go:build integration
 
 // Copyright (c) 2016 Uber Technologies, Inc.
 //
@@ -138,7 +138,7 @@ func testMetadataChange(t *testing.T, oldMetadataFn, newMetadataFn metadataFn) {
 	// must be the longer than the lowest resolution across all policies.
 	finalTime := end.Add(6 * time.Second)
 	clock.SetNow(finalTime)
-	time.Sleep(6 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	require.NoError(t, client.close())
 

src/aggregator/integration/multi_client_one_type_test.go

@@ -1,4 +1,4 @@
-// +build integration
+//go:build integration
 
 // Copyright (c) 2016 Uber Technologies, Inc.
 //
@@ -126,7 +126,7 @@ func testMultiClientOneType(t *testing.T, metadataFn metadataFn) {
 	// must be the longer than the lowest resolution across all policies.
 	finalTime := stop.Add(6 * time.Second)
 	clock.SetNow(finalTime)
-	time.Sleep(4 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	for i := 0; i < numClients; i++ {
 		require.NoError(t, clients[i].close())

src/aggregator/integration/one_client_multi_type_forwarded_test.go

@@ -1,4 +1,4 @@
-// +build integration
+//go:build integration
 
 // Copyright (c) 2018 Uber Technologies, Inc.
 //
@@ -121,7 +121,7 @@ func TestOneClientMultiTypeForwardedMetrics(t *testing.T) {
 	// Move time forward and wait for flushing to happen.
 	finalTime := stop.Add(2 * time.Second)
 	clock.SetNow(finalTime)
-	time.Sleep(2 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	require.NoError(t, client.close())
 

src/aggregator/integration/one_client_multi_type_timed_test.go

@@ -1,4 +1,4 @@
-// +build integration
+//go:build integration
 
 // Copyright (c) 2018 Uber Technologies, Inc.
 //
@@ -119,7 +119,7 @@ func TestOneClientMultiTypeTimedMetrics(t *testing.T) {
 	// Move time forward and wait for flushing to happen.
 	finalTime := stop.Add(time.Minute + 2*time.Second)
 	clock.SetNow(finalTime)
-	time.Sleep(2 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	require.NoError(t, client.close())
 

src/aggregator/integration/one_client_multi_type_untimed_test.go

@@ -1,4 +1,4 @@
-// +build integration
+//go:build integration
 
 // Copyright (c) 2016 Uber Technologies, Inc.
 //
@@ -26,9 +26,17 @@ import (
 	"testing"
 	"time"
 
+	"github.com/m3db/m3/src/cluster/placement"
+
 	"github.com/stretchr/testify/require"
+)
 
-	"github.com/m3db/m3/src/cluster/placement"
+const (
+	// waitForDataToFlush is the amount of time we will wait in these tests between finishing writing data to
+	// the aggregator, and attempting to assert that data went through.
+	// The aggregator generally, and these tests specifically are quite sensitive to time.
+	// The tests probably need a bit of a rethink to wait on (or poll for) an actual condition instead of sleeping.
+	waitForDataToFlush = 10 * time.Second
 )
 
 func TestOneClientMultiTypeUntimedMetricsWithStagedMetadatas(t *testing.T) {
@@ -114,7 +122,7 @@ func testOneClientMultiType(t *testing.T, metadataFn metadataFn) {
 	// must be the longer than the lowest resolution across all policies.
 	finalTime := stop.Add(6 * time.Second)
 	clock.SetNow(finalTime)
-	time.Sleep(4 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	require.NoError(t, client.close())
 

src/aggregator/integration/one_client_passthru_test.go

@@ -1,4 +1,4 @@
-// +build integration
+//go:build integration
 
 // Copyright (c) 2020 Uber Technologies, Inc.
 //

src/aggregator/integration/placement_change_test.go

@@ -1,5 +1,4 @@
 //go:build integration
-// +build integration
 
 // Copyright (c) 2018 Uber Technologies, Inc.
 //
@@ -227,9 +226,9 @@ func TestPlacementChange(t *testing.T) {
 	}
 
 	clock.SetNow(start2)
-	time.Sleep(6 * time.Second)
+	time.Sleep(waitForDataToFlush)
 	setPlacement(t, placementKey, clusterClient, finalPlacement)
-	time.Sleep(6 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	for _, data := range datasets[1] {
 		clock.SetNow(data.timestamp)
@@ -245,7 +244,7 @@ func TestPlacementChange(t *testing.T) {
 
 	// Move time forward and wait for flushing to happen.
 	clock.SetNow(finalTime)
-	time.Sleep(6 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	// Remove all the topic consumers before closing clients and servers. This allows to close the
 	// connections between servers while they still are running. Otherwise, during server shutdown,

src/aggregator/integration/resend_stress_test.go

@@ -1,5 +1,4 @@
 //go:build integration
-// +build integration
 
 // Copyright (c) 2018 Uber Technologies, Inc.
 //

src/aggregator/integration/same_id_multi_type_test.go

@@ -1,4 +1,4 @@
-// +build integration
+//go:build integration
 
 // Copyright (c) 2016 Uber Technologies, Inc.
 //
@@ -138,7 +138,7 @@ func testSameIDMultiType(t *testing.T, metadataFn metadataFn) {
 	// must be the longer than the lowest resolution across all policies.
 	finalTime := stop.Add(6 * time.Second)
 	clock.SetNow(finalTime)
-	time.Sleep(4 * time.Second)
+	time.Sleep(waitForDataToFlush)
 
 	require.NoError(t, client.close())
 

src/aggregator/integration/setup.go

@@ -92,6 +92,7 @@ type testServerSetup struct {
 	// Signals.
 	doneCh   chan struct{}
 	closedCh chan struct{}
+	stopped  bool
 }
 
 func newTestServerSetup(t *testing.T, opts testServerOptions) *testServerSetup {
@@ -452,6 +453,10 @@ func (ts *testServerSetup) sortedResults() []aggregated.MetricWithStoragePolicy
 }
 
 func (ts *testServerSetup) stopServer() error {
+	if ts.stopped {
+		return nil
+	}
+	ts.stopped = true
 	if err := ts.aggregator.Close(); err != nil {
 		return err
 	}
@@ -464,6 +469,9 @@ func (ts *testServerSetup) stopServer() error {
 
 func (ts *testServerSetup) close() {
 	ts.electionCluster.Close()
+	if err := ts.stopServer(); err != nil {
+		panic(err.Error())
+	}
 }
 
 func (tss testServerSetups) newClient(t *testing.T) *client {

src/cluster/client/etcd/client.go

@@ -339,8 +339,14 @@ func newConfigFromCluster(rnd randInt63N, cluster Cluster) (clientv3.Config, err
 	if err != nil {
 		return clientv3.Config{}, err
 	}
+
+	// Support disabling autosync if a user very explicitly requests it (via negative duration).
+	autoSyncInterval := cluster.AutoSyncInterval()
+	if autoSyncInterval < 0 {
+		autoSyncInterval = 0
+	}
 	cfg := clientv3.Config{
-		AutoSyncInterval:   cluster.AutoSyncInterval(),
+		AutoSyncInterval:   autoSyncInterval,
 		DialTimeout:        cluster.DialTimeout(),
 		DialOptions:        cluster.DialOptions(),
 		Endpoints:          cluster.Endpoints(),

src/cluster/client/etcd/client_test.go

@@ -25,18 +25,23 @@ import (
 	"testing"
 	"time"
 
+	"github.com/m3db/m3/src/cluster/kv"
+	"github.com/m3db/m3/src/cluster/services"
+	integration "github.com/m3db/m3/src/integration/resources/docker/dockerexternal/etcdintegration"
+	"github.com/m3db/m3/src/x/retry"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.etcd.io/etcd/tests/v3/framework/integration"
 	"google.golang.org/grpc"
-
-	"github.com/m3db/m3/src/cluster/kv"
-	"github.com/m3db/m3/src/cluster/services"
 )
 
 func TestETCDClientGen(t *testing.T) {
-	cs, err := NewConfigServiceClient(testOptions())
+	cs, err := NewConfigServiceClient(
+		testOptions().
+			// These are error cases; don't retry for no reason.
+			SetRetryOptions(retry.NewOptions().SetMaxRetries(0)),
+	)
 	require.NoError(t, err)
 
 	c := cs.(*csclient)
@@ -414,6 +419,15 @@ func Test_newConfigFromCluster(t *testing.T) {
 		)
 	})
 
+	t.Run("negative autosync on M3 disables autosync for etcd", func(t *testing.T) {
+		inputCfg := newFullConfig()
+		inputCfg.AutoSyncInterval = -1
+		etcdCfg, err := newConfigFromCluster(testRnd, inputCfg.NewCluster())
+		require.NoError(t, err)
+
+		assert.Equal(t, time.Duration(0), etcdCfg.AutoSyncInterval)
+	})
+
 	// Separate test just because the assert.Equal won't work for functions.
 	t.Run("passes through dial options", func(t *testing.T) {
 		clusterCfg := newFullConfig()

src/cluster/client/etcd/config.go

@@ -35,12 +35,22 @@ import (
 
 // ClusterConfig is the config for a zoned etcd cluster.
 type ClusterConfig struct {
-	Zone             string           `yaml:"zone"`
-	Endpoints        []string         `yaml:"endpoints"`
-	KeepAlive        *KeepAliveConfig `yaml:"keepAlive"`
-	TLS              *TLSConfig       `yaml:"tls"`
-	AutoSyncInterval time.Duration    `yaml:"autoSyncInterval"`
-	DialTimeout      time.Duration    `yaml:"dialTimeout"`
+	Zone      string           `yaml:"zone"`
+	Endpoints []string         `yaml:"endpoints"`
+	KeepAlive *KeepAliveConfig `yaml:"keepAlive"`
+	TLS       *TLSConfig       `yaml:"tls"`
+	// AutoSyncInterval configures the etcd client's AutoSyncInterval
+	// (go.etcd.io/etcd/client/[email protected]/config.go:32).
+	// By default, it is 1m.
+	//
+	// Advanced:
+	//
+	// One important difference from the etcd config: we have autosync *on* by default (unlike etcd), meaning that
+	// the zero value here doesn't indicate autosync off.
+	// Instead, users should pass in a negative value to indicate "disable autosync"
+	// Only do this if you truly have a good reason for it! Most production use cases want autosync on.
+	AutoSyncInterval time.Duration `yaml:"autoSyncInterval"`
+	DialTimeout      time.Duration `yaml:"dialTimeout"`
 
 	DialOptions []grpc.DialOption `yaml:"-"` // nonserializable
 }
@@ -59,7 +69,10 @@ func (c ClusterConfig) NewCluster() Cluster {
 		SetKeepAliveOptions(keepAliveOpts).
 		SetTLSOptions(c.TLS.newOptions())
 
-	if c.AutoSyncInterval > 0 {
+	// Autosync should *always* be on, unless the user very explicitly requests it to be off. They can do this via a
+	// negative value (in which case we can assume they know what they're doing).
+	// Therefore, only update if it's nonzero, on the assumption that zero is just the empty value.
+	if c.AutoSyncInterval != 0 {
 		cluster = cluster.SetAutoSyncInterval(c.AutoSyncInterval)
 	}
 

src/cluster/client/etcd/config_test.go

@@ -181,3 +181,10 @@ func TestDefaultConfig(t *testing.T) {
 	require.Equal(t, defaultDialTimeout, cluster.DialTimeout())
 	require.Equal(t, defaultAutoSyncInterval, cluster.AutoSyncInterval())
 }
+
+func TestConfig_negativeAutosync(t *testing.T) {
+	cluster := ClusterConfig{
+		AutoSyncInterval: -5,
+	}.NewCluster()
+	require.Equal(t, time.Duration(-5), cluster.AutoSyncInterval())
+}