[cluster] Fix etcd config default value handling (m3db#4130)

scripts/docker-integration-tests/aggregator/m3aggregator.yml

@@ -17,15 +17,11 @@ m3msg:
   server:
     listenAddress: 0.0.0.0:6000
     retry:
-      maxBackoff: 10s
+      maxBackoff: 1s
       jitter: true
   consumer:
     messagePool:
-      size: 16384
-      watermark:
-        low: 0.2
-        high: 0.5
-
+      size: dynamic
 http:
   listenAddress: 0.0.0.0:6001
   readTimeout: 60s
@@ -39,6 +35,8 @@ kvClient:
     cacheDir: /var/lib/m3kv
     etcdClusters:
       - zone: embedded
+        autoSyncInterval: 10m
+        dialTimeout: 1m
         endpoints:
           - dbnode01:2379
 
@@ -68,7 +66,7 @@ aggregator:
     timerTransformFnType: suffix
     gaugeTransformFnType: empty
     aggregationTypesPool:
-      size: 1024
+      size: 32
     quantilesPool:
       buckets:
         - count: 256
@@ -105,18 +103,15 @@ aggregator:
             namespaces:
               placement: /placement
           messagePool:
-            size: 16384
-            watermark:
-              low: 0.2
-              high: 0.5
+            size: dynamic
   placementManager:
     kvConfig:
       namespace: /placement
       environment: override_test_env
       zone: embedded
     placementWatcher:
       key: m3aggregator
-      initWatchTimeout: 10s
+      initWatchTimeout: 1s
   hashType: murmur32
   bufferDurationBeforeShardCutover: 10m
   bufferDurationAfterShardCutoff: 10m
@@ -130,8 +125,8 @@ aggregator:
     flushTimesKeyFmt: shardset/%d/flush
     flushTimesPersistRetrier:
       initialBackoff: 100ms
-      backoffFactor: 2.0
-      maxBackoff: 2s
+      backoffFactor: 1.0
+      maxBackoff: 500ms
       maxRetries: 3
   electionManager:
     election:
@@ -146,38 +141,27 @@ aggregator:
     campaignRetrier:
       initialBackoff: 100ms
       backoffFactor: 2.0
-      maxBackoff: 2s
+      maxBackoff: 1s
       forever: true
       jitter: true
     changeRetrier:
       initialBackoff: 100ms
       backoffFactor: 2.0
-      maxBackoff: 5s
+      maxBackoff: 1s
       forever: true
       jitter: true
     resignRetrier:
       initialBackoff: 100ms
       backoffFactor: 2.0
-      maxBackoff: 5s
+      maxBackoff: 1s
       forever: true
       jitter: true
     campaignStateCheckInterval: 1s
     shardCutoffCheckOffset: 30s
   flushManager:
-    checkEvery: 1s
-    jitterEnabled: true
-    maxJitters:
-      - flushInterval: 5s
-        maxJitterPercent: 1.0
-      - flushInterval: 10s
-        maxJitterPercent: 0.5
-      - flushInterval: 1m
-        maxJitterPercent: 0.5
-      - flushInterval: 10m
-        maxJitterPercent: 0.5
-      - flushInterval: 1h
-        maxJitterPercent: 0.25
-    numWorkersPerCPU: 0.5 
+    checkEvery: 500ms
+    jitterEnabled: false
+    numWorkersPerCPU: 0.1
     maxBufferSize: 5m
     forcedFlushWindowSize: 10s
   flush:
@@ -192,14 +176,11 @@ aggregator:
                 zone: embedded
                 environment: override_test_env
               messagePool:
-                size: 16384
-                watermark:
-                  low: 0.2
-                  high: 0.5
+                size: dynamic
   passthrough:
     enabled: true
   forwarding:
-    maxConstDelay: 1m # Need to add some buffer window, since timed metrics by default are delayed by 1min.
+    maxConstDelay: 65s # Need to add some buffer window, since timed metrics by default are delayed by 1min.
   entryTTL: 1h
   entryCheckInterval: 10m
   maxTimerBatchSizePerWrite: 140

scripts/docker-integration-tests/aggregator/m3coordinator.yml

@@ -21,6 +21,8 @@ clusters:
           cacheDir: /var/lib/m3kv
           etcdClusters:
             - zone: embedded
+              autoSyncInterval: 10m
+              dialTimeout: 1m
               endpoints:
                 - dbnode01:2379
 
@@ -55,18 +57,13 @@ downsample:
               namespaces:
                 placement: /placement
             connection:
-              numConnections: 4
-            messagePool:
-              size: 16384
-              watermark:
-                low: 0.2
-                high: 0.5
+              numConnections: 1
 
 ingest:
   ingester:
-    workerPoolSize: 10000
+    workerPoolSize: 50
     opPool:
-      size: 10000
+      size: dynamic
     retry:
       maxRetries: 3
       jitter: true
@@ -75,7 +72,7 @@ ingest:
     server:
       listenAddress: "0.0.0.0:7507"
       retry:
-        maxBackoff: 10s
+        maxBackoff: 1s
         jitter: true
 
 storeMetricsType: true

src/cluster/client/etcd/client_test.go

@@ -387,7 +387,8 @@ func Test_newConfigFromCluster(t *testing.T) {
 				Timeout: 7 * time.Second,
 			},
 			TLS:              nil, // TODO: TLS config gets read eagerly here; test it separately.
-			AutoSyncInterval: 20 * time.Second,
+			AutoSyncInterval: 21 * time.Second,
+			DialTimeout:      42 * time.Second,
 		}
 	}
 
