Repository: vitessio/vitess
Pull request #6885 (2020-10-14 22:12:56)

tablet streaming health fix: never silently skip health state changes
Today, if vtgate is slow to receive health updates from a tablet the tablet may
simply throw away the most recent status update instead of buffering it.

Since there's no periodic forced state push and discarded state is the most
recent one, this creates a possibility that a gate can permenently and incorrectly
believe that a tablet is not serving. You could have a healthy master tablet and a
gate that doesn't believe it's healthy and the only fix today would be to do an
operation and restart the gate.

The fix in this PR is to cancel the streaming health RPC if the buffer overflows rather
than discarding the most recent state update. Whenever the gate or its network connection
gets unwedged it'll realize that the streaming health RPC is canceled and open up a new
streaming health RPC where it will get a recent, accurate state.

Fixes https://github.com/vitessio/vitess/issues/5445

Signed-off-by: David Weitzman <dweitzman@pinterest.com>


Patch:
diff --git a/Makefile b/Makefile
index 08565bedc1..3b72035cab 100644
--- a/Makefile
+++ b/Makefile
@@ -69,23 +69,14 @@ endif
 	go install $(EXTRA_BUILD_FLAGS) $(VT_GO_PARALLEL) -ldflags "$(shell tools/build_version_flags.sh)" -gcflags -'N -l' ./go/...
 
 # install copies the files needed to run Vitess into the given directory tree.
-# This target is optimized for docker images. It only installs the files needed for running vitess in docker
 # Usage: make install PREFIX=/path/to/install/root
 install: build
-	# binaries
-	mkdir -p "$${PREFIX}/bin"
-	cp "$${VTROOT}/bin/"{mysqlctld,vtorc,vtctld,vtctlclient,vtgate,vttablet,vtworker,vtbackup} "$${PREFIX}/bin/"
-
-# Install local install the binaries needed to run vitess locally
-# Usage: make install-local PREFIX=/path/to/install/root
-install-local: build
 	# binaries
 	mkdir -p "$${PREFIX}/bin"
 	cp "$${VTROOT}/bin/"{mysqlctl,mysqlctld,vtorc,vtctld,vtctlclient,vtgate,vttablet,vtworker,vtbackup} "$${PREFIX}/bin/"
 
-
 # install copies the files needed to run test Vitess using vtcombo into the given directory tree.
-# Usage: make install-testing PREFIX=/path/to/install/root
+# Usage: make install PREFIX=/path/to/install/root
 install-testing: build
 	# binaries
 	mkdir -p "$${PREFIX}/bin"
diff --git a/examples/operator/operator.yaml b/examples/operator/operator.yaml
index d3584772d4..71b930eec3 100644
--- a/examples/operator/operator.yaml
+++ b/examples/operator/operator.yaml
@@ -1164,8 +1164,11 @@ spec:
               properties:
                 engine:
                   description: 'Engine specifies the Vitess backup engine to use,
-                    either "builtin" or "xtrabackup". Restores will always be done
-                    with whichever engine created a given backup.
+                    either "builtin" or "xtrabackup". Note that if you change this
+                    after a Vitess cluster is already deployed, you must roll the
+                    change out to all tablets and then take a new backup from one
+                    tablet in each shard. Otherwise, new tablets trying to restore
+                    will find that the latest backup was created with the wrong engine.
                     Default: builtin'
                   enum:
                   - builtin
