add support for vtgate traffic mirroring (#15945)

Signed-off-by: Max Englander <[email protected]>
Signed-off-by: Max Englander <[email protected]>
Co-authored-by: Deepthi Sigireddi <[email protected]>

changelog/21.0/21.0.0/summary.md

@@ -8,6 +8,7 @@
     - [Deletion of deprecated metrics](#metric-deletion)
     - [VTTablet Flags](#vttablet-flags)
   - **[Breaking changes](#breaking-changes)**
+  - **[Traffic Mirroring](#traffic-mirroring)**
 
 ## <a id="major-changes"/>Major Changes
 
@@ -38,3 +39,14 @@ The following metrics that were deprecated in the previous release, have now bee
 - `queryserver-enable-settings-pool` flag, added in v15, has been on by default since v17. 
 It is now deprecated and will be removed in a future release.
 
+### <a id="traffic-mirroring"/>Traffic Mirroring
+
+Traffic mirroring is intended to help reduce some of the uncertainty inherent to `MoveTables SwitchTraffic`. When traffic mirroring is enabled, VTGate will mirror a percentage of traffic from one keyspace to another.
+
+Mirror rules may be enabled through `vtctldclient` with `MoveTables MirrorTraffic`. For example:
+
+```bash
+$ vtctldclient --server :15999 MoveTables --target-keyspace customer --workflow commerce2customer MirrorTraffic --percent 5.0
+```
+
+Mirror rules can be inspected with `GetMirrorRules`.

go/cmd/vtcombo/cli/main.go

@@ -206,6 +206,11 @@ func run(cmd *cobra.Command, args []string) (err error) {
 		return fmt.Errorf("Failed to load routing rules: %w", err)
 	}
 
+	// attempt to load any mirror rules specified by tpb
+	if err := vtcombo.InitMirrorRules(context.Background(), ts, tpb.GetMirrorRules()); err != nil {
+		return fmt.Errorf("Failed to load mirror rules: %w", err)
+	}
+
 	servenv.Init()
 	tabletenv.Init()
 

go/cmd/vtctldclient/command/mirror_rules.go

@@ -0,0 +1,58 @@
+/*
+Copyright 2024 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 command
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	"vitess.io/vitess/go/cmd/vtctldclient/cli"
+
+	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
+)
+
+// GetMirrorRules makes a GetMirrorRules gRPC call to a vtctld.
+var GetMirrorRules = &cobra.Command{
+	Use:                   "GetMirrorRules",
+	Short:                 "Displays the VSchema mirror rules.",
+	DisableFlagsInUseLine: true,
+	Args:                  cobra.NoArgs,
+	RunE:                  commandGetMirrorRules,
+}
+
+func commandGetMirrorRules(cmd *cobra.Command, args []string) error {
+	cli.FinishedParsing(cmd)
+
+	resp, err := client.GetMirrorRules(commandCtx, &vtctldatapb.GetMirrorRulesRequest{})
+	if err != nil {
+		return err
+	}
+
+	data, err := cli.MarshalJSON(resp.MirrorRules)
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("%s\n", data)
+
+	return nil
+}
+
+func init() {
+	Root.AddCommand(GetMirrorRules)
+}

go/cmd/vtctldclient/command/vreplication/common/mirrortraffic.go

@@ -0,0 +1,89 @@
+/*
+Copyright 2024 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 common
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	"vitess.io/vitess/go/cmd/vtctldclient/cli"
+
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
+)
+
+func GetMirrorTrafficCommand(opts *SubCommandsOpts) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:                   "mirrortraffic",
+		Short:                 fmt.Sprintf("Mirror traffic for a %s MoveTables workflow.", opts.SubCommand),
+		Example:               fmt.Sprintf(`vtctldclient --server localhost:15999 %s --workflow %s --target-keyspace customer mirrortraffic --percent 5.0`, opts.SubCommand, opts.Workflow),
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"MirrorTraffic"},
+		Args:                  cobra.NoArgs,
+		PreRun: func(cmd *cobra.Command, args []string) {
+			if !cmd.Flags().Lookup("tablet-types").Changed {
+				// We mirror traffic for all tablet types if none are provided.
+				MirrorTrafficOptions.TabletTypes = []topodatapb.TabletType{
+					topodatapb.TabletType_PRIMARY,
+					topodatapb.TabletType_REPLICA,
+					topodatapb.TabletType_RDONLY,
+				}
+			}
+		},
+		RunE: commandMirrorTraffic,
+	}
+	return cmd
+}
+
+func commandMirrorTraffic(cmd *cobra.Command, args []string) error {
+	format, err := GetOutputFormat(cmd)
+	if err != nil {
+		return err
+	}
+
+	cli.FinishedParsing(cmd)
+
+	req := &vtctldatapb.WorkflowMirrorTrafficRequest{
+		Keyspace:    BaseOptions.TargetKeyspace,
+		Workflow:    BaseOptions.Workflow,
+		TabletTypes: MirrorTrafficOptions.TabletTypes,
+		Percent:     MirrorTrafficOptions.Percent,
+	}
+	resp, err := GetClient().WorkflowMirrorTraffic(GetCommandCtx(), req)
+	if err != nil {
+		return err
+	}
+
+	var output []byte
+	if format == "json" {
+		output, err = cli.MarshalJSONPretty(resp)
+		if err != nil {
+			return err
+		}
+	} else {
+		tout := bytes.Buffer{}
+		tout.WriteString(resp.Summary + "\n\n")
+		tout.WriteString(fmt.Sprintf("Start State: %s\n", resp.StartState))
+		tout.WriteString(fmt.Sprintf("Current State: %s\n", resp.CurrentState))
+		output = tout.Bytes()
+	}
+	fmt.Printf("%s\n", output)
+
+	return nil
+}

go/cmd/vtctldclient/command/vreplication/common/utils.go

@@ -224,6 +224,12 @@ func AddCommonCreateFlags(cmd *cobra.Command) {
 	cmd.Flags().BoolVar(&CreateOptions.StopAfterCopy, "stop-after-copy", false, "Stop the workflow after it's finished copying the existing rows and before it starts replicating changes.")
 }
 
+var MirrorTrafficOptions = struct {
+	DryRun      bool
+	Percent     float32
+	TabletTypes []topodatapb.TabletType
+}{}
+
 var SwitchTrafficOptions = struct {
 	Cells                     []string
 	TabletTypes               []topodatapb.TabletType

go/cmd/vtctldclient/command/vreplication/movetables/movetables.go

@@ -20,6 +20,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"vitess.io/vitess/go/cmd/vtctldclient/command/vreplication/common"
+	"vitess.io/vitess/go/vt/topo/topoproto"
 )
 
 var (
@@ -67,6 +68,11 @@ func registerCommands(root *cobra.Command) {
 	base.AddCommand(common.GetStartCommand(opts))
 	base.AddCommand(common.GetStopCommand(opts))
 
+	mirrorTrafficCommand := common.GetMirrorTrafficCommand(opts)
+	mirrorTrafficCommand.Flags().Var((*topoproto.TabletTypeListFlag)(&common.MirrorTrafficOptions.TabletTypes), "tablet-types", "Tablet types to mirror traffic for.")
+	mirrorTrafficCommand.Flags().Float32Var(&common.MirrorTrafficOptions.Percent, "percent", 1.0, "Percentage of traffic to mirror.")
+	base.AddCommand(mirrorTrafficCommand)
+
 	switchTrafficCommand := common.GetSwitchTrafficCommand(opts)
 	common.AddCommonSwitchTrafficFlags(switchTrafficCommand, true)
 	common.AddShardSubsetFlag(switchTrafficCommand, &common.SwitchTrafficOptions.Shards)

go/flags/endtoend/vtctldclient.txt

@@ -42,6 +42,7 @@ Available Commands:
   GetKeyspace                 Returns information about the given keyspace from the topology.
   GetKeyspaceRoutingRules     Displays the currently active keyspace routing rules.
   GetKeyspaces                Returns information about every keyspace in the topology.
+  GetMirrorRules              Displays the VSchema mirror rules.
   GetPermissions              Displays the permissions for a tablet.
   GetRoutingRules             Displays the VSchema routing rules.
   GetSchema                   Displays the full schema for a tablet, optionally restricted to the specified tables/views.

go/test/endtoend/vreplication/movetables_mirrortraffic_test.go

@@ -0,0 +1,114 @@
+/*
+Copyright 2024 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 vreplication
+
+import (
+	"testing"
+
+	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+)
+
+func testMoveTablesMirrorTraffic(t *testing.T, flavor workflowFlavor) {
+	setSidecarDBName("_vt")
+	vc = setupMinimalCluster(t)
+	defer vc.TearDown()
+
+	sourceKeyspace := "product"
+	targetKeyspace := "customer"
+	workflowName := "wf1"
+	tables := []string{"customer", "loadtest", "customer2"}
+
+	_ = setupMinimalCustomerKeyspace(t)
+
+	mtwf := &moveTablesWorkflow{
+		workflowInfo: &workflowInfo{
+			vc:             vc,
+			workflowName:   workflowName,
+			targetKeyspace: targetKeyspace,
+		},
+		sourceKeyspace: sourceKeyspace,
+		tables:         "customer,loadtest,customer2",
+		mirrorFlags:    []string{"--percent", "25"},
+	}
+	mt := newMoveTables(vc, mtwf, flavor)
+
+	// Mirror rules do not exist by default.
+	mt.Create()
+	confirmNoMirrorRules(t)
+
+	waitForWorkflowState(t, vc, ksWorkflow, binlogdatapb.VReplicationWorkflowState_Running.String())
+
+	// Mirror rules can be created after a MoveTables workflow is created.
+	mt.MirrorTraffic()
+	confirmMirrorRulesExist(t)
+	expectMirrorRules(t, sourceKeyspace, targetKeyspace, tables, []topodatapb.TabletType{
+		topodatapb.TabletType_PRIMARY,
+		topodatapb.TabletType_REPLICA,
+		topodatapb.TabletType_RDONLY,
+	}, 25)
+
+	// Mirror rules can be adjusted after mirror rules are in place.
+	mtwf.mirrorFlags[1] = "50"
+	mt.MirrorTraffic()
+	confirmMirrorRulesExist(t)
+	expectMirrorRules(t, sourceKeyspace, targetKeyspace, tables, []topodatapb.TabletType{
+		topodatapb.TabletType_PRIMARY,
+		topodatapb.TabletType_REPLICA,
+		topodatapb.TabletType_RDONLY,
+	}, 50)
+
+	// Mirror rules can be adjusted multiple times after mirror rules are in
+	// place.
+	mtwf.mirrorFlags[1] = "75"
+	mt.MirrorTraffic()
+	confirmMirrorRulesExist(t)
+	expectMirrorRules(t, sourceKeyspace, targetKeyspace, tables, []topodatapb.TabletType{
+		topodatapb.TabletType_PRIMARY,
+		topodatapb.TabletType_REPLICA,
+		topodatapb.TabletType_RDONLY,
+	}, 75)
+
+	lg := newLoadGenerator(t, vc)
+	go func() {
+		lg.start()
+	}()
+	lg.waitForCount(1000)
+
+	mt.SwitchReads()
+	confirmMirrorRulesExist(t)
+
+	// Mirror rules can be adjusted for writes after reads have been switched.
+	mtwf.mirrorFlags[1] = "100"
+	mtwf.mirrorFlags = append(mtwf.mirrorFlags, "--tablet-types", "primary")
+	mt.MirrorTraffic()
+	confirmMirrorRulesExist(t)
+	expectMirrorRules(t, sourceKeyspace, targetKeyspace, tables, []topodatapb.TabletType{
+		topodatapb.TabletType_PRIMARY,
+	}, 100)
+
+	// Mirror rules are removed after writes are switched.
+	mt.SwitchWrites()
+	confirmNoMirrorRules(t)
+}
+
+func TestMoveTablesMirrorTraffic(t *testing.T) {
+	currentWorkflowType = binlogdatapb.VReplicationWorkflowType_MoveTables
+	t.Run(workflowFlavorNames[workflowFlavorVtctld], func(t *testing.T) {
+		testMoveTablesMirrorTraffic(t, workflowFlavorVtctld)
+	})
+}

go/test/endtoend/vreplication/resharding_workflows_v2_test.go

@@ -52,6 +52,7 @@ const (
 
 const (
 	workflowActionCreate         = "Create"
+	workflowActionMirrorTraffic  = "Mirror"
 	workflowActionSwitchTraffic  = "SwitchTraffic"
 	workflowActionReverseTraffic = "ReverseTraffic"
 	workflowActionComplete       = "Complete"
@@ -70,6 +71,7 @@ type workflowExecOptions struct {
 	deferSecondaryKeys bool
 	atomicCopy         bool
 	shardSubset        string
+	percent            float32
 }
 
 var defaultWorkflowExecOptions = &workflowExecOptions{
@@ -222,6 +224,8 @@ func tstWorkflowExecVtctl(t *testing.T, cells, workflow, sourceKs, targetKs, tab
 			}
 			args = append(args, "--initialize-target-sequences") // Only used for MoveTables
 		}
+	case workflowActionMirrorTraffic:
+		args = append(args, "--percent", strconv.FormatFloat(float64(options.percent), byte('f'), -1, 32))
 	default:
 		if options.shardSubset != "" {
 			args = append(args, "--shards", options.shardSubset)

go/test/endtoend/vreplication/vreplication_test_env.go

@@ -17,6 +17,7 @@ limitations under the License.
 package vreplication
 
 var dryRunResultsSwitchWritesCustomerShard = []string{
+	"Mirroring 0.00 percent of traffic from keyspace product to keyspace customer for tablet types [PRIMARY]",
 	"Lock keyspace product",
 	"Lock keyspace customer",
 	"/Stop writes on keyspace product for tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order]: [keyspace:product;shard:0;position:",
@@ -35,6 +36,7 @@ var dryRunResultsSwitchWritesCustomerShard = []string{
 }
 
 var dryRunResultsReadCustomerShard = []string{
+	"Mirroring 0.00 percent of traffic from keyspace product to keyspace customer for tablet types [RDONLY,REPLICA]",
 	"Lock keyspace product",
 	"Switch reads for tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order] to keyspace customer for tablet types [RDONLY,REPLICA]",
 	"Routing rules for tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order] will be updated",