Add option to run vtcombo and vttest local cluster with real topo server

go/cmd/vtcombo/main.go

@@ -59,6 +59,9 @@ var (
 
 	mysqlPort = flag.Int("mysql_port", 3306, "mysql port")
 
+	remoteTopoServer = flag.Bool("remote_topo_server", false, "Should vtcombo use a remote topology server instead of starting its own in-memory topology server. "+
+		"If true, vtcombo will use the flags defined in topo/server.go to open topo server")
+
 	ts              *topo.Server
 	resilientServer *srvtopo.ResilientServer
 )
@@ -135,8 +138,14 @@ func main() {
 		flag.Set("log_dir", "$VTDATAROOT/tmp")
 	}
 
-	// Create topo server. We use a 'memorytopo' implementation.
-	ts = memorytopo.NewServer(tpb.Cells...)
+	if *remoteTopoServer {
+		// Open topo server based on the command line flags defined at topo/server.go
+		// do not create cell info as it should be done by whoever sets up the remote topo server
+		ts = topo.Open()
+	} else {
+		// Create topo server. We use a 'memorytopo' implementation.
+		ts = memorytopo.NewServer(tpb.Cells...)
+	}
 	servenv.Init()
 	tabletenv.Init()
 

go/cmd/vttestserver/main.go

