etcd_docker 3: Incorporate docker based etcd integration package into…

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/v3@v3.6.0-alpha.0/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())
+}