Add a way to know if DemotePrimary is blocked and send it in the health stream

go/vt/vttablet/tabletmanager/rpc_replication.go

@@ -19,6 +19,7 @@ package tabletmanager
 import (
 	"context"
 	"fmt"
+	"runtime"
 	"strings"
 	"time"
 
@@ -29,6 +30,7 @@ import (
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/mysqlctl"
 	"vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/topo/topoproto"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver"
@@ -520,6 +522,23 @@ func (tm *TabletManager) demotePrimary(ctx context.Context, revertPartialFailure
 	}
 	defer tm.unlock()
 
+	finishCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		select {
+		case <-finishCtx.Done():
+		// Finished running DemotePrimary. Nothing to do.
+		case <-time.After(10 * topo.RemoteOperationTimeout):
+			// We waited for over 10 times of remote operation timeout, but DemotePrimary is still not done.
+			// Collect more information and signal demote primary is indefinitely stuck.
+			log.Errorf("DemotePrimary seems to be blocked. Collecting more information.")
+			tm.QueryServiceControl.SetDemotePrimaryBlocked()
+			buf := make([]byte, 1<<16) // 64 KB buffer size
+			stackSize := runtime.Stack(buf, true)
+			log.Errorf("Stack trace:\n%s", string(buf[:stackSize]))
+		}
+	}()
+
 	tablet := tm.Tablet()
 	wasPrimary := tablet.Type == topodatapb.TabletType_PRIMARY
 	wasServing := tm.QueryServiceControl.IsServing()

go/vt/vttablet/tabletserver/controller.go

@@ -119,6 +119,9 @@ type Controller interface {
 
 	// WaitForPreparedTwoPCTransactions waits for all prepared transactions to be resolved.
 	WaitForPreparedTwoPCTransactions(ctx context.Context) error
+
+	// SetDemotePrimaryBlocked marks that demote primary is blocked in the state manager.
+	SetDemotePrimaryBlocked()
 }
 
 // Ensure TabletServer satisfies Controller interface.

go/vt/vttablet/tabletserver/state_manager.go

@@ -87,18 +87,19 @@ type stateManager struct {
 	//
 	// If a transition fails, we set retrying to true and launch
 	// retryTransition which loops until the state converges.
-	mu             sync.Mutex
-	wantState      servingState
-	wantTabletType topodatapb.TabletType
-	state          servingState
-	target         *querypb.Target
-	ptsTimestamp   time.Time
-	retrying       bool
-	replHealthy    bool
-	lameduck       bool
-	alsoAllow      []topodatapb.TabletType
-	reason         string
-	transitionErr  error
+	mu                   sync.Mutex
+	wantState            servingState
+	wantTabletType       topodatapb.TabletType
+	state                servingState
+	target               *querypb.Target
+	ptsTimestamp         time.Time
+	retrying             bool
+	replHealthy          bool
+	demotePrimaryBlocked bool
+	lameduck             bool
+	alsoAllow            []topodatapb.TabletType
+	reason               string
+	transitionErr        error
 
 	rw *requestsWaiter
 
@@ -715,6 +716,10 @@ func (sm *stateManager) Broadcast() {
 	defer sm.mu.Unlock()
 
 	lag, err := sm.refreshReplHealthLocked()
+	if sm.demotePrimaryBlocked {
+		// If we are blocked from demoting primary, we should send an error for it.
+		err = vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "Demoting primary is blocked")
+	}
 	sm.hs.ChangeState(sm.target.TabletType, sm.ptsTimestamp, lag, err, sm.isServingLocked())
 }
 

go/vt/vttablet/tabletserver/state_manager_test.go

@@ -669,6 +669,42 @@ func TestStateManagerNotify(t *testing.T) {
 	sm.StopService()
 }
 
+func TestDemotePrimaryBlocked(t *testing.T) {
+	sm := newTestStateManager()
+	defer sm.StopService()
+	err := sm.SetServingType(topodatapb.TabletType_PRIMARY, testNow, StateServing, "")
+	require.NoError(t, err)
+	// Stopping the ticker so that we don't get unexpected health streams.
+	sm.hcticks.Stop()
+
+	ch := make(chan *querypb.StreamHealthResponse, 5)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := sm.hs.Stream(context.Background(), func(shr *querypb.StreamHealthResponse) error {
+			ch <- shr
+			return nil
+		})
+		assert.Contains(t, err.Error(), "tabletserver is shutdown")
+	}()
+	defer wg.Wait()
+
+	// Send a broadcast message and check we have no error there.
+	sm.Broadcast()
+	gotshr := <-ch
+	require.Empty(t, gotshr.RealtimeStats.HealthError)
+
+	// If demote primary is blocked, then we should get an error.
+	sm.demotePrimaryBlocked = true
+	sm.Broadcast()
+	gotshr = <-ch
+	require.EqualValues(t, "Demoting primary is blocked", gotshr.RealtimeStats.HealthError)
+
+	// Stop the state manager.
+	sm.StopService()
+}
+
 func TestRefreshReplHealthLocked(t *testing.T) {
 	sm := newTestStateManager()
 	defer sm.StopService()

go/vt/vttablet/tabletserver/tabletserver.go

@@ -758,6 +758,11 @@ func (tsv *TabletServer) WaitForPreparedTwoPCTransactions(ctx context.Context) e
 	}
 }
 
+// SetDemotePrimaryBlocked marks that demote primary is blocked in the state manager.
+func (tsv *TabletServer) SetDemotePrimaryBlocked() {
+	tsv.sm.demotePrimaryBlocked = true
+}
+
 // CreateTransaction creates the metadata for a 2PC transaction.
 func (tsv *TabletServer) CreateTransaction(ctx context.Context, target *querypb.Target, dtid string, participants []*querypb.Target) (err error) {
 	return tsv.execRequest(

go/vt/vttablet/tabletservermock/controller.go

@@ -274,6 +274,11 @@ func (tqsc *Controller) WaitForPreparedTwoPCTransactions(context.Context) error
 	return nil
 }
 
+// SetDemotePrimaryBlocked is part of the tabletserver.Controller interface
+func (tqsc *Controller) SetDemotePrimaryBlocked() {
+	tqsc.MethodCalled["SetDemotePrimaryBlocked"] = true
+}
+
 // EnterLameduck implements tabletserver.Controller.
 func (tqsc *Controller) EnterLameduck() {
 	tqsc.mu.Lock()