@@ -398,10 +399,10 @@ func Test_newConfigFromCluster(t *testing.T) {
 		assert.Equal(t,
 			clientv3.Config{
 				Endpoints:            []string{"i1"},
-				AutoSyncInterval:     20000000000,
-				DialTimeout:          15000000000,
-				DialKeepAliveTime:    5000000010, // generated using fake rnd above
-				DialKeepAliveTimeout: 7000000000,
+				AutoSyncInterval:     21 * time.Second,
+				DialTimeout:          42 * time.Second,
+				DialKeepAliveTime:    5*time.Second + 10, // generated using fake rnd above
+				DialKeepAliveTimeout: 7 * time.Second,
 				MaxCallSendMsgSize:   33554432,
 				MaxCallRecvMsgSize:   33554432,
 				RejectOldCluster:     false,

src/cluster/client/etcd/config.go

@@ -40,6 +40,7 @@ type ClusterConfig struct {
 	KeepAlive        *KeepAliveConfig `yaml:"keepAlive"`
 	TLS              *TLSConfig       `yaml:"tls"`
 	AutoSyncInterval time.Duration    `yaml:"autoSyncInterval"`
+	DialTimeout      time.Duration    `yaml:"dialTimeout"`
 
 	DialOptions []grpc.DialOption `yaml:"-"` // nonserializable
 }
@@ -50,13 +51,23 @@ func (c ClusterConfig) NewCluster() Cluster {
 	if c.KeepAlive != nil {
 		keepAliveOpts = c.KeepAlive.NewOptions()
 	}
-	return NewCluster().
+
+	cluster := NewCluster().
 		SetZone(c.Zone).
 		SetEndpoints(c.Endpoints).
 		SetDialOptions(c.DialOptions).
 		SetKeepAliveOptions(keepAliveOpts).
-		SetTLSOptions(c.TLS.newOptions()).
-		SetAutoSyncInterval(c.AutoSyncInterval)
+		SetTLSOptions(c.TLS.newOptions())
+
+	if c.AutoSyncInterval > 0 {
+		cluster = cluster.SetAutoSyncInterval(c.AutoSyncInterval)
+	}
+
+	if c.DialTimeout > 0 {
+		cluster = cluster.SetDialTimeout(c.DialTimeout)
+	}
+
+	return cluster
 }
 
 // TLSConfig is the config for TLS.

src/cluster/client/etcd/config_test.go

@@ -64,7 +64,8 @@ etcdClusters:
       period: 10s
       jitter: 5s
       timeout: 1s
-    autoSyncInterval: 60s
+    autoSyncInterval: 160s
+    dialTimeout: 42s
   - zone: z2
     endpoints:
       - etcd3:2379
@@ -93,7 +94,7 @@ m3sd:
 	require.Equal(t, "/tmp/cache.json", cfg.CacheDir)
 	require.Equal(t, int64(1), cfg.WatchWithRevision)
 	require.Equal(t, []ClusterConfig{
-		ClusterConfig{
+		{
 			Zone:      "z1",
 			Endpoints: []string{"etcd1:2379", "etcd2:2379"},
 			KeepAlive: &KeepAliveConfig{
@@ -102,17 +103,18 @@ m3sd:
 				Jitter:  5 * time.Second,
 				Timeout: time.Second,
 			},
-			AutoSyncInterval: time.Second * 60,
+			AutoSyncInterval: 160 * time.Second,
+			DialTimeout:      42 * time.Second,
 		},
-		ClusterConfig{
+		{
 			Zone:      "z2",
 			Endpoints: []string{"etcd3:2379", "etcd4:2379"},
 			TLS: &TLSConfig{
 				CrtPath: "foo.crt.pem",
 				KeyPath: "foo.key.pem",
 			},
 		},
-		ClusterConfig{
+		{
 			Zone:      "z3",
 			Endpoints: []string{"etcd5:2379", "etcd6:2379"},
 			TLS: &TLSConfig{
@@ -132,7 +134,8 @@ m3sd:
 	require.Equal(t, 10*time.Second, keepAliveOpts.KeepAlivePeriod())
 	require.Equal(t, 5*time.Second, keepAliveOpts.KeepAlivePeriodMaxJitter())
 	require.Equal(t, time.Second, keepAliveOpts.KeepAliveTimeout())
-	require.Equal(t, 60*time.Second, cluster1.AutoSyncInterval())
+	require.Equal(t, 160*time.Second, cluster1.AutoSyncInterval())
+	require.Equal(t, 42*time.Second, cluster1.DialTimeout())
 
 	cluster2, exists := opts.ClusterForZone("z2")
 	require.True(t, exists)
@@ -172,3 +175,9 @@ m3sd:
 		require.Equal(t, os.FileMode(0744), *cfg2.NewDirectoryMode)
 	})
 }
+
+func TestDefaultConfig(t *testing.T) {
+	cluster := ClusterConfig{}.NewCluster()
+	require.Equal(t, defaultDialTimeout, cluster.DialTimeout())
+	require.Equal(t, defaultAutoSyncInterval, cluster.AutoSyncInterval())
+}

src/cmd/services/m3dbnode/config/config_test.go

@@ -638,6 +638,7 @@ func TestConfiguration(t *testing.T) {
             keepAlive: null
             tls: null
             autoSyncInterval: 0s
+            dialTimeout: 0s
           m3sd:
             initTimeout: null
           watchWithRevision: 0