Store the health check subscriber name to improve error message

go/vt/discovery/fake_healthcheck.go

@@ -109,7 +109,7 @@ func (fhc *FakeHealthCheck) GetTabletHealth(kst KeyspaceShardTabletType, alias *
 }
 
 // Subscribe returns the channel in the struct. Subscribe should only be called in one place for this fake health check
-func (fhc *FakeHealthCheck) Subscribe() chan *TabletHealth {
+func (fhc *FakeHealthCheck) Subscribe(string) chan *TabletHealth {
 	return fhc.ch
 }
 

go/vt/discovery/healthcheck.go

@@ -39,6 +39,7 @@ import (
 	"fmt"
 	"hash/crc32"
 	"net/http"
+	"runtime"
 	"sort"
 	"strings"
 	"sync"
@@ -241,7 +242,7 @@ type HealthCheck interface {
 	GetTabletHealthByAlias(alias *topodata.TabletAlias) (*TabletHealth, error)
 
 	// Subscribe adds a listener. Used by vtgate buffer to learn about primary changes.
-	Subscribe() chan *TabletHealth
+	Subscribe(name string) chan *TabletHealth
 
 	// Unsubscribe removes a listener.
 	Unsubscribe(c chan *TabletHealth)
@@ -296,7 +297,7 @@ type HealthCheckImpl struct {
 	// mutex to protect subscribers
 	subMu sync.Mutex
 	// subscribers
-	subscribers map[chan *TabletHealth]struct{}
+	subscribers map[chan *TabletHealth]string
 	// loadTablets trigger is used to immediately load a new primary tablet when the current one has been demoted
 	loadTabletsTrigger chan struct{}
 }
@@ -361,7 +362,7 @@ func NewHealthCheck(ctx context.Context, retryDelay, healthCheckTimeout time.Dur
 		healthByAlias:      make(map[tabletAliasString]*tabletHealthCheck),
 		healthData:         make(map[KeyspaceShardTabletType]map[tabletAliasString]*TabletHealth),
 		healthy:            make(map[KeyspaceShardTabletType][]*TabletHealth),
-		subscribers:        make(map[chan *TabletHealth]struct{}),
+		subscribers:        make(map[chan *TabletHealth]string),
 		cellAliases:        make(map[string]string),
 		loadTabletsTrigger: make(chan struct{}, 1),
 	}
@@ -632,11 +633,11 @@ func (hc *HealthCheckImpl) recomputeHealthy(key KeyspaceShardTabletType) {
 }
 
 // Subscribe adds a listener. Used by vtgate buffer to learn about primary changes.
-func (hc *HealthCheckImpl) Subscribe() chan *TabletHealth {
+func (hc *HealthCheckImpl) Subscribe(subscriber string) chan *TabletHealth {
 	hc.subMu.Lock()
 	defer hc.subMu.Unlock()
 	c := make(chan *TabletHealth, 2048)
-	hc.subscribers[c] = struct{}{}
+	hc.subscribers[c] = subscriber
 	return c
 }
 
@@ -647,16 +648,24 @@ func (hc *HealthCheckImpl) Unsubscribe(c chan *TabletHealth) {
 	delete(hc.subscribers, c)
 }
 
+var printStack = sync.OnceFunc(func() {
+	buf := make([]byte, 10240) // Allocate buffer large enough
+	n := runtime.Stack(buf, true)
+	fmt.Printf("All Goroutines Stack Trace:\n%s\n", buf[:n])
+})
+
 func (hc *HealthCheckImpl) broadcast(th *TabletHealth) {
 	hc.subMu.Lock()
 	defer hc.subMu.Unlock()
-	for c := range hc.subscribers {
+	for c, subscriber := range hc.subscribers {
 		select {
 		case c <- th:
 		default:
 			// If the channel is full, we drop the message.
 			hcChannelFullCounter.Add(1)
-			log.Warningf("HealthCheck broadcast channel is full, dropping message for %s", topotools.TabletIdent(th.Tablet))
+			log.Warningf("HealthCheck broadcast channel is full for %v, dropping message for %s", subscriber, topotools.TabletIdent(th.Tablet))
+			// Print the stack trace only once.
+			printStack()
 		}
 	}
 }

go/vt/discovery/healthcheck_test.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -30,6 +31,7 @@ import (
 	"github.com/google/safehtml/template"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/exp/maps"
 
 	"vitess.io/vitess/go/test/utils"
 	"vitess.io/vitess/go/vt/grpcclient"
@@ -150,7 +152,7 @@ func TestHealthCheck(t *testing.T) {
 	conn := createFakeConn(tablet, input)
 
 	// create a channel and subscribe to healthcheck
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	testChecksum(t, 0, hc.stateChecksum())
 	hc.AddTablet(tablet)
 	testChecksum(t, 1027934207, hc.stateChecksum())
@@ -289,7 +291,7 @@ func TestHealthCheckStreamError(t *testing.T) {
 
 	tablet := createTestTablet(0, "cell", "a")
 	input := make(chan *querypb.StreamHealthResponse)
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	fc := createFakeConn(tablet, input)
 	fc.errCh = make(chan error)
 	hc.AddTablet(tablet)
@@ -353,7 +355,7 @@ func TestHealthCheckErrorOnPrimary(t *testing.T) {
 
 	tablet := createTestTablet(0, "cell", "a")
 	input := make(chan *querypb.StreamHealthResponse)
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	fc := createFakeConn(tablet, input)
 	fc.errCh = make(chan error)
 	hc.AddTablet(tablet)
@@ -414,7 +416,7 @@ func TestHealthCheckErrorOnPrimaryAfterExternalReparent(t *testing.T) {
 	hc := createTestHc(ctx, ts)
 	defer hc.Close()
 
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 
 	tablet1 := createTestTablet(0, "cell", "a")
 	input1 := make(chan *querypb.StreamHealthResponse)
@@ -498,7 +500,7 @@ func TestHealthCheckVerifiesTabletAlias(t *testing.T) {
 	tablet := createTestTablet(0, "cell", "a")
 	input := make(chan *querypb.StreamHealthResponse, 1)
 	fc := createFakeConn(tablet, input)
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 
 	hc.AddTablet(tablet)
 
@@ -543,7 +545,7 @@ func TestHealthCheckCloseWaitsForGoRoutines(t *testing.T) {
 	tablet := createTestTablet(0, "cell", "a")
 	input := make(chan *querypb.StreamHealthResponse, 1)
 	createFakeConn(tablet, input)
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 
 	hc.AddTablet(tablet)
 
@@ -610,7 +612,7 @@ func TestHealthCheckTimeout(t *testing.T) {
 	tablet := createTestTablet(0, "cell", "a")
 	input := make(chan *querypb.StreamHealthResponse)
 	fc := createFakeConn(tablet, input)
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	hc.AddTablet(tablet)
 	// Immediately after AddTablet() there will be the first notification.
 	want := &TabletHealth{
@@ -692,7 +694,7 @@ func TestWaitForAllServingTablets(t *testing.T) {
 	createFakeConn(tablet, input)
 
 	// create a channel and subscribe to healthcheck
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	hc.AddTablet(tablet)
 	// there will be a first result, get and discard it
 	<-resultChan
@@ -760,7 +762,7 @@ func TestRemoveTablet(t *testing.T) {
 	createFakeConn(tablet, input)
 
 	// create a channel and subscribe to healthcheck
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	hc.AddTablet(tablet)
 	// there will be a first result, get and discard it
 	<-resultChan
@@ -896,7 +898,7 @@ func TestRemoveTabletDuringExternalReparenting(t *testing.T) {
 	thirdTabletConn := createFakeConn(thirdTablet, thirdTabletHealthStream)
 	thirdTabletConn.errCh = make(chan error)
 
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 
 	hc.AddTablet(firstTablet)
 	<-resultChan
@@ -991,7 +993,7 @@ func TestGetHealthyTablets(t *testing.T) {
 	createFakeConn(tablet, input)
 
 	// create a channel and subscribe to healthcheck
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	hc.AddTablet(tablet)
 	// there will be a first result, get and discard it
 	<-resultChan
@@ -1181,7 +1183,7 @@ func TestPrimaryInOtherCell(t *testing.T) {
 	input := make(chan *querypb.StreamHealthResponse)
 	fc := createFakeConn(tablet, input)
 	// create a channel and subscribe to healthcheck
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	hc.AddTablet(tablet)
 	// should get a result, but this will hang if multi-cell logic is broken
 	// so wait and timeout
@@ -1241,7 +1243,7 @@ func TestReplicaInOtherCell(t *testing.T) {
 	input := make(chan *querypb.StreamHealthResponse)
 	fc := createFakeConn(local, input)
 	// create a channel and subscribe to healthcheck
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	hc.AddTablet(local)
 
 	ticker := time.NewTicker(1 * time.Second)
@@ -1286,7 +1288,7 @@ func TestReplicaInOtherCell(t *testing.T) {
 	input2 := make(chan *querypb.StreamHealthResponse)
 	fc2 := createFakeConn(remote, input2)
 	// create a channel and subscribe to healthcheck
-	resultChan2 := hc.Subscribe()
+	resultChan2 := hc.Subscribe("")
 	hc.AddTablet(remote)
 	// should get a result, but this will hang if multi-cell logic is broken
 	// so wait and timeout
@@ -1352,7 +1354,7 @@ func TestCellAliases(t *testing.T) {
 	input := make(chan *querypb.StreamHealthResponse)
 	fc := createFakeConn(tablet, input)
 	// create a channel and subscribe to healthcheck
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 	hc.AddTablet(tablet)
 	// should get a result, but this will hang if cell alias logic is broken
 	// so wait and timeout
@@ -1405,7 +1407,7 @@ func TestHealthCheckChecksGrpcPort(t *testing.T) {
 
 	tablet := createTestTablet(0, "cell", "a")
 	tablet.PortMap["grpc"] = 0
-	resultChan := hc.Subscribe()
+	resultChan := hc.Subscribe("")
 
 	// AddTablet should not add the tablet because port is 0
 	hc.AddTablet(tablet)
@@ -1446,6 +1448,33 @@ func TestTemplate(t *testing.T) {
 	require.Nil(t, err, "error executing template: %v", err)
 }
 
+// TestHealthCheckImplSubscriberName tests that we have the subscirber name in the healthcheck.
+func TestHealthCheckImplSubscriberName(t *testing.T) {
+	ctx := utils.LeakCheckContext(t)
+
+	hc := NewHealthCheck(ctx, 1*time.Millisecond, time.Hour, nil, "", "", nil)
+	defer hc.Close()
+
+	subsName := "SubscriberName1"
+	subsName2 := "SubscriberName2"
+	ch := hc.Subscribe(subsName)
+	ch2 := hc.Subscribe(subsName2)
+
+	subsNames := maps.Values(hc.subscribers)
+	slices.Sort(subsNames)
+	require.Equal(t, 2, len(subsNames), "expected 2 subscribers")
+	require.EqualValues(t, []string{subsName, subsName2}, subsNames, "unexpected subscribers")
+
+	hc.Unsubscribe(ch)
+	subsNames = maps.Values(hc.subscribers)
+	require.Equal(t, 1, len(subsNames), "expected 1 subscriber")
+	require.EqualValues(t, []string{subsName2}, subsNames, "unexpected subscribers")
+
+	hc.Unsubscribe(ch2)
+	subsNames = maps.Values(hc.subscribers)
+	require.Empty(t, subsNames, "expected no subscribers")
+}
+
 func TestDebugURLFormatting(t *testing.T) {
 	defer utils.EnsureNoLeaks(t)
 	TabletURLTemplateString = "https://{{.GetHostNameLevel 0}}.bastion.{{.Tablet.Alias.Cell}}.corp"
@@ -1490,7 +1519,7 @@ func TestConcurrentUpdates(t *testing.T) {
 
 	// Subscribe to the healthcheck
 	// Make the receiver keep track of the updates received.
-	ch := hc.Subscribe()
+	ch := hc.Subscribe("")
 	var totalCount atomic.Int32
 	go func() {
 		for range ch {

go/vt/discovery/keyspace_events.go

@@ -221,7 +221,7 @@ func (kew *KeyspaceEventWatcher) broadcast(ev *KeyspaceEvent) {
 }
 
 func (kew *KeyspaceEventWatcher) run(ctx context.Context) {
-	hcChan := kew.hc.Subscribe()
+	hcChan := kew.hc.Subscribe("KeyspaceEventWatcher")
 	bufferCtx, bufferCancel := context.WithCancel(ctx)
 
 	go func() {

go/vt/throttler/demo/throttler_demo.go

@@ -246,7 +246,7 @@ func newClient(ctx context.Context, primary *primary, replica *replica, ts *topo
 		throttler:   t,
 		stopChan:    make(chan struct{}),
 	}
-	healthcheckCh := c.healthCheck.Subscribe()
+	healthcheckCh := c.healthCheck.Subscribe("ThrottlerDemo")
 	c.healthcheckCh = healthcheckCh
 	c.healthCheck.AddTablet(replica.fakeTablet.Tablet)
 	return c

go/vt/vtgate/vtgate.go

@@ -319,7 +319,7 @@ func Init(
 	// ScatterConn depends on TxConn to perform forced rollbacks.
 	sc := NewScatterConn("VttabletCall", tc, gw)
 	// TxResolver depends on TxConn to complete distributed transaction.
-	tr := txresolver.NewTxResolver(gw.hc.Subscribe(), tc)
+	tr := txresolver.NewTxResolver(gw.hc.Subscribe("TxResolver"), tc)
 	srvResolver := srvtopo.NewResolver(serv, gw, cell)
 	resolver := NewResolver(srvResolver, serv, cell, sc)
 	vsm := newVStreamManager(srvResolver, serv, cell)
@@ -345,7 +345,7 @@ func Init(
 	var si SchemaInfo // default nil
 	var st *vtschema.Tracker
 	if enableSchemaChangeSignal {
-		st = vtschema.NewTracker(gw.hc.Subscribe(), enableViews, enableUdfs, env.Parser())
+		st = vtschema.NewTracker(gw.hc.Subscribe("SchemaTracker"), enableViews, enableUdfs, env.Parser())
 		addKeyspacesToTracker(ctx, srvResolver, st, gw)
 		si = st
 	}