Repository: vitessio/vitess
Pull request #8298 (2021-06-09 18:24:34)

Expose topo_consul_lock_session_checks to allow customized consul session check
Signed-off-by: crowu <y.wu4515@gmail.com>

<!--
  Thank you for your contribution to the Vitess project.
  How to contribute: https://vitess.io/docs/contributing/
  Please first make sure there is an open Issue to discuss the feature/fix suggested in this PR.
  If this is a new feature, please mark the Issue as "RFC".
 -->

<!-- if this PR is Work in Progress please create it as a Draft Pull Request -->

## Description
<!-- A few sentences describing the overall goals of the pull request's commits. -->
<!-- If this is a bug fix and you think the fix should be backported, please write so. -->
When Consul creates lock, it needs to create a session for the lock. By default, session uses serfHealth check for liveness check: https://www.consul.io/api-docs/session, which is a node level healthy-check from gossip protocal. 

This means when the process restarted, if the node is still healthy, the session will remain alive up to 2*TTL (15sec) = 30 sec - during this period of time, other process cannot get the lock.

To address that, this PR exposes an optional flag to pass in customized session checks so the the session will expire automatically once the health check failed.

## Related Issue(s)
<!-- List related issues and pull requests. If this PR fixes an issue, please add it using Fixes #????  -->
https://github.com/vitessio/vitess/issues/8297

## Checklist
- [x] Tests were added or are not required
- [x] Documentation was added or is not required

## Deployment Notes
<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->

narcsfz: I see tests failed with error:
```
--- FAIL: TestConsulTopoWithChecks (0.24s)
Error:     testing.go:49: === checkKeyspace
Error:     server_flaky_test.go:302: OpenServer() failed: open /tmp/consul_auth_client_static_file.json160052189: no such file or directory
```

