storeliveness: smear storeliveness heartbeats via coordinator

Previously, heartbeat messages were sent immediately when enqueued via
`EnqueueMessage`. In large clusters, many stores might all send heartbeats
simultaneously at the same tick interval, causing a spike in runnable
goroutines that caused issues elsewhere in the database process.

This patch introduces a heartbeat coordinator that batches and smears
heartbeat sends over a configurable duration using a taskpacer. The
coordinator is enabled by default via the
`kv.store_liveness.use_heartbeat_coordinator` cluster setting (defaults to
true), and can be configured via
`kv.store_liveness.heartbeat_coordinator.refresh` (default: 10ms) and
`kv.store_liveness.heartbeat_coordinator.smear` (default: 1ms) settings.

When enabled, messages are enqueued but not sent immediately. Instead,
they wait in per-node queues until `SendAllEnqueuedMessages()` is called,
which signals the coordinator. The coordinator waits briefly
(`batchDuration`) to collect messages from multiple stores, then paces the
signaling to each queue's `processQueue` goroutine using the pacer to
spread sends over time.

Fixes: #148210
Release note: None

Author: dodeca12

pkg/kv/kvserver/store_test.go

@@ -248,7 +248,7 @@ func createTestStoreWithoutStart(
 		supportGracePeriod := rpcContext.StoreLivenessWithdrawalGracePeriod()
 		options := storeliveness.NewOptions(livenessInterval, heartbeatInterval, supportGracePeriod)
 		transport, err := storeliveness.NewTransport(
-			cfg.AmbientCtx, stopper, cfg.Clock, cfg.NodeDialer, grpcServer, drpcServer, nil, /* knobs */
+			cfg.AmbientCtx, stopper, cfg.Clock, cfg.NodeDialer, grpcServer, drpcServer, cfg.Settings, nil, /* knobs */
 		)
 		require.NoError(t, err)
 		knobs := cfg.TestingKnobs.StoreLivenessKnobs

pkg/kv/kvserver/storeliveness/BUILD.bazel

@@ -34,6 +34,7 @@ go_library(
         "//pkg/util/protoutil",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
+        "//pkg/util/taskpacer",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",

pkg/kv/kvserver/storeliveness/support_manager.go

@@ -31,10 +31,43 @@ var Enabled = settings.RegisterBoolSetting(
 	true,
 )
 
+// TODO(dodeca12): Currently this complexity allows the fallback to immediate
+// heartbeat sends. Once the smearing has battle-tested, remove this and
+// default to using the smeared heartbeat sends approach (no more fallback).
+var UseHeartbeatCoordinatorEnabled = settings.RegisterBoolSetting(
+	settings.SystemOnly,
+	"kv.store_liveness.heartbeat_coordinator.enabled",
+	"if enabled, heartbeat sends are the responsibility of the heartbeat "+
+		"coordinator. Otherwise, heartbeat sends are sent when they are enqueued "+
+		"at the sendQueue, bypassing the heartbeat coordinator",
+
+	true,
+)
+
+var HeartbeatCoordinatorRefresh = settings.RegisterDurationSetting(
+	settings.SystemOnly,
+	"kv.store_liveness.heartbeat_coordinator.refresh",
+	"the total time window within which all queued heartbeat messages should be "+
+		"sent; the pacer distributes the work across this duration to complete "+
+		"by the deadline",
+	10*time.Millisecond,
+	settings.PositiveDuration,
+)
+
+var HeartbeatCoordinatorSmear = settings.RegisterDurationSetting(
+	settings.SystemOnly,
+	"kv.store_liveness.heartbeat_coordinator.smear",
+	"the interval between sending successive batches of heartbeats across all "+
+		"send queues",
+	1*time.Millisecond,
+	settings.PositiveDuration,
+)
+
 // MessageSender is the interface that defines how Store Liveness messages are
 // sent. Transport is the production implementation of MessageSender.
 type MessageSender interface {
 	EnqueueMessage(ctx context.Context, msg slpb.Message) (sent bool)
+	SendAllEnqueuedMessages(ctx context.Context)
 }
 
 // SupportManager orchestrates requesting and providing Store Liveness support.
@@ -168,6 +201,14 @@ func (sm *SupportManager) SupportFromEnabled(ctx context.Context) bool {
 	return Enabled.Get(&sm.settings.SV)
 }
 
+// SendViaHeartbeatCoordinator returns true if the cluster setting to send heartbeats
+// via the heartbeat coordinator (the goroutine in `transport.go`) is enabled.
+// The coordinator is responsible for smearing the heartbeats over
+// a certain duration, so expect heartbeats sends to be paced when enabled.
+func (sm *SupportManager) SendViaHeartbeatCoordinator(ctx context.Context) bool {
+	return UseHeartbeatCoordinatorEnabled.Get(&sm.settings.SV)
+}
+
 // Start starts the main processing goroutine in startLoop as an async task.
 func (sm *SupportManager) Start(ctx context.Context) error {
 	// onRestart is called synchronously before the start of the main loop in
@@ -329,6 +370,9 @@ func (sm *SupportManager) sendHeartbeats(ctx context.Context) {
 			log.KvExec.Warningf(ctx, "failed to send heartbeat to store %+v", msg.To)
 		}
 	}
+	if sm.SendViaHeartbeatCoordinator(ctx) {
+		sm.sender.SendAllEnqueuedMessages(ctx)
+	}
 	sm.metrics.HeartbeatSuccesses.Inc(int64(successes))
 	sm.metrics.HeartbeatFailures.Inc(int64(len(heartbeats) - successes))
 	log.KvExec.VInfof(ctx, 2, "sent heartbeats to %d stores", successes)
@@ -426,6 +470,9 @@ func (sm *SupportManager) handleMessages(ctx context.Context, msgs []*slpb.Messa
 	for _, response := range responses {
 		_ = sm.sender.EnqueueMessage(ctx, response)
 	}
+	if sm.SendViaHeartbeatCoordinator(ctx) {
+		sm.sender.SendAllEnqueuedMessages(ctx)
+	}
 	log.KvExec.VInfof(ctx, 2, "sent %d heartbeat responses", len(responses))
 }
 

pkg/kv/kvserver/storeliveness/testutils.go

@@ -101,6 +101,10 @@ func (tms *testMessageSender) EnqueueMessage(_ context.Context, msg slpb.Message
 	return true
 }
 
+func (tms *testMessageSender) SendAllEnqueuedMessages(_ context.Context) {
+	// No-op for testing
+}
+
 func (tms *testMessageSender) drainSentMessages() []slpb.Message {
 	tms.mu.Lock()
 	defer tms.mu.Unlock()