Pass Semi sync information from durability policy and use it to log discrepancies

go/cmd/vtctl/vtctl.go

@@ -45,9 +45,9 @@ import (
 )
 
 var (
-	waitTime     = flag.Duration("wait-time", 24*time.Hour, "time to wait on an action")
-	detachedMode = flag.Bool("detach", false, "detached mode - run vtcl detached from the terminal")
-	durability   = flag.String("durability", "none", "type of durability to enforce. Default is none. Other values are dictated by registered plugins")
+	waitTime         = flag.Duration("wait-time", 24*time.Hour, "time to wait on an action")
+	detachedMode     = flag.Bool("detach", false, "detached mode - run vtcl detached from the terminal")
+	durabilityPolicy = flag.String("durability_policy", "none", "type of durability to enforce. Default is none. Other values are dictated by registered plugins")
 )
 
 func init() {
@@ -93,7 +93,7 @@ func main() {
 		log.Warningf("cannot connect to syslog: %v", err)
 	}
 
-	if err := reparentutil.SetDurabilityPolicy(*durability, nil); err != nil {
+	if err := reparentutil.SetDurabilityPolicy(*durabilityPolicy, nil); err != nil {
 		log.Errorf("error in setting durability policy: %v", err)
 		exit.Return(1)
 	}

go/cmd/vtworker/vtworker.go

@@ -37,13 +37,15 @@ import (
 	"vitess.io/vitess/go/vt/logutil"
 	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/topo"
+	"vitess.io/vitess/go/vt/vtctl/reparentutil"
 	"vitess.io/vitess/go/vt/worker"
 )
 
 var (
 	cell                   = flag.String("cell", "", "cell to pick servers from")
 	commandDisplayInterval = flag.Duration("command_display_interval", time.Second, "Interval between each status update when vtworker is executing a single command from the command line")
 	username               = flag.String("username", "", "If set, value is set as immediate caller id in the request and used by vttablet for TableACL check")
+	durabilityPolicy       = flag.String("durability_policy", "none", "type of durability to enforce. Default is none. Other values are dictated by registered plugins")
 )
 
 func init() {
@@ -78,6 +80,11 @@ func main() {
 		os.Exit(0)
 	}
 
+	if err := reparentutil.SetDurabilityPolicy(*durabilityPolicy, nil); err != nil {
+		log.Errorf("error in setting durability policy: %v", err)
+		exit.Return(1)
+	}
+
 	ts := topo.Open()
 	defer ts.Close()
 

go/test/endtoend/backup/transform/backup_transform_utils.go

@@ -68,6 +68,7 @@ func TestMainSetup(m *testing.M, useMysqlctld bool) {
 		localCluster = cluster.NewCluster(cell, hostname)
 		defer localCluster.Teardown()
 
+		localCluster.VtctldExtraArgs = append(localCluster.VtctldExtraArgs, "-durability_policy=semi_sync")
 		// Start topo server
 		err := localCluster.StartTopo()
 		if err != nil {

go/test/endtoend/backup/vtbackup/main_test.go

@@ -61,6 +61,7 @@ func TestMain(m *testing.M) {
 		localCluster = cluster.NewCluster(cell, hostname)
 		defer localCluster.Teardown()
 
+		localCluster.VtctldExtraArgs = append(localCluster.VtctldExtraArgs, "-durability_policy=semi_sync")
 		// Start topo server
 		err := localCluster.StartTopo()
 		if err != nil {

go/test/endtoend/backup/vtctlbackup/backup_utils.go

@@ -83,6 +83,7 @@ var (
 // LaunchCluster : starts the cluster as per given params.
 func LaunchCluster(setupType int, streamMode string, stripes int) (int, error) {
 	localCluster = cluster.NewCluster(cell, hostname)
+	localCluster.VtctldExtraArgs = append(localCluster.VtctldExtraArgs, "-durability_policy=semi_sync")
 
 	// Start topo server
 	err := localCluster.StartTopo()

go/test/endtoend/recovery/pitr/shardedpitr_test.go

@@ -371,6 +371,7 @@ func removeTablets(t *testing.T, tablets []*cluster.Vttablet) {
 
 func initializeCluster(t *testing.T) {
 	clusterInstance = cluster.NewCluster(cell, hostname)
+	clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "-durability_policy=semi_sync")
 
 	// Start topo server
 	err := clusterInstance.StartTopo()

go/test/endtoend/recovery/pitrtls/shardedpitr_tls_test.go

@@ -108,6 +108,7 @@ func removeTablets(t *testing.T, tablets []*cluster.Vttablet) {
 
 func initializeCluster(t *testing.T) {
 	clusterInstance = cluster.NewCluster(cell, hostname)
+	clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "-durability_policy=semi_sync")
 
 	// Start topo server
 	err := clusterInstance.StartTopo()

go/test/endtoend/recovery/shardedrecovery/sharded_recovery_test.go

@@ -466,6 +466,7 @@ func removeTablets(t *testing.T, tablets []*cluster.Vttablet) {
 func initializeCluster(t *testing.T) (int, error) {
 
 	localCluster = cluster.NewCluster(cell, hostname)
+	localCluster.VtctldExtraArgs = append(localCluster.VtctldExtraArgs, "-durability_policy=semi_sync")
 
 	// Start topo server
 	err := localCluster.StartTopo()

go/test/endtoend/recovery/unshardedrecovery/recovery.go

@@ -82,6 +82,7 @@ func TestMainImpl(m *testing.M) {
 		localCluster = cluster.NewCluster(cell, hostname)
 		defer localCluster.Teardown()
 
+		localCluster.VtctldExtraArgs = append(localCluster.VtctldExtraArgs, "-durability_policy=semi_sync")
 		// Start topo server
 		err := localCluster.StartTopo()
 		if err != nil {

go/test/endtoend/reparent/emergencyreparent/ers_test.go

@@ -30,7 +30,7 @@ import (
 
 func TestTrivialERS(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -55,7 +55,7 @@ func TestTrivialERS(t *testing.T) {
 
 func TestReparentIgnoreReplicas(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	var err error
@@ -97,7 +97,7 @@ func TestReparentIgnoreReplicas(t *testing.T) {
 
 func TestReparentDownPrimary(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -133,7 +133,7 @@ func TestReparentDownPrimary(t *testing.T) {
 
 func TestReparentNoChoiceDownPrimary(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	var err error
@@ -165,3 +165,61 @@ func TestReparentNoChoiceDownPrimary(t *testing.T) {
 	// bring back the old primary as a replica, check that it catches up
 	utils.ResurrectTablet(ctx, t, clusterInstance, tablets[0])
 }
+
+func TestSemiSyncSetupCorrectly(t *testing.T) {
+	t.Run("semi-sync enabled", func(t *testing.T) {
+		defer cluster.PanicHandler(t)
+		clusterInstance := utils.SetupReparentCluster(t, true)
+		defer utils.TeardownCluster(clusterInstance)
+		tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+
+		utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
+		// Run forced reparent operation, this should proceed unimpeded.
+		out, err := utils.Ers(clusterInstance, tablets[1], "60s", "30s")
+		require.NoError(t, err, out)
+
+		utils.ConfirmReplication(t, tablets[1], []*cluster.Vttablet{tablets[0], tablets[2], tablets[3]})
+
+		for _, tablet := range tablets {
+			utils.CheckSemiSyncSetupCorrectly(t, tablet, "ON")
+		}
+
+		// Run forced reparent operation, this should proceed unimpeded.
+		out, err = utils.Prs(t, clusterInstance, tablets[0])
+		require.NoError(t, err, out)
+
+		utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
+
+		for _, tablet := range tablets {
+			utils.CheckSemiSyncSetupCorrectly(t, tablet, "ON")
+		}
+	})
+
+	t.Run("semi-sync disabled", func(t *testing.T) {
+		defer cluster.PanicHandler(t)
+		clusterInstance := utils.SetupReparentCluster(t, false)
+		defer utils.TeardownCluster(clusterInstance)
+		tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+
+		utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
+		// Run forced reparent operation, this should proceed unimpeded.
+		out, err := utils.Ers(clusterInstance, tablets[1], "60s", "30s")
+		require.NoError(t, err, out)
+
+		utils.ConfirmReplication(t, tablets[1], []*cluster.Vttablet{tablets[0], tablets[2], tablets[3]})
+
+		for _, tablet := range tablets {
+			utils.CheckSemiSyncSetupCorrectly(t, tablet, "OFF")
+		}
+
+		// Run forced reparent operation, this should proceed unimpeded.
+		out, err = utils.Prs(t, clusterInstance, tablets[0])
+		require.NoError(t, err, out)
+
+		utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
+
+		for _, tablet := range tablets {
+			utils.CheckSemiSyncSetupCorrectly(t, tablet, "OFF")
+		}
+	})
+}

go/test/endtoend/reparent/newfeaturetest/reparent_test.go

@@ -37,6 +37,7 @@ func TestPRSForInitialization(t *testing.T) {
 	var tablets []*cluster.Vttablet
 	clusterInstance := cluster.NewCluster("zone1", "localhost")
 	keyspace := &cluster.Keyspace{Name: utils.KeyspaceName}
+	clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "-durability_policy=semi_sync")
 	// Start topo server
 	err := clusterInstance.StartTopo()
 	require.NoError(t, err)
@@ -107,7 +108,7 @@ func TestPRSForInitialization(t *testing.T) {
 // TestERSPromoteRdonly tests that we never end up promoting a rdonly instance as the primary
 func TestERSPromoteRdonly(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	var err error
@@ -135,7 +136,7 @@ func TestERSPromoteRdonly(t *testing.T) {
 // TestERSPreventCrossCellPromotion tests that we promote a replica in the same cell as the previous primary if prevent cross cell promotion flag is set
 func TestERSPreventCrossCellPromotion(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	var err error
@@ -158,7 +159,7 @@ func TestERSPreventCrossCellPromotion(t *testing.T) {
 // caught up to it by pulling transactions from it
 func TestPullFromRdonly(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	var err error
@@ -223,7 +224,7 @@ func TestPullFromRdonly(t *testing.T) {
 // is stopped on the primary elect.
 func TestNoReplicationStatusAndReplicationStopped(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	utils.ConfirmReplication(t, tablets[0], []*cluster.Vttablet{tablets[1], tablets[2], tablets[3]})
@@ -253,6 +254,7 @@ func TestERSForInitialization(t *testing.T) {
 	var tablets []*cluster.Vttablet
 	clusterInstance := cluster.NewCluster("zone1", "localhost")
 	keyspace := &cluster.Keyspace{Name: utils.KeyspaceName}
+	clusterInstance.VtctldExtraArgs = append(clusterInstance.VtctldExtraArgs, "-durability_policy=semi_sync")
 	// Start topo server
 	err := clusterInstance.StartTopo()
 	require.NoError(t, err)

go/test/endtoend/reparent/plannedreparent/reparent_test.go

@@ -33,7 +33,7 @@ import (
 
 func TestPrimaryToSpareStateChangeImpossible(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -45,7 +45,7 @@ func TestPrimaryToSpareStateChangeImpossible(t *testing.T) {
 
 func TestReparentCrossCell(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -59,7 +59,7 @@ func TestReparentCrossCell(t *testing.T) {
 
 func TestReparentGraceful(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -84,7 +84,7 @@ func TestReparentGraceful(t *testing.T) {
 // TestPRSWithDrainedLaggingTablet tests that PRS succeeds even if we have a lagging drained tablet
 func TestPRSWithDrainedLaggingTablet(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -111,7 +111,7 @@ func TestPRSWithDrainedLaggingTablet(t *testing.T) {
 
 func TestReparentReplicaOffline(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -128,7 +128,7 @@ func TestReparentReplicaOffline(t *testing.T) {
 
 func TestReparentAvoid(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 	utils.DeleteTablet(t, clusterInstance, tablets[2])
@@ -160,14 +160,14 @@ func TestReparentAvoid(t *testing.T) {
 
 func TestReparentFromOutside(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	reparentFromOutside(t, clusterInstance, false)
 }
 
 func TestReparentFromOutsideWithNoPrimary(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -256,7 +256,7 @@ func reparentFromOutside(t *testing.T, clusterInstance *cluster.LocalProcessClus
 
 func TestReparentWithDownReplica(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -299,7 +299,7 @@ func TestReparentWithDownReplica(t *testing.T) {
 
 func TestChangeTypeSemiSync(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 
@@ -363,7 +363,7 @@ func TestChangeTypeSemiSync(t *testing.T) {
 
 func TestReparentDoesntHangIfPrimaryFails(t *testing.T) {
 	defer cluster.PanicHandler(t)
-	clusterInstance := utils.SetupReparentCluster(t)
+	clusterInstance := utils.SetupReparentCluster(t, true)
 	defer utils.TeardownCluster(clusterInstance)
 	tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets