Merge remote-tracking branch 'upstream/main' into cluster-watch-range

Signed-off-by: Manan Gupta <[email protected]>

.github/workflows/check_make_vtadmin_web_proto.yml

@@ -61,7 +61,7 @@ jobs:
       uses: actions/setup-node@1e60f620b9541d16bece96c5465dc8ee9832be0b # v4.0.3
       with:
         # node-version should match package.json
-        node-version: '20.12.2'
+        node-version: '22.13.1'
 
     - name: Install npm dependencies
       if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.proto_changes == 'true'

.github/workflows/static_checks_etc.yml

@@ -230,7 +230,7 @@ jobs:
         uses: actions/setup-node@1e60f620b9541d16bece96c5465dc8ee9832be0b # v4.0.3
         with:
           # make proto requires newer node than the pre-installed one
-          node-version: '20.12.2'
+          node-version: '22.13.1'
 
       - name: check_make_proto
         if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.proto_changes == 'true'

.github/workflows/vtadmin_web_build.yml

@@ -44,7 +44,7 @@ jobs:
         if: steps.skip-workflow.outputs.skip-workflow == 'false'
         with:
           # node-version should match package.json
-          node-version: '20.12.2'
+          node-version: '22.13.1'
 
       - name: Install dependencies
         if: steps.skip-workflow.outputs.skip-workflow == 'false'

.github/workflows/vtadmin_web_lint.yml

@@ -44,7 +44,7 @@ jobs:
         if: steps.skip-workflow.outputs.skip-workflow == 'false'
         with:
           # node-version should match package.json
-          node-version: '20.12.2'
+          node-version: '22.13.1'
 
       - name: Install dependencies
         if: steps.skip-workflow.outputs.skip-workflow == 'false'

.github/workflows/vtadmin_web_unit_tests.yml

@@ -44,7 +44,7 @@ jobs:
         if: steps.skip-workflow.outputs.skip-workflow == 'false'
         with:
           # node-version should match package.json
-          node-version: '20.12.2'
+          node-version: '22.13.1'
 
       - name: Install dependencies
         if: steps.skip-workflow.outputs.skip-workflow == 'false'

changelog/22.0/22.0.0/summary.md

