Merge pull request #152456 from iskettaneh/backport25.2-144187

release-25.2: gossip: batch both client and server gossip messages

Author: iskettaneh

docs/generated/metrics/metrics.html

@@ -170,6 +170,8 @@
 <tr><td>STORAGE</td><td>gossip.connections.refused</td><td>Number of refused incoming gossip connections</td><td>Connections</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>gossip.infos.received</td><td>Number of received gossip Info objects</td><td>Infos</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>gossip.infos.sent</td><td>Number of sent gossip Info objects</td><td>Infos</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>gossip.messages.received</td><td>Number of received gossip messages</td><td>Messages</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>gossip.messages.sent</td><td>Number of sent gossip messages</td><td>Messages</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>intentage</td><td>Cumulative age of locks</td><td>Age</td><td>GAUGE</td><td>SECONDS</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>intentbytes</td><td>Number of bytes in intent KV pairs</td><td>Storage</td><td>GAUGE</td><td>BYTES</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>STORAGE</td><td>intentcount</td><td>Count of intent keys</td><td>Keys</td><td>GAUGE</td><td>COUNT</td><td>AVG</td><td>NONE</td></tr>

pkg/gossip/client.go

@@ -219,8 +219,10 @@ func (c *client) sendGossip(g *Gossip, stream Gossip_GossipClient, firstReq bool
 		infosSent := int64(len(delta))
 		c.clientMetrics.BytesSent.Inc(bytesSent)
 		c.clientMetrics.InfosSent.Inc(infosSent)
+		c.clientMetrics.MessagesSent.Inc(1)
 		c.nodeMetrics.BytesSent.Inc(bytesSent)
 		c.nodeMetrics.InfosSent.Inc(infosSent)
+		c.nodeMetrics.MessagesSent.Inc(1)
 
 		if log.V(1) {
 			ctx := c.AnnotateCtx(stream.Context())
@@ -247,8 +249,10 @@ func (c *client) handleResponse(ctx context.Context, g *Gossip, reply *Response)
 	infosReceived := int64(len(reply.Delta))
 	c.clientMetrics.BytesReceived.Inc(bytesReceived)
 	c.clientMetrics.InfosReceived.Inc(infosReceived)
+	c.clientMetrics.MessagesReceived.Inc(1)
 	c.nodeMetrics.BytesReceived.Inc(bytesReceived)
 	c.nodeMetrics.InfosReceived.Inc(infosReceived)
+	c.nodeMetrics.MessagesReceived.Inc(1)
 
 	// Combine remote node's infostore delta with ours.
 	if reply.Delta != nil {
@@ -326,6 +330,7 @@ func (c *client) gossip(
 
 	errCh := make(chan error, 1)
 	initCh := make(chan struct{}, 1)
+
 	// This wait group is used to allow the caller to wait until gossip
 	// processing is terminated.
 	wg.Add(1)
@@ -390,6 +395,9 @@ func (c *client) gossip(
 		case <-initTimer.C:
 			maybeRegister()
 		case <-sendGossipChan:
+			// We need to send the gossip delta to the remote server. Wait a bit to
+			// batch the updates in one message.
+			batchAndConsume(sendGossipChan, infosBatchDelay)
 			if err := c.sendGossip(g, stream, count == 0); err != nil {
 				return err
 			}

pkg/gossip/gossip.go

@@ -118,6 +118,10 @@ const (
 	// we didn't need to tighten the last time we checked.
 	gossipTightenInterval = time.Second
 
+	// infosBatchDelay controls how much time do we wait to batch infos before
+	// sending them.
+	infosBatchDelay = 10 * time.Millisecond
+
 	unknownNodeID roachpb.NodeID = 0
 )
 
@@ -141,6 +145,18 @@ var (
 		Measurement: "Connections",
 		Unit:        metric.Unit_COUNT,
 	}
+	MetaMessagesSent = metric.Metadata{
+		Name:        "gossip.messages.sent",
+		Help:        "Number of sent gossip messages",
+		Measurement: "Messages",
+		Unit:        metric.Unit_COUNT,
+	}
+	MetaMessagesReceived = metric.Metadata{
+		Name:        "gossip.messages.received",
+		Help:        "Number of received gossip messages",
+		Measurement: "Messages",
+		Unit:        metric.Unit_COUNT,
+	}
 	MetaInfosSent = metric.Metadata{
 		Name:        "gossip.infos.sent",
 		Help:        "Number of sent gossip Info objects",

pkg/gossip/gossip.proto

@@ -76,6 +76,8 @@ message MetricSnap {
   int64 bytes_sent = 3;
   int64 infos_received = 4;
   int64 infos_sent = 5;
+  int64 messages_received = 7;
+  int64 messages_sent = 8;
   int64 conns_refused = 6;
 }
 

pkg/gossip/gossip_test.go

@@ -1058,3 +1058,85 @@ func TestServerSendsHighStampsDiff(t *testing.T) {
 		return nil
 	})
 }
+
+// TestGossipBatching verifies that both server and client gossip updates are
+// batched.
+func TestGossipBatching(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	skip.UnderDeadlock(t, "might be flaky since it relies on some upper-bound timing")
+	skip.UnderRace(t, "might be flaky since it relies on some upper-bound timing")
+
+	stopper := stop.NewStopper()
+	defer stopper.Stop(context.Background())
+
+	// Shared cluster ID by all gossipers
+	clusterID := uuid.MakeV4()
+
+	local, localCtx := startGossip(clusterID, 1, stopper, t, metric.NewRegistry())
+	remote, remoteCtx := startGossip(clusterID, 2, stopper, t, metric.NewRegistry())
+	remote.mu.Lock()
+	rAddr := remote.mu.is.NodeAddr
+	remote.mu.Unlock()
+	local.manage(localCtx)
+	remote.manage(remoteCtx)
+
+	// Start a client connection to the remote node
+	local.mu.Lock()
+	local.startClientLocked(rAddr, roachpb.Locality{}, localCtx)
+	local.mu.Unlock()
+
+	// Wait for connection to be established
+	var c *client
+	testutils.SucceedsSoon(t, func() error {
+		c = local.findClient(func(c *client) bool { return c.addr.String() == rAddr.String() })
+		if c == nil {
+			return fmt.Errorf("client not found")
+		}
+		return nil
+	})
+
+	// Prepare 10,000 keys to gossip. This is a large enough number to allow
+	// batching to kick in.
+	numKeys := 10_000
+	localKeys := make([]string, numKeys)
+	remoteKeys := make([]string, numKeys)
+	for i := 0; i < numKeys; i++ {
+		localKeys[i] = fmt.Sprintf("local-key-%d", i)
+		remoteKeys[i] = fmt.Sprintf("remote-key-%d", i)
+	}
+
+	// Gossip the keys to both local and remote nodes.
+	for i := range numKeys {
+		require.NoError(t, local.AddInfo(localKeys[i], []byte("value"), time.Hour))
+		require.NoError(t, remote.AddInfo(remoteKeys[i], []byte("value"), time.Hour))
+	}
+
+	// Wait for updates to propagate
+	testutils.SucceedsSoon(t, func() error {
+		for i := range numKeys {
+			if _, err := local.GetInfo(remoteKeys[i]); err != nil {
+				return err
+			}
+			if _, err := remote.GetInfo(localKeys[i]); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+
+	// Record the number of messages both the client and the server sent, and
+	// assert that it's within the expected bounds.
+	serverMessagesSentCount := remote.serverMetrics.MessagesSent.Count()
+	clientMessagesSentCount := local.serverMetrics.MessagesSent.Count()
+
+	fmt.Printf("client msgs sent: %+v\n", clientMessagesSentCount)
+	fmt.Printf("server msgs sent: %+v\n", serverMessagesSentCount)
+
+	// upperBoundMessages is the maximum number of sent messages we expect to see.
+	// Note that in reality with batching, we see 3-10 messages sent in this test,
+	// However, in order to avoid flakiness, we set a very high number here. The
+	// test would fail even with this high number if we don't have batching.
+	upperBoundMessages := int64(500)
+	require.LessOrEqual(t, serverMessagesSentCount, upperBoundMessages)
+	require.LessOrEqual(t, clientMessagesSentCount, upperBoundMessages)
+}

pkg/gossip/server.go

@@ -125,8 +125,10 @@ func (s *server) Gossip(stream Gossip_GossipServer) error {
 			infoCount := int64(len(reply.Delta))
 			s.nodeMetrics.BytesSent.Inc(bytesSent)
 			s.nodeMetrics.InfosSent.Inc(infoCount)
+			s.nodeMetrics.MessagesSent.Inc(1)
 			s.serverMetrics.BytesSent.Inc(bytesSent)
 			s.serverMetrics.InfosSent.Inc(infoCount)
+			s.serverMetrics.MessagesSent.Inc(1)
 
 			return stream.Send(reply)
 		}
@@ -199,6 +201,9 @@ func (s *server) Gossip(stream Gossip_GossipServer) error {
 		case err := <-errCh:
 			return err
 		case <-ready:
+			// We just sleep here instead of calling batchAndConsume() because the
+			// channel is closed, and sleeping won't block the sender of the channel.
+			time.Sleep(infosBatchDelay)
 		}
 	}
 }
@@ -307,8 +312,10 @@ func (s *server) gossipReceiver(
 		infosReceived := int64(len(args.Delta))
 		s.nodeMetrics.BytesReceived.Inc(bytesReceived)
 		s.nodeMetrics.InfosReceived.Inc(infosReceived)
+		s.nodeMetrics.MessagesReceived.Inc(1)
 		s.serverMetrics.BytesReceived.Inc(bytesReceived)
 		s.serverMetrics.InfosReceived.Inc(infosReceived)
+		s.serverMetrics.MessagesReceived.Inc(1)
 
 		freshCount, err := s.mu.is.combine(args.Delta, args.NodeID)
 		if err != nil {

pkg/gossip/status.go

@@ -26,6 +26,8 @@ type Metrics struct {
 	BytesSent                   *metric.Counter
 	InfosReceived               *metric.Counter
 	InfosSent                   *metric.Counter
+	MessagesSent                *metric.Counter
+	MessagesReceived            *metric.Counter
 	CallbacksProcessed          *metric.Counter
 	CallbacksPending            *metric.Gauge
 	CallbacksProcessingDuration metric.IHistogram
@@ -39,6 +41,8 @@ func makeMetrics() Metrics {
 		BytesSent:          metric.NewCounter(MetaBytesSent),
 		InfosReceived:      metric.NewCounter(MetaInfosReceived),
 		InfosSent:          metric.NewCounter(MetaInfosSent),
+		MessagesReceived:   metric.NewCounter(MetaMessagesReceived),
+		MessagesSent:       metric.NewCounter(MetaMessagesSent),
 		CallbacksProcessed: metric.NewCounter(MetaCallbacksProcessed),
 		CallbacksPending:   metric.NewGauge(MetaCallbacksPending),
 		CallbacksProcessingDuration: metric.NewHistogram(metric.HistogramOptions{
@@ -63,11 +67,13 @@ func (m Metrics) String() string {
 // Snapshot returns a snapshot of the metrics.
 func (m Metrics) Snapshot() MetricSnap {
 	return MetricSnap{
-		ConnsRefused:  m.ConnectionsRefused.Count(),
-		BytesReceived: m.BytesReceived.Count(),
-		BytesSent:     m.BytesSent.Count(),
-		InfosReceived: m.InfosReceived.Count(),
-		InfosSent:     m.InfosSent.Count(),
+		ConnsRefused:     m.ConnectionsRefused.Count(),
+		BytesReceived:    m.BytesReceived.Count(),
+		BytesSent:        m.BytesSent.Count(),
+		InfosReceived:    m.InfosReceived.Count(),
+		InfosSent:        m.InfosSent.Count(),
+		MessagesReceived: m.MessagesReceived.Count(),
+		MessagesSent:     m.MessagesSent.Count(),
 	}
 }
 
@@ -77,7 +83,8 @@ func (m MetricSnap) String() string {
 
 // SafeFormat implements the redact.SafeFormatter interface.
 func (m MetricSnap) SafeFormat(w redact.SafePrinter, _ rune) {
-	w.Printf("infos %d/%d sent/received, bytes %dB/%dB sent/received",
+	w.Printf("messages %d/%d sent/received, infos %d/%d sent/received, bytes %dB/%dB sent/received",
+		m.MessagesSent, m.MessagesReceived,
 		m.InfosSent, m.InfosReceived,
 		m.BytesSent, m.BytesReceived)
 	if m.ConnsRefused > 0 {

pkg/gossip/status_test.go

@@ -21,14 +21,16 @@ func TestGossipStatus(t *testing.T) {
 		},
 		MaxConns: 3,
 		MetricSnap: MetricSnap{
-			BytesReceived: 1000,
-			BytesSent:     2000,
-			InfosReceived: 10,
-			InfosSent:     20,
-			ConnsRefused:  17,
+			BytesReceived:    1000,
+			BytesSent:        2000,
+			InfosReceived:    10,
+			InfosSent:        20,
+			MessagesSent:     2,
+			MessagesReceived: 1,
+			ConnsRefused:     17,
 		},
 	}
-	if exp, act := `gossip server (2/3 cur/max conns, infos 20/10 sent/received, bytes 2000B/1000B sent/received, refused 17 conns)
+	if exp, act := `gossip server (2/3 cur/max conns, messages 2/1 sent/received, infos 20/10 sent/received, bytes 2000B/1000B sent/received, refused 17 conns)
   1: localhost:1234 (17s)
   4: localhost:4567 (18s)
 `, ss.String(); exp != act {
@@ -39,13 +41,13 @@ func TestGossipStatus(t *testing.T) {
 		ConnStatus: []OutgoingConnStatus{
 			{
 				ConnStatus: ss.ConnStatus[0],
-				MetricSnap: MetricSnap{BytesReceived: 77, BytesSent: 88, InfosReceived: 11, InfosSent: 22},
+				MetricSnap: MetricSnap{BytesReceived: 77, BytesSent: 88, InfosReceived: 11, InfosSent: 22, MessagesReceived: 3, MessagesSent: 4},
 			},
 		},
 		MaxConns: 3,
 	}
 	if exp, act := `gossip client (1/3 cur/max conns)
-  1: localhost:1234 (17s: infos 22/11 sent/received, bytes 88B/77B sent/received)
+  1: localhost:1234 (17s: messages 4/3 sent/received, infos 22/11 sent/received, bytes 88B/77B sent/received)
 `, cs.String(); exp != act {
 		t.Errorf("expected:\n%q\ngot:\n%q", exp, act)
 	}

pkg/gossip/util.go

@@ -8,9 +8,11 @@ package gossip
 import (
 	"bytes"
 	"sort"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/config"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 )
 
 // SystemConfigDeltaFilter keeps track of SystemConfig values so that unmodified
@@ -92,3 +94,18 @@ func (df *SystemConfigDeltaFilter) ForModified(
 		}
 	}
 }
+
+// batchAndConsume waits on a channel to allow batching more events. It keeps
+// while consuming events as they come to avoid blocking the channel producer.
+func batchAndConsume(ch <-chan struct{}, batchDuration time.Duration) {
+	var batchTimer timeutil.Timer
+	defer batchTimer.Stop()
+	batchTimer.Reset(batchDuration)
+	for !batchTimer.Read {
+		select {
+		case <-ch: // event happened while we are waiting for our batchTimer to end.
+		case <-batchTimer.C:
+			batchTimer.Read = true
+		}
+	}
+}