But it works on my IDE locally..
narcsfz: hm.. looks like [commenting out](https://github.com/vitessio/vitess/pull/8298/commits/1749fb5515c357714d55d2af7c7edba456bba7f1) my change fixes the test.. I will look into that
narcsfz: ptal @deepthi @rafael 
deepthi: We'll need @rafael to review this (unless @guidoiaquinti or @ajm188 can).
In the meantime, can you squash the commits into a single one?
narcsfz: Thanks, I cleaned up the commits
deepthi: @narcsfz can you resolve the merge conflicts and push again?

Patch:
diff --git a/go/vt/topo/consultopo/lock.go b/go/vt/topo/consultopo/lock.go
index b89ced6013..6afa9571c9 100644
--- a/go/vt/topo/consultopo/lock.go
+++ b/go/vt/topo/consultopo/lock.go
@@ -51,11 +51,23 @@ func (s *Server) Lock(ctx context.Context, dirPath, contents string) (topo.LockD
 
 	lockPath := path.Join(s.root, dirPath, locksFilename)
 
-	// Build the lock structure.
-	l, err := s.client.LockOpts(&api.LockOptions{
+	lockOpts := &api.LockOptions{
 		Key:   lockPath,
 		Value: []byte(contents),
-	})
+		SessionOpts: &api.SessionEntry{
+			Name: api.DefaultLockSessionName,
+			TTL:  api.DefaultLockSessionTTL,
+		},
+	}
+	lockOpts.SessionOpts.Checks = s.lockChecks
+	if s.lockDelay > 0 {
+		lockOpts.SessionOpts.LockDelay = s.lockDelay
+	}
+	if s.lockTTL != "" {
+		lockOpts.SessionOpts.TTL = s.lockTTL
+	}
+	// Build the lock structure.
+	l, err := s.client.LockOpts(lockOpts)
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/vt/topo/consultopo/server.go b/go/vt/topo/consultopo/server.go
index a90db57a54..0b3cbcdd97 100644
--- a/go/vt/topo/consultopo/server.go
+++ b/go/vt/topo/consultopo/server.go
@@ -24,7 +24,9 @@ import (
 	"flag"
 	"fmt"
 	"io/ioutil"
+	"strings"
 	"sync"
+	"time"
 
 	"github.com/hashicorp/consul/api"
 
@@ -36,6 +38,10 @@ import (
 
 var (
 	consulAuthClientStaticFile = flag.String("consul_auth_static_file", "", "JSON File to read the topos/tokens from.")
+	// serfHealth is the default check from consul
+	consulLockSessionChecks = flag.String("topo_consul_lock_session_checks", "serfHealth", "List of checks for consul session.")
+	consulLockSessionTTL    = flag.String("topo_consul_lock_session_ttl", "", "TTL for consul session.")
+	consulLockDelay         = flag.Duration("topo_consul_lock_delay", 15*time.Second, "LockDelay for consul session.")
 )
 
 // ClientAuthCred credential to use for consul clusters
@@ -93,6 +99,10 @@ type Server struct {
 	// locks is a map of *lockInstance structures.
 	// The key is the filepath of the Lock file.
 	locks map[string]*lockInstance
+
+	lockChecks []string
+	lockTTL    string
+	lockDelay  time.Duration
 }
 
 // lockInstance keeps track of one lock held by this client.
@@ -126,13 +136,24 @@ func NewServer(cell, serverAddr, root string) (*Server, error) {
 	}
 
 	return &Server{
-		client: client,
-		kv:     client.KV(),
-		root:   root,
-		locks:  make(map[string]*lockInstance),
+		client:     client,
+		kv:         client.KV(),
+		root:       root,
+		locks:      make(map[string]*lockInstance),
+		lockChecks: parseConsulLockSessionChecks(*consulLockSessionChecks),
+		lockTTL:    *consulLockSessionTTL,
+		lockDelay:  *consulLockDelay,
 	}, nil
 }
 
+func parseConsulLockSessionChecks(s string) []string {
+	var res []string
+	if len(s) == 0 {
+		return res
+	}
+	return strings.Split(*consulLockSessionChecks, ",")
+}
+
 // Close implements topo.Server.Close.
 // It will nil out the global and cells fields, so any attempt to
 // re-use this server will panic.
diff --git a/go/vt/topo/k8stopo/server.go b/go/vt/topo/k8stopo/server.go
index bb71a20e2d..b51e9fc144 100644
--- a/go/vt/topo/k8stopo/server.go
+++ b/go/vt/topo/k8stopo/server.go
@@ -49,7 +49,7 @@ import (
 
 var (
 	// kubeconfigPath is a string that gives the location of a valid kubeconfig file
-	kubeconfigPath = flag.String("topo_k8s_kubeconfig", "", "Path to a valid kubeconfig file. When running as a k8s pod inside the same cluster you wish to use as the topo, you may omit this and the below arguments, and Vitess is capable of auto-discovering the correct values. https://kubernetes.io/docs/tasks/access-application-cluster/access-cluster/#accessing-the-api-from-a-pod")
+	kubeconfigPath = flag.String("topo_k8s_kubeconfig", "", "Path to a valid kubeconfig file.")
 
 	// configContext is a string that can be used to override the default context
 	configContext = flag.String("topo_k8s_context", "", "The kubeconfig context to use, overrides the 'current-context' from the config")
diff --git a/go/vt/vtgate/evalengine/arithmetic.go b/go/vt/vtgate/evalengine/arithmetic.go
index 7bc883acda..3a887d7776 100644
--- a/go/vt/vtgate/evalengine/arithmetic.go
+++ b/go/vt/vtgate/evalengine/arithmetic.go
@@ -243,7 +243,7 @@ func isByteComparable(v sqltypes.Value) bool {
 		return true
 	}
 	switch v.Type() {
-	case sqltypes.Timestamp, sqltypes.Date, sqltypes.Time, sqltypes.Datetime, sqltypes.Enum, sqltypes.Set, sqltypes.TypeJSON, sqltypes.Bit:
+	case sqltypes.Timestamp, sqltypes.Date, sqltypes.Time, sqltypes.Datetime, sqltypes.Enum, sqltypes.Set, sqltypes.TypeJSON:
 		return true
 	}
 	return false
diff --git a/go/vt/wrangler/vdiff.go b/go/vt/wrangler/vdiff.go
index 868a4104ea..98dc89c3ae 100644
--- a/go/vt/wrangler/vdiff.go
+++ b/go/vt/wrangler/vdiff.go
@@ -253,10 +253,10 @@ func (wr *Wrangler) VDiff(ctx context.Context, targetKeyspace, workflowName, sou
 		}
 		// Perform the diff of source and target streams.
 		dr, err := td.diff(ctx, df.ts.wr, &rowsToCompare, debug, onlyPks)
+		dr.TableName = table
 		if err != nil {
 			return nil, vterrors.Wrap(err, "diff")
 		}
-		dr.TableName = table
 		diffReports[table] = dr
 	}
 	if format == "json" {


Test patch:
diff --git a/go/test/endtoend/cluster/vtctl_process.go b/go/test/endtoend/cluster/vtctl_process.go
index b27cdea991..b2dfe753f3 100644
--- a/go/test/endtoend/cluster/vtctl_process.go
+++ b/go/test/endtoend/cluster/vtctl_process.go
@@ -58,11 +58,19 @@ func (vtctl *VtctlProcess) AddCellInfo(Cell string) (err error) {
 
 // CreateKeyspace executes vtctl command to create keyspace
 func (vtctl *VtctlProcess) CreateKeyspace(keyspace string) (err error) {
-	output, err := vtctl.ExecuteCommandWithOutput("CreateKeyspace", keyspace)
-	if err != nil {
-		log.Errorf("CreateKeyspace returned err: %s, output: %s", err, output)
+	tmpProcess := exec.Command(
+		vtctl.Binary,
+		"-topo_implementation", vtctl.TopoImplementation,
+		"-topo_global_server_address", vtctl.TopoGlobalAddress,
+		"-topo_global_root", vtctl.TopoGlobalRoot,
+	)
+	if *isCoverage {
+		tmpProcess.Args = append(tmpProcess.Args, "-test.coverprofile="+getCoveragePath("vtctl-create-ks.out"))
 	}
-	return err
+	tmpProcess.Args = append(tmpProcess.Args,
+		"CreateKeyspace", keyspace)
+	log.Infof("Running CreateKeyspace with command: %v", strings.Join(tmpProcess.Args, " "))
+	return tmpProcess.Run()
 }
 
 // ExecuteCommandWithOutput executes any vtctlclient command and returns output
diff --git a/go/test/endtoend/vreplication/config.go b/go/test/endtoend/vreplication/config.go
index 74e67622e2..8ffe0724ae 100644
--- a/go/test/endtoend/vreplication/config.go
+++ b/go/test/endtoend/vreplication/config.go
@@ -3,7 +3,7 @@ package vreplication
 var (
 	initialProductSchema = `
 create table product(pid int, description varbinary(128), primary key(pid));
-create table customer(cid int, name varbinary(128), meta json default null, typ enum('individual','soho','enterprise'), sport set('football','cricket','baseball'),ts timestamp not null default current_timestamp, bits bit(2) default b'11', primary key(cid))  CHARSET=utf8mb4;
+create table customer(cid int, name varbinary(128), meta json default null, typ enum('individual','soho','enterprise'), sport set('football','cricket','baseball'),ts timestamp not null default current_timestamp, primary key(cid))  CHARSET=utf8mb4;
 create table customer_seq(id int, next_id bigint, cache bigint, primary key(id)) comment 'vitess_sequence';
 create table merchant(mname varchar(128), category varchar(128), primary key(mname)) DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
 create table orders(oid int, cid int, pid int, mname varchar(128), price int, qty int, total int as (qty * price), total2 int as (qty * price) stored, primary key(oid));
diff --git a/go/vt/topo/consultopo/server_flaky_test.go b/go/vt/topo/consultopo/server_flaky_test.go
index 235d4fbc22..73dacc4008 100644
--- a/go/vt/topo/consultopo/server_flaky_test.go
+++ b/go/vt/topo/consultopo/server_flaky_test.go
@@ -170,6 +170,52 @@ func TestConsulTopo(t *testing.T) {
 	})
 }
 
+func TestConsulTopoWithChecks(t *testing.T) {
+	// One test is going to wait that full period, so make it shorter.
+	*watchPollDuration = 100 * time.Millisecond
+	*consulLockSessionChecks = "serfHealth"
+	*consulLockSessionTTL = "15s"
+
+	// Start a single consul in the background.
+	cmd, configFilename, serverAddr := startConsul(t, "")
+	defer func() {
+		// Alerts command did not run successful
+		if err := cmd.Process.Kill(); err != nil {
+			log.Errorf("cmd process kill has an error: %v", err)
+		}
+		// Alerts command did not run successful
+		if err := cmd.Wait(); err != nil {
+			log.Errorf("cmd wait has an error: %v", err)
+		}
+
+		os.Remove(configFilename)
+	}()
+
+	// Run the TopoServerTestSuite tests.
+	testIndex := 0
+	test.TopoServerTestSuite(t, func() *topo.Server {
+		// Each test will use its own sub-directories.
+		testRoot := fmt.Sprintf("test-%v", testIndex)
+		testIndex++
+
+		// Create the server on the new root.
+		ts, err := topo.OpenServer("consul", serverAddr, path.Join(testRoot, topo.GlobalCell))
+		if err != nil {
+			t.Fatalf("OpenServer() failed: %v", err)
+		}
+
+		// Create the CellInfo.
+		if err := ts.CreateCellInfo(context.Background(), test.LocalCellName, &topodatapb.CellInfo{
+			ServerAddress: serverAddr,
+			Root:          path.Join(testRoot, test.LocalCellName),
+		}); err != nil {
+			t.Fatalf("CreateCellInfo() failed: %v", err)
+		}
+
+		return ts
+	})
+}
+
 func TestConsulTopoWithAuth(t *testing.T) {
 	// One test is going to wait that full period, so make it shorter.
 	*watchPollDuration = 100 * time.Millisecond
diff --git a/go/vt/vtgate/evalengine/arithmetic_test.go b/go/vt/vtgate/evalengine/arithmetic_test.go
index 5575196baa..eb37012252 100644
--- a/go/vt/vtgate/evalengine/arithmetic_test.go
+++ b/go/vt/vtgate/evalengine/arithmetic_test.go
@@ -567,21 +567,6 @@ func TestNullsafeCompare(t *testing.T) {
 		v1:  TestValue(querypb.Type_DATETIME, "1000-01-01 00:00:00"),
 		v2:  TestValue(querypb.Type_BINARY, "2000-01-01 00:00:00"),
 		out: -1,
-	}, {
-		// Date/Time types
-		v1:  TestValue(querypb.Type_BIT, "101"),
-		v2:  TestValue(querypb.Type_BIT, "101"),
-		out: 0,
-	}, {
-		// Date/Time types
-		v1:  TestValue(querypb.Type_BIT, "1"),
-		v2:  TestValue(querypb.Type_BIT, "0"),
-		out: 1,
-	}, {
-		// Date/Time types
-		v1:  TestValue(querypb.Type_BIT, "0"),
-		v2:  TestValue(querypb.Type_BIT, "1"),
-		out: -1,
 	}}
 	for _, tcase := range tcases {
 		got, err := NullsafeCompare(tcase.v1, tcase.v2)
diff --git a/go/vt/vtgate/vindexes/consistent_lookup_test.go b/go/vt/vtgate/vindexes/consistent_lookup_test.go
index b4df8cd2c1..268b89b368 100644
--- a/go/vt/vtgate/vindexes/consistent_lookup_test.go
+++ b/go/vt/vtgate/vindexes/consistent_lookup_test.go
@@ -451,6 +451,7 @@ func TestConsistentLookupUpdateBecauseUncomparableTypes(t *testing.T) {
 		{querypb.Type_TEXT, "some string"},
 		{querypb.Type_VARCHAR, "some string"},
 		{querypb.Type_CHAR, "some string"},
+		{querypb.Type_BIT, "some string"},
 		{querypb.Type_GEOMETRY, "some string"},
 	}