gossip: batch both client and server gossip messages

This commit adds an optimization in gossip where before sending a gossip
info update message, we wait for a small amount of time before sending
the updates. This gives some time for updates to be batched, reducing
the number of messages in the system.

Fixes: #119420

Release note: None

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

@@ -330,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)
@@ -394,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
 )
 

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

@@ -201,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)
 		}
 	}
 }

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
+		}
+	}
+}