Merge pull request #7484 from planetscale/consul-qs-fix

Fix for Query Serving when Toposerver is Down

go/test/endtoend/cluster/cluster_process.go

@@ -56,6 +56,7 @@ type LocalProcessCluster struct {
 	Cell               string
 	BaseTabletUID      int
 	Hostname           string
+	TopoFlavor         string
 	TopoPort           int
 	TmpDirectory       string
 	OriginalVTDATAROOT string
@@ -173,6 +174,8 @@ func (cluster *LocalProcessCluster) StartTopo() (err error) {
 	if cluster.Cell == "" {
 		cluster.Cell = DefaultCell
 	}
+
+	topoFlavor = cluster.TopoFlavorString()
 	cluster.TopoPort = cluster.GetAndReservePort()
 	cluster.TmpDirectory = path.Join(os.Getenv("VTDATAROOT"), fmt.Sprintf("/tmp_%d", cluster.GetAndReservePort()))
 	cluster.TopoProcess = *TopoProcessInstance(cluster.TopoPort, cluster.GetAndReservePort(), cluster.Hostname, *topoFlavor, "global")
@@ -758,9 +761,13 @@ func (cluster *LocalProcessCluster) StartVttablet(tablet *Vttablet, servingStatu
 	return tablet.VttabletProcess.Setup()
 }
 
-//func (cluster *LocalProcessCluster) NewOrcInstance() OrchestratorProcess {
-//
-//}
+// TopoFlavorString returns the topo flavor
+func (cluster *LocalProcessCluster) TopoFlavorString() *string {
+	if cluster.TopoFlavor != "" {
+		return &cluster.TopoFlavor
+	}
+	return topoFlavor
+}
 
 func getCoveragePath(fileName string) string {
 	covDir := os.Getenv("COV_DIR")

go/test/endtoend/cluster/topo_process.go

@@ -230,7 +230,7 @@ func (topo *TopoProcess) TearDown(Cell string, originalVtRoot string, currentRoo
 		// Attempt graceful shutdown with SIGTERM first
 		_ = topo.proc.Process.Signal(syscall.SIGTERM)
 
-		if !*keepData {
+		if !(*keepData || keepdata) {
 			_ = os.RemoveAll(topo.DataDirectory)
 			_ = os.RemoveAll(currentRoot)
 		}

go/test/endtoend/topotest/consul/main_test.go

@@ -0,0 +1,148 @@
+/*
+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 consul
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"vitess.io/vitess/go/vt/log"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/test/endtoend/cluster"
+)
+
+var (
+	clusterInstance *cluster.LocalProcessCluster
+	cell            = "zone1"
+	hostname        = "localhost"
+	KeyspaceName    = "customer"
+	SchemaSQL       = `
+CREATE TABLE t1 (
+    c1 BIGINT NOT NULL,
+    c2 BIGINT NOT NULL,
+    c3 BIGINT,
+    c4 varchar(100),
+    PRIMARY KEY (c1),
+    UNIQUE KEY (c2),
+    UNIQUE KEY (c3),
+    UNIQUE KEY (c4)
+) ENGINE=Innodb;`
+	VSchema = `
+{
+    "sharded": false,
+    "tables": {
+        "t1": {}
+    }
+}
+`
+)
+
+func TestMain(m *testing.M) {
+	defer cluster.PanicHandler(nil)
+	flag.Parse()
+
+	exitCode := func() int {
+		clusterInstance = cluster.NewCluster(cell, hostname)
+		defer clusterInstance.Teardown()
+
+		// Start topo server
+		clusterInstance.TopoFlavor = "consul"
+		if err := clusterInstance.StartTopo(); err != nil {
+			return 1
+		}
+
+		// Start keyspace
+		Keyspace := &cluster.Keyspace{
+			Name:      KeyspaceName,
+			SchemaSQL: SchemaSQL,
+			VSchema:   VSchema,
+		}
+		if err := clusterInstance.StartUnshardedKeyspace(*Keyspace, 0, false); err != nil {
+			log.Fatal(err.Error())
+			return 1
+		}
+
+		// Start vtgate
+		if err := clusterInstance.StartVtgate(); err != nil {
+			log.Fatal(err.Error())
+			return 1
+		}
+
+		return m.Run()
+	}()
+	os.Exit(exitCode)
+}
+
+func TestTopoDownServingQuery(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	vtParams := mysql.ConnParams{
+		Host: "localhost",
+		Port: clusterInstance.VtgateMySQLPort,
+	}
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	defer exec(t, conn, `delete from t1`)
+
+	execMulti(t, conn, `insert into t1(c1, c2, c3, c4) values (300,100,300,'abc'); ;; insert into t1(c1, c2, c3, c4) values (301,101,301,'abcd');;`)
+	assertMatches(t, conn, `select c1,c2,c3 from t1`, `[[INT64(300) INT64(100) INT64(300)] [INT64(301) INT64(101) INT64(301)]]`)
+	clusterInstance.TopoProcess.TearDown(clusterInstance.Cell, clusterInstance.OriginalVTDATAROOT, clusterInstance.CurrentVTDATAROOT, true, *clusterInstance.TopoFlavorString())
+	time.Sleep(3 * time.Second)
+	assertMatches(t, conn, `select c1,c2,c3 from t1`, `[[INT64(300) INT64(100) INT64(300)] [INT64(301) INT64(101) INT64(301)]]`)
+}
+
+func exec(t *testing.T, conn *mysql.Conn, query string) *sqltypes.Result {
+	t.Helper()
+	qr, err := conn.ExecuteFetch(query, 1000, true)
+	require.NoError(t, err)
+	return qr
+}
+
+func execMulti(t *testing.T, conn *mysql.Conn, query string) []*sqltypes.Result {
+	t.Helper()
+	var res []*sqltypes.Result
+	qr, more, err := conn.ExecuteFetchMulti(query, 1000, true)
+	res = append(res, qr)
+	require.NoError(t, err)
+	for more == true {
+		qr, more, _, err = conn.ReadQueryResult(1000, true)
+		require.NoError(t, err)
+		res = append(res, qr)
+	}
+	return res
+}
+
+func assertMatches(t *testing.T, conn *mysql.Conn, query, expected string) {
+	t.Helper()
+	qr := exec(t, conn, query)
+	got := fmt.Sprintf("%v", qr.Rows)
+	diff := cmp.Diff(expected, got)
+	if diff != "" {
+		t.Errorf("Query: %s (-want +got):\n%s", query, diff)
+	}
+}

go/test/endtoend/topotest/etcd2/main_test.go

@@ -0,0 +1,147 @@
+/*
+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 ectd2
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"vitess.io/vitess/go/vt/log"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/test/endtoend/cluster"
+)
+
+var (
+	clusterInstance *cluster.LocalProcessCluster
+	cell            = "zone1"
+	hostname        = "localhost"
+	KeyspaceName    = "customer"
+	SchemaSQL       = `
+CREATE TABLE t1 (
+    c1 BIGINT NOT NULL,
+    c2 BIGINT NOT NULL,
+    c3 BIGINT,
+    c4 varchar(100),
+    PRIMARY KEY (c1),
+    UNIQUE KEY (c2),
+    UNIQUE KEY (c3),
+    UNIQUE KEY (c4)
+) ENGINE=Innodb;`
+	VSchema = `
+{
+    "sharded": false,
+    "tables": {
+        "t1": {}
+    }
+}
+`
+)
+
+func TestMain(m *testing.M) {
+	defer cluster.PanicHandler(nil)
+	flag.Parse()
+
+	exitCode := func() int {
+		clusterInstance = cluster.NewCluster(cell, hostname)
+		defer clusterInstance.Teardown()
+
+		// Start topo server
+		if err := clusterInstance.StartTopo(); err != nil {
+			return 1
+		}
+
+		// Start keyspace
+		Keyspace := &cluster.Keyspace{
+			Name:      KeyspaceName,
+			SchemaSQL: SchemaSQL,
+			VSchema:   VSchema,
+		}
+		if err := clusterInstance.StartUnshardedKeyspace(*Keyspace, 0, false); err != nil {
+			log.Fatal(err.Error())
+			return 1
+		}
+
+		// Start vtgate
+		if err := clusterInstance.StartVtgate(); err != nil {
+			log.Fatal(err.Error())
+			return 1
+		}
+
+		return m.Run()
+	}()
+	os.Exit(exitCode)
+}
+
+func TestTopoDownServingQuery(t *testing.T) {
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+	vtParams := mysql.ConnParams{
+		Host: "localhost",
+		Port: clusterInstance.VtgateMySQLPort,
+	}
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.Nil(t, err)
+	defer conn.Close()
+
+	defer exec(t, conn, `delete from t1`)
+
+	execMulti(t, conn, `insert into t1(c1, c2, c3, c4) values (300,100,300,'abc'); ;; insert into t1(c1, c2, c3, c4) values (301,101,301,'abcd');;`)
+	assertMatches(t, conn, `select c1,c2,c3 from t1`, `[[INT64(300) INT64(100) INT64(300)] [INT64(301) INT64(101) INT64(301)]]`)
+	clusterInstance.TopoProcess.TearDown(clusterInstance.Cell, clusterInstance.OriginalVTDATAROOT, clusterInstance.CurrentVTDATAROOT, true, *clusterInstance.TopoFlavorString())
+	time.Sleep(3 * time.Second)
+	assertMatches(t, conn, `select c1,c2,c3 from t1`, `[[INT64(300) INT64(100) INT64(300)] [INT64(301) INT64(101) INT64(301)]]`)
+}
+
+func exec(t *testing.T, conn *mysql.Conn, query string) *sqltypes.Result {
+	t.Helper()
+	qr, err := conn.ExecuteFetch(query, 1000, true)
+	require.NoError(t, err)
+	return qr
+}
+
+func execMulti(t *testing.T, conn *mysql.Conn, query string) []*sqltypes.Result {
+	t.Helper()
+	var res []*sqltypes.Result
+	qr, more, err := conn.ExecuteFetchMulti(query, 1000, true)
+	res = append(res, qr)
+	require.NoError(t, err)
+	for more == true {
+		qr, more, _, err = conn.ReadQueryResult(1000, true)
+		require.NoError(t, err)
+		res = append(res, qr)
+	}
+	return res
+}
+
+func assertMatches(t *testing.T, conn *mysql.Conn, query, expected string) {
+	t.Helper()
+	qr := exec(t, conn, query)
+	got := fmt.Sprintf("%v", qr.Rows)
+	diff := cmp.Diff(expected, got)
+	if diff != "" {
+		t.Errorf("Query: %s (-want +got):\n%s", query, diff)
+	}
+}