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

#4147)

* etcd_docker 2: Add a docker based etcdintegration package

PR 2 for #4144

High level approach is as described in #4144 . This PR adds:

- Functions to spin up a 1 node etcd cluster using docker (in `dockerexternal`)
- A drop in replacement for the etcd/integration package using `dockerexternal`

commit-id:e4e80f1d

* etcd_docker 3: Incorporate docker based etcd integration package into unittests.

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

docker-compose.yml

@@ -8,6 +8,17 @@ services:
     volumes:
       - .:/go/src/github.com/m3db/m3
       - /usr/bin/buildkite-agent:/usr/bin/buildkite-agent
+      # Support running docker within docker. That is, buildkite jobs themselves run in a container; that container
+      # needs to be able to spin up functioning docker containers.
+      - /var/run/docker.sock:/var/run/docker.sock
+    extra_hosts:
+      # Allow routing from the buildkite container to the host machine, as host.docker.internal. This allows us to do
+      # the following:
+      #  - Spin up an etcd container with ports published to the host machine
+      #  - Connect to the etcd container from the buildkite test process using host.docker.internal
+      # See
+      # https://medium.com/@TimvanBaarsen/how-to-connect-to-the-docker-host-from-inside-a-docker-container-112b4c71bc66
+      - "host.docker.internal:host-gateway"
     environment:
       - CI
       - BUILDKITE

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,
+			//	UseBridge: true,
 		}),
 	}
+	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 {
+		require.NoError(tc.t, 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/multi_server_forwarding_pipeline_test.go

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

src/aggregator/integration/multi_server_resend_test.go

@@ -1,5 +1,4 @@
 //go:build integration
-// +build integration
 
 // Copyright (c) 2018 Uber Technologies, Inc.
 //
@@ -142,6 +141,7 @@ func TestMultiServerResendAggregatedValues(t *testing.T) {
 
 	// Election cluster setup.
 	electionCluster := newTestCluster(t)
+	defer electionCluster.Close()
 
 	// Sharding function maps all metrics to shard 0 except for the rollup metric,
 	// which gets mapped to the last shard.

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

@@ -88,6 +88,7 @@ type testServerSetup struct {
 	// Signals.
 	doneCh   chan struct{}
 	closedCh chan struct{}
+	stopped  bool
 }
 
 func newTestServerSetup(t *testing.T, opts testServerOptions) *testServerSetup {
@@ -448,6 +449,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
 	}
@@ -460,6 +465,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()