@@ -151,6 +151,11 @@ func init() {
 	flag.StringVar(&config.ForeignKeyMode, "foreign_key_mode", "allow", "This is to provide how to handle foreign key constraint in create/alter table. Valid values are: allow, disallow")
 	flag.BoolVar(&config.EnableOnlineDDL, "enable_online_ddl", true, "Allow users to submit, review and control Online DDL")
 	flag.BoolVar(&config.EnableDirectDDL, "enable_direct_ddl", true, "Allow users to submit direct DDL statements")
+
+	// flags for using an actual topo implementation for vtcombo instead of in-memory topo. useful for test setup where topo server is shared across multiple vtcombo processes or other components
+	flag.StringVar(&config.RemoteTopoImplementation, "remote_topo_implementation", "", "the topology implementation to use for vtcombo process")
+	flag.StringVar(&config.RemoteTopoGlobalServerAddress, "remote_topo_global_server_address", "", "the address of the global topology server for vtcombo process")
+	flag.StringVar(&config.RemoteTopoGlobalRoot, "remote_topo_global_root", "", "the path of the global topology data in the global topology server for vtcombo process")
 }
 
 func (t *topoFlags) buildTopology() (*vttestpb.VTTestTopology, error) {

go/vt/vttest/environment.go

@@ -24,6 +24,8 @@ import (
 	"strings"
 	"time"
 
+	"vitess.io/vitess/go/vt/proto/vttest"
+
 	// we use gRPC everywhere, so import the vtgate client.
 	_ "vitess.io/vitess/go/vt/vtgate/grpcvtgateconn"
 )
@@ -45,6 +47,12 @@ type Environment interface {
 	// See: vttest.MySQLManager for the interface the manager must implement
 	MySQLManager(mycnf []string, snapshot string) (MySQLManager, error)
 
+	// TopoManager is the constructor for the Topology manager that will
+	// be used by the cluster. It's only used when we run the local cluster with
+	// a remote topo server instead of in-memory topo server within vtcombo process
+	// See: vttest.TopoManager for the interface of topo manager
+	TopoManager(topoImplementation, topoServerAddress, topoRoot string, topology *vttest.VTTestTopology) TopoManager
+
 	// Directory is the path where the local cluster will store all its
 	// data and metadata. For local testing, this should probably be an
 	// unique temporary directory.
@@ -147,6 +155,16 @@ func (env *LocalTestEnv) MySQLManager(mycnf []string, snapshot string) (MySQLMan
 	}, nil
 }
 
+// TopoManager implements TopoManager for LocalTestEnv
+func (env *LocalTestEnv) TopoManager(topoImplementation, topoServerAddress, topoRoot string, topology *vttest.VTTestTopology) TopoManager {
+	return &Topoctl{
+		TopoImplementation:      topoImplementation,
+		TopoGlobalServerAddress: topoServerAddress,
+		TopoGlobalRoot:          topoRoot,
+		Topology:                topology,
+	}
+}
+
 // DefaultProtocol implements DefaultProtocol for LocalTestEnv.
 // It is always GRPC.
 func (env *LocalTestEnv) DefaultProtocol() string {

go/vt/vttest/local_cluster.go

@@ -125,6 +125,13 @@ type Config struct {
 
 	// Allow users to submit direct DDL statements
 	EnableDirectDDL bool
+
+	// Allow users to start a local cluster using a remote topo server
+	RemoteTopoImplementation string
+
+	RemoteTopoGlobalServerAddress string
+
+	RemoteTopoGlobalRoot string
 }
 
 // InitSchemas is a shortcut for tests that just want to setup a single
@@ -200,6 +207,7 @@ type LocalCluster struct {
 	Env Environment
 
 	mysql MySQLManager
+	topo  TopoManager
 	vt    *VtProcess
 }
 
@@ -237,6 +245,16 @@ func (db *LocalCluster) Setup() error {
 
 	log.Infof("LocalCluster environment: %+v", db.Env)
 
+	// Set up topo manager if we are using a remote topo server
+	if db.RemoteTopoImplementation != "" {
+		db.topo = db.Env.TopoManager(db.RemoteTopoImplementation, db.RemoteTopoGlobalServerAddress, db.RemoteTopoGlobalRoot, db.Topology)
+		log.Infof("Initializing Topo Manager (%T)...", db.topo)
+		if err := db.topo.Setup(); err != nil {
+			log.Errorf("Failed to set up Topo Manager: %s", err)
+			return err
+		}
+	}
+
 	db.mysql, err = db.Env.MySQLManager(db.ExtraMyCnf, db.SnapshotFile)
 	if err != nil {
 		return err

go/vt/vttest/plugin_consultopo.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2021 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 agreedto 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 vttest
+
+// This plugin imports consultopo to register the consul implementation of TopoServer.
+
+import (
+	_ "vitess.io/vitess/go/vt/topo/consultopo"
+)

go/vt/vttest/plugin_etcd2topo.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2021 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 vttest
+
+// This plugin imports etcd2topo to register the etcd2 implementation of TopoServer.
+
+import (
+	_ "vitess.io/vitess/go/vt/topo/etcd2topo"
+)

go/vt/vttest/plugin_grpctmclient.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2021 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 vttest
+
+// Imports and register the gRPC tabletmanager client
+
+import (
+	_ "vitess.io/vitess/go/vt/vttablet/grpctmclient"
+)

go/vt/vttest/plugin_zk2topo.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2021 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 agreedto 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 vttest
+
+// Imports and register the zk2 TopologyServer
+
+import (
+	_ "vitess.io/vitess/go/vt/topo/zk2topo"
+)

go/vt/vttest/topoctl.go

@@ -0,0 +1,59 @@
+package vttest
+
+import (
+	"context"
+	"time"
+
+	"vitess.io/vitess/go/vt/log"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	"vitess.io/vitess/go/vt/proto/vttest"
+	"vitess.io/vitess/go/vt/topo"
+)
+
+// TopoManager is an interface to manage things in a remote topology server for local cluster
+type TopoManager interface {
+	Setup() error
+}
+
+// Topoctl implements TopoManager. For now it only sets up cells in remote topology server if any of them doesn't exist.
+type Topoctl struct {
+	TopoImplementation      string
+	TopoGlobalServerAddress string
+	TopoGlobalRoot          string
+	Topology                *vttest.VTTestTopology
+}
+
+func (ctl *Topoctl) Setup() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	topoServer, err := topo.OpenServer(ctl.TopoImplementation, ctl.TopoGlobalServerAddress, ctl.TopoGlobalRoot)
+	if err != nil {
+		return err
+	}
+
+	log.Infof("Creating cells if they don't exist in the provided topo server %s %s %s", ctl.TopoImplementation, ctl.TopoGlobalServerAddress, ctl.TopoGlobalRoot)
+	// Create cells if it doesn't exist to be idempotent. Should work when we share the same topo server across multiple local clusters.
+	for _, cell := range ctl.Topology.Cells {
+		_, err := topoServer.GetCellInfo(ctx, cell, true)
+		// Cell already exists. no-op
+		if err == nil {
+			continue
+		}
+
+		// Return any error that's not NoNode
+		if !topo.IsErrType(err, topo.NoNode) {
+			return err
+		}
+
+		// try to create the cell that doesn't exist
+		// Use empty cell info till we have a use case to set up local cell properly
+		err = topoServer.CreateCellInfo(ctx, cell, &topodatapb.CellInfo{})
+		if err != nil {
+			return err
+		}
+		log.Infof("Created cell info for %s in the topo server %s %s %s", cell, ctl.TopoImplementation, ctl.TopoGlobalServerAddress, ctl.TopoGlobalRoot)
+	}
+
+	return nil
+}

go/vt/vttest/vtprocess.go

@@ -285,5 +285,14 @@ func VtcomboProcess(env Environment, args *Config, mysql MySQLManager) *VtProces
 		"-mysql_server_bind_address", vtcomboMysqlBindAddress,
 	}...)
 
+	if args.RemoteTopoImplementation != "" {
+		vt.ExtraArgs = append(vt.ExtraArgs, []string{
+			"-remote_topo_server",
+			"-topo_implementation", args.RemoteTopoImplementation,
+			"-topo_global_server_address", args.RemoteTopoGlobalServerAddress,
+			"-topo_global_root", args.RemoteTopoGlobalRoot,
+		}...)
+	}
+
 	return vt
 }