[dbnode/client] Count Write Acks from paired (leaving, initializing) …

…shards (#4191)

* writing into leaving and init and count the success as pair

site/content/includes/m3dbnode/annotated_config.yaml

@@ -362,7 +362,14 @@ db:
         writeShardsInitializing: <bool>
         # Whether or not writes to leaving shards count towards consistency
         # Default = false
+        # NOTE: shardsLeavingCountTowardsConsistency and shardsLeavingAndInitializingCountTowardsConsistency both
+        # cannot be true
         shardsLeavingCountTowardsConsistency: <bool>
+        # Whether or not writes to both leaving and initializing shards as pair counts toward consistency
+        # Default = false
+        # NOTE: shardsLeavingCountTowardsConsistency and shardsLeavingAndInitializingCountTowardsConsistency both
+        # cannot be true
+        shardsLeavingAndInitializingCountTowardsConsistency: <bool>
   # Specifies the pooling policy
   pooling:
     # Initial alloc size for a block

src/cluster/placement/algo/sharded_helper.go

@@ -827,7 +827,6 @@ func markShardsAvailable(p placement.Placement, instanceID string, shardIDs []ui
 				return nil, err
 			}
 		}
-
 		sourceShards.Remove(shardID)
 		if sourceShards.NumShards() == 0 {
 			p = p.SetInstances(removeInstanceFromList(p.Instances(), sourceInstance.ID()))

src/cluster/shard/shard.go

@@ -108,7 +108,6 @@ func (s *shard) State() State                      { return s.state }
 func (s *shard) SetState(state State) Shard        { s.state = state; return s }
 func (s *shard) SourceID() string                  { return s.sourceID }
 func (s *shard) SetSourceID(sourceID string) Shard { s.sourceID = sourceID; return s }
-
 func (s *shard) CutoverNanos() int64 {
 	if s.cutoverNanos != UnInitializedValue {
 		return s.cutoverNanos

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

@@ -414,6 +414,7 @@ func TestConfiguration(t *testing.T) {
     fetchSeriesBlocksBatchSize: null
     writeShardsInitializing: null
     shardsLeavingCountTowardsConsistency: null
+    shardsLeavingAndInitializingCountTowardsConsistency: null
     iterateEqualTimestampStrategy: null
   gcPercentage: 100
   tick: null

src/cmd/tools/dtest/util/seed/generator.go

@@ -156,6 +156,10 @@ type fakeShardSet struct {
 	shardID uint32
 }
 
+func (f *fakeShardSet) LookupShard(id uint32) (shard.Shard, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
 func (f *fakeShardSet) All() []shard.Shard {
 	sh := shard.NewShard(f.shardID)
 	return []shard.Shard{sh}

src/dbnode/client/config.go

@@ -124,6 +124,10 @@ type Configuration struct {
 	// count towards consistency, by default they do not.
 	ShardsLeavingCountTowardsConsistency *bool `yaml:"shardsLeavingCountTowardsConsistency"`
 
+	// ShardsLeavingAndInitializingCountTowardsConsistency sets whether or not writes to leaving and initializing shards
+	// count towards consistency, by default they do not.
+	ShardsLeavingAndInitializingCountTowardsConsistency *bool `yaml:"shardsLeavingAndInitializingCountTowardsConsistency"`
+
 	// IterateEqualTimestampStrategy specifies the iterate equal timestamp strategy.
 	IterateEqualTimestampStrategy *encoding.IterateEqualTimestampStrategy `yaml:"iterateEqualTimestampStrategy"`
 }
@@ -213,6 +217,12 @@ func (c *Configuration) Validate() error {
 			*c.AsyncWriteMaxConcurrency)
 	}
 
+	if c.ShardsLeavingCountTowardsConsistency != nil && c.ShardsLeavingAndInitializingCountTowardsConsistency != nil &&
+		*c.ShardsLeavingCountTowardsConsistency && *c.ShardsLeavingAndInitializingCountTowardsConsistency {
+		return fmt.Errorf("m3db client cannot have both shardsLeavingCountTowardsConsistency and " +
+			"shardsLeavingAndInitializingCountTowardsConsistency as true")
+	}
+
 	if err := c.Proto.Validate(); err != nil {
 		return fmt.Errorf("error validating M3DB client proto configuration: %v", err)
 	}
@@ -451,6 +461,10 @@ func (c Configuration) NewAdminClient(
 	if c.WriteShardsInitializing != nil {
 		v = v.SetWriteShardsInitializing(*c.WriteShardsInitializing)
 	}
+	if c.ShardsLeavingAndInitializingCountTowardsConsistency != nil {
+		v = v.SetShardsLeavingAndInitializingCountTowardsConsistency(*c.ShardsLeavingAndInitializingCountTowardsConsistency)
+	}
+
 	if c.ShardsLeavingCountTowardsConsistency != nil {
 		v = v.SetShardsLeavingCountTowardsConsistency(*c.ShardsLeavingCountTowardsConsistency)
 	}

src/dbnode/client/config_test.go

@@ -21,6 +21,7 @@
 package client
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
 	"testing"
@@ -132,3 +133,18 @@ proto:
 
 	assert.Equal(t, expected, cfg)
 }
+
+func TestValidateConfig(t *testing.T) {
+	var (
+		boolTrue = true
+	)
+
+	config := Configuration{
+		ShardsLeavingCountTowardsConsistency:                &boolTrue,
+		ShardsLeavingAndInitializingCountTowardsConsistency: &boolTrue,
+	}
+	err := config.Validate()
+	require.Error(t, err)
+	require.Equal(t, err, fmt.Errorf("m3db client cannot have both shardsLeavingCountTowardsConsistency and "+
+		"shardsLeavingAndInitializingCountTowardsConsistency as true"))
+}