diff --git a/go/vt/key/destination.go b/go/vt/key/destination.go
index 235f561c2d..6694650497 100644
--- a/go/vt/key/destination.go
+++ b/go/vt/key/destination.go
@@ -20,7 +20,6 @@ import (
 	"bytes"
 	"encoding/hex"
 	"math/rand"
-	"sort"
 	"strings"
 
 	"vitess.io/vitess/go/vt/vterrors"
@@ -150,10 +149,6 @@ func (d DestinationExactKeyRange) String() string {
 }
 
 func processExactKeyRange(allShards []*topodatapb.ShardReference, kr *topodatapb.KeyRange, addShard func(shard string) error) error {
-	sort.SliceStable(allShards, func(i, j int) bool {
-		return KeyRangeStartSmaller(allShards[i].GetKeyRange(), allShards[j].GetKeyRange())
-	})
-
 	shardnum := 0
 	for shardnum < len(allShards) {
 		if KeyRangeStartEqual(kr, allShards[shardnum].KeyRange) {
diff --git a/go/vt/key/key.go b/go/vt/key/key.go
index 98143a1ff1..0bd7f137a3 100644
--- a/go/vt/key/key.go
+++ b/go/vt/key/key.go
@@ -185,17 +185,6 @@ func KeyRangeEqual(left, right *topodatapb.KeyRange) bool {
 		bytes.Equal(left.End, right.End)
 }
 
-// KeyRangeStartEqual returns true if right's keyrange start is _after_ left's start
-func KeyRangeStartSmaller(left, right *topodatapb.KeyRange) bool {
-	if left == nil {
-		return right != nil
-	}
-	if right == nil {
-		return false
-	}
-	return bytes.Compare(left.Start, right.Start) < 0
-}
-
 // KeyRangeStartEqual returns true if both key ranges have the same start
 func KeyRangeStartEqual(left, right *topodatapb.KeyRange) bool {
 	if left == nil {
diff --git a/go/vt/mysqlctl/azblobbackupstorage/azblob.go b/go/vt/mysqlctl/azblobbackupstorage/azblob.go
index cb77ef6153..5ec3827c77 100644
--- a/go/vt/mysqlctl/azblobbackupstorage/azblob.go
+++ b/go/vt/mysqlctl/azblobbackupstorage/azblob.go
@@ -276,20 +276,15 @@ func (bs *AZBlobBackupStorage) containerURL() (*azblob.ContainerURL, error) {
 
 // ListBackups implements BackupStorage.
 func (bs *AZBlobBackupStorage) ListBackups(ctx context.Context, dir string) ([]backupstorage.BackupHandle, error) {
-	var searchPrefix string
-	if dir == "/" {
-		searchPrefix = "/"
-	} else {
-		searchPrefix = objName(dir, "")
-	}
-
-	log.Infof("ListBackups: [azblob] container: %s, directory: %v", *containerName, searchPrefix)
+	log.Infof("ListBackups: [azblob] container: %s, directory: %v", *containerName, objName(dir, ""))
 
 	containerURL, err := bs.containerURL()
 	if err != nil {
 		return nil, err
 	}
 
+	searchPrefix := objName(dir, "")
+
 	result := make([]backupstorage.BackupHandle, 0)
 	var subdirs []string
 
diff --git a/go/vt/mysqlctl/gcsbackupstorage/gcs.go b/go/vt/mysqlctl/gcsbackupstorage/gcs.go
index 1a7266a71a..7ca9f4796d 100644
--- a/go/vt/mysqlctl/gcsbackupstorage/gcs.go
+++ b/go/vt/mysqlctl/gcsbackupstorage/gcs.go
@@ -132,14 +132,7 @@ func (bs *GCSBackupStorage) ListBackups(ctx context.Context, dir string) ([]back
 
 	// List prefixes that begin with dir (i.e. list subdirs).
 	var subdirs []string
-
-	var searchPrefix string
-	if dir == "/" {
-		searchPrefix = ""
-	} else {
-		searchPrefix = objName(dir, "" /* include trailing slash */)
-	}
-
+	searchPrefix := objName(dir, "" /* include trailing slash */)
 	query := &storage.Query{
 		Delimiter: "/",
 		Prefix:    searchPrefix,
diff --git a/go/vt/vttablet/tabletserver/health_streamer.go b/go/vt/vttablet/tabletserver/health_streamer.go
index 85393c76ad..4eaf80fddc 100644
--- a/go/vt/vttablet/tabletserver/health_streamer.go
+++ b/go/vt/vttablet/tabletserver/health_streamer.go
@@ -17,6 +17,7 @@ limitations under the License.
 package tabletserver
 
 import (
+	"flag"
 	"fmt"
 	"io"
 	"sync"
@@ -25,6 +26,7 @@ import (
 	"github.com/golang/protobuf/proto"
 	"golang.org/x/net/context"
 	"vitess.io/vitess/go/history"
+	"vitess.io/vitess/go/vt/log"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
@@ -39,6 +41,8 @@ var (
 	blpFunc = vreplication.StatusSummary
 
 	errUnintialized = "tabletserver uninitialized"
+
+	streamHealthBufferSize = flag.Uint("stream_health_buffer_size", 20, "max streaming health entries to buffer per streaming health client")
 )
 
 // healthStreamer streams health information to callers.
@@ -115,7 +119,10 @@ func (hs *healthStreamer) Stream(ctx context.Context, callback func(*querypb.Str
 			return nil
 		case <-hsCtx.Done():
 			return vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "tabletserver is shutdown")
-		case shr := <-ch:
+		case shr, ok := <-ch:
+			if !ok {
+				return vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "stream health buffer overflowed. client should reconnect for up-to-date status")
+			}
 			if err := callback(shr); err != nil {
 				if err == io.EOF {
 					return nil
@@ -134,7 +141,7 @@ func (hs *healthStreamer) register() (chan *querypb.StreamHealthResponse, contex
 		return nil, nil
 	}
 
-	ch := make(chan *querypb.StreamHealthResponse, 1)
+	ch := make(chan *querypb.StreamHealthResponse, *streamHealthBufferSize)
 	hs.clients[ch] = struct{}{}
 
 	// Send the current state immediately.
@@ -176,6 +183,21 @@ func (hs *healthStreamer) ChangeState(tabletType topodatapb.TabletType, terTimes
 		select {
 		case ch <- shr:
 		default:
+			// We can't block this state change on broadcasting to a streaming health client, but we
+			// also don't want to silently fail to inform a streaming health client of a state change
+			// because it can allow a vtgate to get wedged in a state where it's wrong about whether
+			// a tablet is healthy and can't automatically recover (see
+			//  https://github.com/vitessio/vitess/issues/5445). If we can't send a health update
+			// to this client we'll close() the channel which will ultimate fail the streaming health
+			// RPC and cause vtgates to reconnect.
+			//
+			// An alternative approach for streaming health would be to force a periodic broadcast even
+			// when there hasn't been an update and/or move away from using channels toward a model where
+			// old updates can be purged from the buffer in favor of more recent updates (since only the
+			// most recent health state really matters to gates).
+			log.Warning("A streaming health buffer is full. Closing the channel")
+			close(ch)
+			delete(hs.clients, ch)
 		}
 	}
 	hs.history.Add(&historyRecord{
@@ -187,7 +209,7 @@ func (hs *healthStreamer) ChangeState(tabletType topodatapb.TabletType, terTimes
 	})
 }
 
-func (hs *healthStreamer) ApppendDetails(details []*kv) []*kv {
+func (hs *healthStreamer) AppendDetails(details []*kv) []*kv {
 	hs.mu.Lock()
 	defer hs.mu.Unlock()
 	if hs.state.Target.TabletType == topodatapb.TabletType_MASTER {
diff --git a/go/vt/vttablet/tabletserver/state_manager.go b/go/vt/vttablet/tabletserver/state_manager.go
index 4a43be11ef..272340217e 100644
--- a/go/vt/vttablet/tabletserver/state_manager.go
+++ b/go/vt/vttablet/tabletserver/state_manager.go
@@ -654,7 +654,7 @@ func (sm *stateManager) isServingLocked() bool {
 	return sm.state == StateServing && sm.wantState == StateServing && sm.replHealthy && !sm.lameduck
 }
 
-func (sm *stateManager) ApppendDetails(details []*kv) []*kv {
+func (sm *stateManager) AppendDetails(details []*kv) []*kv {
 	sm.mu.Lock()
 	defer sm.mu.Unlock()
 
diff --git a/go/vt/vttablet/tabletserver/status.go b/go/vt/vttablet/tabletserver/status.go
index d8d1244f0f..ca95a2c2c4 100644
--- a/go/vt/vttablet/tabletserver/status.go
+++ b/go/vt/vttablet/tabletserver/status.go
@@ -246,8 +246,8 @@ func (tsv *TabletServer) AddStatusPart() {
 		} else {
 			status.Latest = &historyRecord{}
 		}
-		status.Details = tsv.sm.ApppendDetails(nil)
-		status.Details = tsv.hs.ApppendDetails(status.Details)
+		status.Details = tsv.sm.AppendDetails(nil)
+		status.Details = tsv.hs.AppendDetails(status.Details)
 		rates := tsv.stats.QPSRates.Get()
 		if qps, ok := rates["All"]; ok && len(qps) > 0 {
 			status.CurrentQPS = qps[0]
@@ -257,8 +257,8 @@ func (tsv *TabletServer) AddStatusPart() {
 
 	tsv.exporter.HandleFunc("/debug/status_details", func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "application/json; charset=utf-8")
-		details := tsv.sm.ApppendDetails(nil)
-		details = tsv.hs.ApppendDetails(details)
+		details := tsv.sm.AppendDetails(nil)
+		details = tsv.hs.AppendDetails(details)
 		b, err := json.MarshalIndent(details, "", " ")
 		if err != nil {
 			w.Write([]byte(err.Error()))


Test patch:
diff --git a/go/mysql/auth_server_static_test.go b/go/mysql/auth_server_static_test.go
index f195b4c72d..2586da2f60 100644
--- a/go/mysql/auth_server_static_test.go
+++ b/go/mysql/auth_server_static_test.go
@@ -207,6 +207,10 @@ func hupTestWithRotation(t *testing.T, aStatic *AuthServerStatic, tmpFile *os.Fi
 		t.Fatalf("couldn't overwrite temp file: %v", err)
 	}
 
+	if aStatic.getEntries()[oldStr][0].Password != oldStr {
+		t.Fatalf("%s's Password should still be '%s'", oldStr, oldStr)
+	}
+
 	time.Sleep(20 * time.Millisecond) // wait for signal handler
 
 	if aStatic.getEntries()[oldStr] != nil {
diff --git a/go/test/endtoend/messaging/message_test.go b/go/test/endtoend/messaging/message_test.go
index b93b9c7c5e..e5ada917cb 100644
--- a/go/test/endtoend/messaging/message_test.go
+++ b/go/test/endtoend/messaging/message_test.go
@@ -331,7 +331,6 @@ func TestConnection(t *testing.T) {
 	_, err = stream.MessageStream(userKeyspace, "", nil, name)
 	require.Nil(t, err)
 	// validate client count of vttablet
-	time.Sleep(time.Second)
 	assert.Equal(t, 1, getClientCount(shard0Master))
 	assert.Equal(t, 1, getClientCount(shard1Master))
 	// second connection with vtgate, secont connection
@@ -341,7 +340,6 @@ func TestConnection(t *testing.T) {
 	_, err = stream1.MessageStream(userKeyspace, "", nil, name)
 	require.Nil(t, err)
 	// validate client count of vttablet
-	time.Sleep(time.Second)
 	assert.Equal(t, 2, getClientCount(shard0Master))
 	assert.Equal(t, 2, getClientCount(shard1Master))