@@ -20,6 +20,8 @@
 - **[Minor Changes](#minor-changes)**
   - **[VTTablet Flags](#flags-vttablet)**
   - **[Topology read concurrency behaviour changes](#topo-read-concurrency-changes)**
+  - **[VTAdmin](#vtadmin)**
+    - [Updated to node v22.13.1](#updated-node)
 
 ## <a id="major-changes"/>Major Changes</a>
 
@@ -156,3 +158,10 @@ While the flag will continue to accept float values (interpreted as seconds) for
 The `--topo_read_concurrency` flag was added to all components that access the topology and the provided limit is now applied separately for each global or local cell _(default `32`)_.
 
 All topology read calls _(`Get`, `GetVersion`, `List` and `ListDir`)_ now respect this per-cell limit. Previous to this version a single limit was applied to all cell calls and it was not respected by many topology calls.
+
+### <a id="vtadmin"/>VTAdmin
+
+#### <a id="updated-node"/>vtadmin-web updated to node v22.13.1 (LTS)
+
+Building `vtadmin-web` now requires node >= v22.13.0 (LTS). Breaking changes from v20 to v22 can be found at https://nodejs.org/en/blog/release/v22.13.0 -- with no known issues that apply to VTAdmin.
+Full details on the node v20.12.2 release can be found at https://nodejs.org/en/blog/release/v22.13.1.

doc/internal/release/how-to-release.md

@@ -364,7 +364,8 @@ You will need administrator privileges on the vitess repository to be able to ma
 
     ```bash
     cd ./java/
-    mvn clean deploy -P release -DskipTests
+    # For <= v21.0, we must use -DskipTests in the mvn command below
+    mvn clean deploy -P release
     cd ..
     ```
 

docker/binaries/vtadmin/Dockerfile

@@ -17,7 +17,7 @@ ARG DEBIAN_VER=bookworm-slim
 
 FROM vitess/lite:${VT_BASE_VER} AS lite
 
-FROM node:20-${DEBIAN_VER} as node
+FROM node:22-${DEBIAN_VER} as node
 
 # Prepare directory structure.
 RUN mkdir -p /vt/web

go/cmd/vtcombo/cli/vschema_watcher.go

@@ -56,17 +56,20 @@ func loadKeyspacesFromDir(ctx context.Context, dir string, ts *topo.Server) {
 				log.Fatalf("Unable to read keyspace file %v: %v", ksFile, err)
 			}
 
-			keyspace := &vschemapb.Keyspace{}
-			err = json.Unmarshal(jsonData, keyspace)
+			ksvs := &topo.KeyspaceVSchemaInfo{
+				Name:     ks.Name,
+				Keyspace: &vschemapb.Keyspace{},
+			}
+			err = json.Unmarshal(jsonData, ksvs.Keyspace)
 			if err != nil {
 				log.Fatalf("Unable to parse keyspace file %v: %v", ksFile, err)
 			}
 
-			_, err = vindexes.BuildKeyspace(keyspace, env.Parser())
+			_, err = vindexes.BuildKeyspace(ksvs.Keyspace, env.Parser())
 			if err != nil {
 				log.Fatalf("Invalid keyspace definition: %v", err)
 			}
-			ts.SaveVSchema(ctx, ks.Name, keyspace)
+			ts.SaveVSchema(ctx, ksvs)
 			log.Infof("Loaded keyspace %v from %v\n", ks.Name, ksFile)
 		}
 	}

go/cmd/vtctldclient/command/vreplication/lookupvindex/lookupvindex.go

@@ -72,6 +72,15 @@ var (
 
 	externalizeOptions = struct {
 		Keyspace string
+		Delete   bool
+	}{}
+
+	internalizeOptions = struct {
+		Keyspace string
+	}{}
+
+	completeOptions = struct {
+		Keyspace string
 	}{}
 
 	parseAndValidateCreate = func(cmd *cobra.Command, args []string) error {
@@ -142,6 +151,18 @@ var (
 		RunE:                  commandCancel,
 	}
 
+	// complete makes a LookupVindexComplete call to a vtctld.
+	complete = &cobra.Command{
+		Use:                   "complete",
+		Short:                 "Complete the LookupVindex workflow. The Vindex must have been previously externalized. If you want to delete the workflow without externalizing the Vindex then use the cancel command instead.",
+		Example:               `vtctldclient --server localhost:15999 LookupVindex --name corder_lookup_vdx --table-keyspace customer complete`,
+		SilenceUsage:          true,
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"Complete"},
+		Args:                  cobra.NoArgs,
+		RunE:                  commandComplete,
+	}
+
 	// create makes a LookupVindexCreate call to a vtctld.
 	create = &cobra.Command{
 		Use:                   "create",
@@ -158,7 +179,7 @@ var (
 	// externalize makes a LookupVindexExternalize call to a vtctld.
 	externalize = &cobra.Command{
 		Use:                   "externalize",
-		Short:                 "Externalize the Lookup Vindex. If the Vindex has an owner the VReplication workflow will also be deleted.",
+		Short:                 "Externalize the Lookup Vindex. If the Vindex has an owner the VReplication workflow will also be stopped/deleted.",
 		Example:               `vtctldclient --server localhost:15999 LookupVindex --name corder_lookup_vdx --table-keyspace customer externalize`,
 		SilenceUsage:          true,
 		DisableFlagsInUseLine: true,
@@ -167,6 +188,18 @@ var (
 		RunE:                  commandExternalize,
 	}
 
+	// internalize makes a LookupVindexInternalize call to a vtctld.
+	internalize = &cobra.Command{
+		Use:                   "internalize",
+		Short:                 "Internalize the Vindex again to continue the backfill, making it unusable for queries again.",
+		Example:               `vtctldclient --server localhost:15999 LookupVindex --name corder_lookup_vdx --table-keyspace customer internalize`,
+		SilenceUsage:          true,
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"Internalize"},
+		Args:                  cobra.NoArgs,
+		RunE:                  commandInternalize,
+	}
+
 	// show makes a GetWorkflows call to a vtctld.
 	show = &cobra.Command{
 		Use:                   "show",
@@ -199,6 +232,30 @@ func commandCancel(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+func commandComplete(cmd *cobra.Command, args []string) error {
+	if completeOptions.Keyspace == "" {
+		completeOptions.Keyspace = baseOptions.TableKeyspace
+	}
+	cli.FinishedParsing(cmd)
+
+	_, err := common.GetClient().LookupVindexComplete(common.GetCommandCtx(), &vtctldatapb.LookupVindexCompleteRequest{
+		Keyspace: completeOptions.Keyspace,
+		// The name of the workflow and lookup vindex.
+		Name: baseOptions.Name,
+		// Where the lookup table and VReplication workflow were created.
+		TableKeyspace: baseOptions.TableKeyspace,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	output := fmt.Sprintf("LookupVindex %s has been completed and the VReplication workflow has been deleted.", baseOptions.Name)
+	fmt.Println(output)
+
+	return nil
+}
+
 func commandCreate(cmd *cobra.Command, args []string) error {
 	tsp := common.GetTabletSelectionPreference(cmd)
 	cli.FinishedParsing(cmd)
@@ -236,21 +293,49 @@ func commandExternalize(cmd *cobra.Command, args []string) error {
 		Name: baseOptions.Name,
 		// Where the lookup table and VReplication workflow were created.
 		TableKeyspace: baseOptions.TableKeyspace,
+		// Delete the workflow after externalizing, instead of stopping.
+		DeleteWorkflow: externalizeOptions.Delete,
 	})
 
 	if err != nil {
 		return err
 	}
 
 	output := fmt.Sprintf("LookupVindex %s has been externalized", baseOptions.Name)
-	if resp.WorkflowDeleted {
-		output = output + fmt.Sprintf(" and the %s VReplication workflow has been deleted", baseOptions.Name)
+	if resp.WorkflowStopped {
+		output = output + " and the VReplication workflow has been stopped."
+	} else if resp.WorkflowDeleted {
+		output = output + " and the VReplication workflow has been deleted."
 	}
 	fmt.Println(output)
 
 	return nil
 }
 
+func commandInternalize(cmd *cobra.Command, args []string) error {
+	if internalizeOptions.Keyspace == "" {
+		internalizeOptions.Keyspace = baseOptions.TableKeyspace
+	}
+	cli.FinishedParsing(cmd)
+
+	_, err := common.GetClient().LookupVindexInternalize(common.GetCommandCtx(), &vtctldatapb.LookupVindexInternalizeRequest{
+		Keyspace: internalizeOptions.Keyspace,
+		// The name of the workflow and lookup vindex.
+		Name: baseOptions.Name,
+		// Where the lookup table and VReplication workflow were created.
+		TableKeyspace: baseOptions.TableKeyspace,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	output := fmt.Sprintf("LookupVindex %s has been internalized and the VReplication workflow has been started.", baseOptions.Name)
+	fmt.Println(output)
+
+	return nil
+}
+
 func commandShow(cmd *cobra.Command, args []string) error {
 	cli.FinishedParsing(cmd)
 
@@ -304,12 +389,19 @@ func registerCommands(root *cobra.Command) {
 	// for the VReplication workflow used.
 	base.AddCommand(show)
 
-	// This will also delete the VReplication workflow if the
+	// This will also stop the VReplication workflow if the
 	// vindex has an owner as the lookup vindex will then be
 	// managed by VTGate.
 	externalize.Flags().StringVar(&externalizeOptions.Keyspace, "keyspace", "", "The keyspace containing the Lookup Vindex. If no value is specified then the table-keyspace will be used.")
+	externalize.Flags().BoolVar(&externalizeOptions.Delete, "delete", false, "Delete the VReplication workflow after externalizing the Vindex, instead of stopping (default false).")
 	base.AddCommand(externalize)
 
+	internalize.Flags().StringVar(&internalizeOptions.Keyspace, "keyspace", "", "The keyspace containing the Lookup Vindex. If no value is specified then the table-keyspace will be used.")
+	base.AddCommand(internalize)
+
+	complete.Flags().StringVar(&completeOptions.Keyspace, "keyspace", "", "The keyspace containing the Lookup Vindex. If no value is specified then the table-keyspace will be used.")
+	base.AddCommand(complete)
+
 	// The cancel command deletes the VReplication workflow used
 	// to backfill the lookup vindex. It ends up making a
 	// WorkflowDelete VtctldServer call.

go/cmd/vtctldclient/command/vreplication/workflow/state.go

@@ -60,14 +60,17 @@ func commandUpdateState(cmd *cobra.Command, args []string) error {
 	cli.FinishedParsing(cmd)
 
 	var state binlogdatapb.VReplicationWorkflowState
+	var shards []string
 	switch strings.ToLower(cmd.Name()) {
 	case "start":
 		if err := common.CanRestartWorkflow(baseOptions.Keyspace, baseOptions.Workflow); err != nil {
 			return err
 		}
 		state = binlogdatapb.VReplicationWorkflowState_Running
+		shards = baseOptions.Shards
 	case "stop":
 		state = binlogdatapb.VReplicationWorkflowState_Stopped
+		shards = baseOptions.Shards
 	default:
 		return fmt.Errorf("invalid workflow state: %s", args[0])
 	}
@@ -80,6 +83,7 @@ func commandUpdateState(cmd *cobra.Command, args []string) error {
 			Cells:       textutil.SimulatedNullStringSlice,
 			TabletTypes: textutil.SimulatedNullTabletTypeSlice,
 			State:       &state,
+			Shards:      shards,
 		},
 	}
 

go/mysql/binlog_dump.go

@@ -76,11 +76,13 @@ func (c *Conn) parseComBinlogDumpGTID(data []byte) (logFile string, logPos uint6
 	if !ok {
 		return logFile, logPos, position, readPacketErr
 	}
-	if gtid := string(data[pos : pos+int(dataSize)]); gtid != "" {
-		position, err = replication.DecodePosition(gtid)
+	if gtidBytes := data[pos : pos+int(dataSize)]; len(gtidBytes) != 0 {
+		gtid, err := replication.NewMysql56GTIDSetFromSIDBlock(gtidBytes)
 		if err != nil {
-			return logFile, logPos, position, err
+			return logFile, logPos, position, vterrors.Wrapf(err, "error parsing GTID from BinlogDumpGTID packet")
 		}
+		// ComBinlogDumpGTID is a MySQL specific protocol. The GTID flavor is necessarily MySQL 56
+		position = replication.Position{GTIDSet: gtid}
 	}
 	if flags2&BinlogDumpNonBlock != 0 {
 		return logFile, logPos, position, io.EOF

go/mysql/flavor_mysql.go

@@ -218,7 +218,10 @@ func (mysqlFlavor) sendBinlogDumpCommand(c *Conn, serverID uint32, binlogFilenam
 	}
 
 	// Build the command.
-	sidBlock := gtidSet.SIDBlock()
+	var sidBlock []byte
+	if gtidSet != nil {
+		sidBlock = gtidSet.SIDBlock()
+	}
 	var flags2 uint16
 	if binlogFilename != "" {
 		flags2 |= BinlogThroughPosition

go/mysql/replication.go

@@ -81,7 +81,8 @@ func (c *Conn) AnalyzeSemiSyncAckRequest(buf []byte) (strippedBuf []byte, ackReq
 // WriteComBinlogDumpGTID writes a ComBinlogDumpGTID command.
 // Only works with MySQL 5.6+ (and not MariaDB).
 // See http://dev.mysql.com/doc/internals/en/com-binlog-dump-gtid.html for syntax.
-func (c *Conn) WriteComBinlogDumpGTID(serverID uint32, binlogFilename string, binlogPos uint64, flags uint16, gtidSet []byte) error {
+// sidBlock must be the result of a gtidSet.SIDBlock() function.
+func (c *Conn) WriteComBinlogDumpGTID(serverID uint32, binlogFilename string, binlogPos uint64, flags uint16, sidBlock []byte) error {
 	c.sequence = 0
 	length := 1 + // ComBinlogDumpGTID
 		2 + // flags
@@ -90,16 +91,16 @@ func (c *Conn) WriteComBinlogDumpGTID(serverID uint32, binlogFilename string, bi
 		len(binlogFilename) + // binlog-filename
 		8 + // binlog-pos
 		4 + // data-size
-		len(gtidSet) // data
+		len(sidBlock) // data
 	data, pos := c.startEphemeralPacketWithHeader(length)
 	pos = writeByte(data, pos, ComBinlogDumpGTID)             // nolint
 	pos = writeUint16(data, pos, flags)                       // nolint
 	pos = writeUint32(data, pos, serverID)                    // nolint
 	pos = writeUint32(data, pos, uint32(len(binlogFilename))) // nolint
 	pos = writeEOFString(data, pos, binlogFilename)           // nolint
 	pos = writeUint64(data, pos, binlogPos)                   // nolint
-	pos = writeUint32(data, pos, uint32(len(gtidSet)))        // nolint
-	pos += copy(data[pos:], gtidSet)                          // nolint
+	pos = writeUint32(data, pos, uint32(len(sidBlock)))       // nolint
+	pos += copy(data[pos:], sidBlock)                         // nolint
 	if err := c.writeEphemeralPacket(); err != nil {
 		return sqlerror.NewSQLErrorf(sqlerror.CRServerGone, sqlerror.SSUnknownSQLState, "%v", err)
 	}