Have a common ERS for both VtOrc and Vtctl

go/cmd/vtcombo/main.go

@@ -217,7 +217,10 @@ func main() {
 	vtg := vtgate.Init(context.Background(), resilientServer, tpb.Cells[0], tabletTypesToWait)
 
 	// vtctld configuration and init
-	vtctld.InitVtctld(ts)
+	err = vtctld.InitVtctld(ts)
+	if err != nil {
+		exit.Return(1)
+	}
 
 	servenv.OnRun(func() {
 		addStatusParts(vtg)

go/cmd/vtctl/vtctl.go

@@ -38,6 +38,7 @@ import (
 	"vitess.io/vitess/go/vt/vtctl"
 	"vitess.io/vitess/go/vt/vtctl/grpcvtctldserver"
 	"vitess.io/vitess/go/vt/vtctl/localvtctldclient"
+	"vitess.io/vitess/go/vt/vtctl/reparentutil"
 	"vitess.io/vitess/go/vt/vttablet/tmclient"
 	"vitess.io/vitess/go/vt/workflow"
 	"vitess.io/vitess/go/vt/wrangler"
@@ -46,6 +47,7 @@ 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")
 )
 
 func init() {
@@ -91,6 +93,11 @@ func main() {
 		log.Warningf("cannot connect to syslog: %v", err)
 	}
 
+	if err := reparentutil.SetDurabilityPolicy(*durability, nil); err != nil {
+		log.Errorf("error in setting durability policy: %v", err)
+		exit.Return(1)
+	}
+
 	closer := trace.StartTracing("vtctl")
 	defer trace.LogErrorsWhenClosing(closer)
 

go/cmd/vtctld/main.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"vitess.io/vitess/go/exit"
 	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/vtctld"
@@ -40,7 +41,10 @@ func main() {
 	defer ts.Close()
 
 	// Init the vtctld core
-	vtctld.InitVtctld(ts)
+	err := vtctld.InitVtctld(ts)
+	if err != nil {
+		exit.Return(1)
+	}
 
 	// Register http debug/health
 	vtctld.RegisterDebugHealthHandler(ts)

go/cmd/vtctldclient/command/reparents.go

@@ -76,6 +76,7 @@ var emergencyReparentShardOptions = struct {
 	WaitReplicasTimeout       time.Duration
 	NewPrimaryAliasStr        string
 	IgnoreReplicaAliasStrList []string
+	PreventCrossCellPromotion bool
 }{}
 
 func commandEmergencyReparentShard(cmd *cobra.Command, args []string) error {
@@ -108,11 +109,12 @@ func commandEmergencyReparentShard(cmd *cobra.Command, args []string) error {
 	cli.FinishedParsing(cmd)
 
 	resp, err := client.EmergencyReparentShard(commandCtx, &vtctldatapb.EmergencyReparentShardRequest{
-		Keyspace:            keyspace,
-		Shard:               shard,
-		NewPrimary:          newPrimaryAlias,
-		IgnoreReplicas:      ignoreReplicaAliases,
-		WaitReplicasTimeout: protoutil.DurationToProto(emergencyReparentShardOptions.WaitReplicasTimeout),
+		Keyspace:                  keyspace,
+		Shard:                     shard,
+		NewPrimary:                newPrimaryAlias,
+		IgnoreReplicas:            ignoreReplicaAliases,
+		WaitReplicasTimeout:       protoutil.DurationToProto(emergencyReparentShardOptions.WaitReplicasTimeout),
+		PreventCrossCellPromotion: emergencyReparentShardOptions.PreventCrossCellPromotion,
 	})
 	if err != nil {
 		return err
@@ -261,6 +263,7 @@ func commandTabletExternallyReparented(cmd *cobra.Command, args []string) error
 func init() {
 	EmergencyReparentShard.Flags().DurationVar(&emergencyReparentShardOptions.WaitReplicasTimeout, "wait-replicas-timeout", *topo.RemoteOperationTimeout, "Time to wait for replicas to catch up in reparenting.")
 	EmergencyReparentShard.Flags().StringVar(&emergencyReparentShardOptions.NewPrimaryAliasStr, "new-primary", "", "Alias of a tablet that should be the new primary. If not specified, the vtctld will select the best candidate to promote.")
+	EmergencyReparentShard.Flags().BoolVar(&emergencyReparentShardOptions.PreventCrossCellPromotion, "prevent-cross-cell-promotion", false, "Only promotes a new primary from the same cell as the previous primary")
 	EmergencyReparentShard.Flags().StringSliceVarP(&emergencyReparentShardOptions.IgnoreReplicaAliasStrList, "ignore-replicas", "i", nil, "Comma-separated, repeated list of replica tablet aliases to ignore during the emergency reparent.")
 	Root.AddCommand(EmergencyReparentShard)
 

go/mysql/replication_status.go

@@ -123,16 +123,7 @@ func (s *ReplicationStatus) FindErrantGTIDs(otherReplicaStatuses []*ReplicationS
 		if !ok {
 			panic("The receiver ReplicationStatus contained a Mysql56GTIDSet in its relay log, but a replica's ReplicationStatus is of another flavor. This should never happen.")
 		}
-		// Copy and throw out primary SID from consideration, so we don't mutate input.
-		otherSetNoPrimarySID := make(Mysql56GTIDSet, len(otherSet))
-		for sid, intervals := range otherSet {
-			if sid == status.SourceUUID {
-				continue
-			}
-			otherSetNoPrimarySID[sid] = intervals
-		}
-
-		otherSets = append(otherSets, otherSetNoPrimarySID)
+		otherSets = append(otherSets, otherSet)
 	}
 
 	// Copy set for final diffSet so we don't mutate receiver.

go/mysql/replication_status_test.go

@@ -18,6 +18,8 @@ package mysql
 
 import (
 	"testing"
+
+	"github.com/stretchr/testify/require"
 )
 
 func TestStatusReplicationRunning(t *testing.T) {
@@ -81,22 +83,33 @@ func TestFindErrantGTIDs(t *testing.T) {
 		sourceSID: []interval{{2, 6}, {15, 45}},
 	}
 
-	status1 := ReplicationStatus{SourceUUID: sourceSID, RelayLogPosition: Position{GTIDSet: set1}}
-	status2 := ReplicationStatus{SourceUUID: sourceSID, RelayLogPosition: Position{GTIDSet: set2}}
-	status3 := ReplicationStatus{SourceUUID: sourceSID, RelayLogPosition: Position{GTIDSet: set3}}
-
-	got, err := status1.FindErrantGTIDs([]*ReplicationStatus{&status2, &status3})
-	if err != nil {
-		t.Errorf("%v", err)
-	}
-
-	want := Mysql56GTIDSet{
-		sid1: []interval{{39, 39}, {40, 49}, {71, 75}},
-		sid2: []interval{{1, 2}, {6, 7}, {20, 21}, {26, 31}, {38, 50}, {60, 66}},
-		sid4: []interval{{1, 30}},
-	}
-
-	if !got.Equal(want) {
-		t.Errorf("got %#v; want %#v", got, want)
+	testcases := []struct {
+		mainRepStatus    *ReplicationStatus
+		otherRepStatuses []*ReplicationStatus
+		want             Mysql56GTIDSet
+	}{{
+		mainRepStatus: &ReplicationStatus{SourceUUID: sourceSID, RelayLogPosition: Position{GTIDSet: set1}},
+		otherRepStatuses: []*ReplicationStatus{
+			{SourceUUID: sourceSID, RelayLogPosition: Position{GTIDSet: set2}},
+			{SourceUUID: sourceSID, RelayLogPosition: Position{GTIDSet: set3}},
+		},
+		want: Mysql56GTIDSet{
+			sid1: []interval{{39, 39}, {40, 49}, {71, 75}},
+			sid2: []interval{{1, 2}, {6, 7}, {20, 21}, {26, 31}, {38, 50}, {60, 66}},
+			sid4: []interval{{1, 30}},
+		},
+	}, {
+		mainRepStatus:    &ReplicationStatus{SourceUUID: sourceSID, RelayLogPosition: Position{GTIDSet: set1}},
+		otherRepStatuses: []*ReplicationStatus{{SourceUUID: sid1, RelayLogPosition: Position{GTIDSet: set1}}},
+		// servers with the same GTID sets should not be diagnosed with errant GTIDs
+		want: nil,
+	}}
+
+	for _, testcase := range testcases {
+		t.Run("", func(t *testing.T) {
+			got, err := testcase.mainRepStatus.FindErrantGTIDs(testcase.otherRepStatuses)
+			require.NoError(t, err)
+			require.Equal(t, testcase.want, got)
+		})
 	}
 }

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

@@ -0,0 +1,204 @@
+/*
+Copyright 2019 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package emergencyreparent
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/test/endtoend/cluster"
+	"vitess.io/vitess/go/vt/log"
+)
+
+func TestTrivialERS(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	setupReparentCluster(t)
+	defer teardownCluster()
+
+	confirmReplication(t, tab1, []*cluster.Vttablet{tab2, tab3, tab4})
+
+	// We should be able to do a series of ERS-es, even if nothing
+	// is down, without issue
+	for i := 1; i <= 4; i++ {
+		out, err := ers(nil, "60s", "30s")
+		log.Infof("ERS loop %d.  EmergencyReparentShard Output: %v", i, out)
+		require.NoError(t, err)
+		time.Sleep(5 * time.Second)
+	}
+	// We should do the same for vtctl binary
+	for i := 1; i <= 4; i++ {
+		out, err := ersWithVtctl()
+		log.Infof("ERS-vtctl loop %d.  EmergencyReparentShard Output: %v", i, out)
+		require.NoError(t, err)
+		time.Sleep(5 * time.Second)
+	}
+}
+
+func TestReparentIgnoreReplicas(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	setupReparentCluster(t)
+	defer teardownCluster()
+	var err error
+
+	ctx := context.Background()
+
+	confirmReplication(t, tab1, []*cluster.Vttablet{tab2, tab3, tab4})
+
+	// Make the current primary agent and database unavailable.
+	stopTablet(t, tab1, true)
+
+	// Take down a replica - this should cause the emergency reparent to fail.
+	stopTablet(t, tab3, true)
+
+	// We expect this one to fail because we have an unreachable replica
+	out, err := ers(nil, "60s", "30s")
+	require.NotNil(t, err, out)
+
+	// Now let's run it again, but set the command to ignore the unreachable replica.
+	out, err = ersIgnoreTablet(nil, "60s", "30s", []*cluster.Vttablet{tab3}, false)
+	require.Nil(t, err, out)
+
+	// We'll bring back the replica we took down.
+	restartTablet(t, tab3)
+
+	// Check that old primary tablet is left around for human intervention.
+	confirmOldPrimaryIsHangingAround(t)
+	deleteTablet(t, tab1)
+	validateTopology(t, false)
+
+	newPrimary := getNewPrimary(t)
+	// Check new primary has latest transaction.
+	err = checkInsertedValues(ctx, t, newPrimary, insertVal)
+	require.Nil(t, err)
+
+	// bring back the old primary as a replica, check that it catches up
+	resurrectTablet(ctx, t, tab1)
+}
+
+// TestERSPromoteRdonly tests that we never end up promoting a rdonly instance as the primary
+func TestERSPromoteRdonly(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	setupReparentCluster(t)
+	defer teardownCluster()
+	var err error
+
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", tab2.Alias, "rdonly")
+	require.NoError(t, err)
+
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", tab3.Alias, "rdonly")
+	require.NoError(t, err)
+
+	confirmReplication(t, tab1, []*cluster.Vttablet{tab2, tab3, tab4})
+
+	// Make the current primary agent and database unavailable.
+	stopTablet(t, tab1, true)
+
+	// We expect this one to fail because we have ignored all the replicas and have only the rdonly's which should not be promoted
+	out, err := ersIgnoreTablet(nil, "30s", "30s", []*cluster.Vttablet{tab4}, false)
+	require.NotNil(t, err, out)
+
+	out, err = clusterInstance.VtctlclientProcess.ExecuteCommandWithOutput("GetShard", keyspaceShard)
+	require.NoError(t, err)
+	require.Contains(t, out, `"uid": 101`, "the primary should still be 101 in the shard info")
+}
+
+// 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)
+	setupReparentCluster(t)
+	defer teardownCluster()
+	var err error
+
+	// confirm that replication is going smoothly
+	confirmReplication(t, tab1, []*cluster.Vttablet{tab2, tab3, tab4})
+
+	// Make the current primary agent and database unavailable.
+	stopTablet(t, tab1, true)
+
+	// We expect that tab3 will be promoted since it is in the same cell as the previous primary
+	out, err := ersIgnoreTablet(nil, "60s", "30s", []*cluster.Vttablet{tab2}, true)
+	require.NoError(t, err, out)
+
+	newPrimary := getNewPrimary(t)
+	require.Equal(t, newPrimary.Alias, tab3.Alias, "tab3 should be the promoted primary")
+}
+
+// TestPullFromRdonly tests that if a rdonly tablet is the most advanced, then our promoted primary should have
+// caught up to it by pulling transactions from it
+func TestPullFromRdonly(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	setupReparentCluster(t)
+	defer teardownCluster()
+	var err error
+
+	ctx := context.Background()
+	// make tab2 a rdonly tablet.
+	// rename tablet so that the test is not confusing
+	rdonly := tab2
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", rdonly.Alias, "rdonly")
+	require.NoError(t, err)
+
+	// confirm that all the tablets can replicate successfully right now
+	confirmReplication(t, tab1, []*cluster.Vttablet{rdonly, tab3, tab4})
+
+	// stop replication on the other two tablets
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StopReplication", tab3.Alias)
+	require.NoError(t, err)
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StopReplication", tab4.Alias)
+	require.NoError(t, err)
+
+	// stop semi-sync on the primary so that any transaction now added does not require an ack
+	runSQL(ctx, t, "SET GLOBAL rpl_semi_sync_master_enabled = false", tab1)
+
+	// confirm that rdonly is able to replicate from our primary
+	// This will also introduce a new transaction into the rdonly tablet which the other 2 replicas don't have
+	confirmReplication(t, tab1, []*cluster.Vttablet{rdonly})
+
+	// Make the current primary agent and database unavailable.
+	stopTablet(t, tab1, true)
+
+	// start the replication back on the two tablets
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StartReplication", tab3.Alias)
+	require.NoError(t, err)
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StartReplication", tab4.Alias)
+	require.NoError(t, err)
+
+	// check that tab3 and tab4 still only has 1 value
+	err = checkCountOfInsertedValues(ctx, t, tab3, 1)
+	require.NoError(t, err)
+	err = checkCountOfInsertedValues(ctx, t, tab4, 1)
+	require.NoError(t, err)
+
+	// At this point we have successfully made our rdonly tablet more advanced than tab3 and tab4 without introducing errant GTIDs
+	// We have simulated a network partition in which the primary and rdonly got isolated and then the primary went down leaving the rdonly most advanced
+
+	// We expect that tab3 will be promoted since it is in the same cell as the previous primary
+	// since we are preventing cross cell promotions
+	// Also it must be fully caught up
+	out, err := ersIgnoreTablet(nil, "60s", "30s", nil, true)
+	require.NoError(t, err, out)
+
+	newPrimary := getNewPrimary(t)
+	require.Equal(t, newPrimary.Alias, tab3.Alias, "tab3 should be the promoted primary")
+
+	// check that the new primary has the last transaction that only the rdonly had
+	err = checkInsertedValues(ctx, t, newPrimary, insertVal)
+	require.NoError(t, err)
+}