Merge #84031

84031: rpc: fix the maxoffset check on the incoming path r=erikgrinaker a=knz

Fixes #84017. 
Fixes #84027.

### Bug fix 1

Back in PR #34197 we mistakenly removed the .Offset field
sent by each RPC heartbeat, through which the remote node monitors
the current/client node's offset.

This looks bad, but is actually somewhat innocuous. This is because
we update the offset map and do the check on two situations:

- when a server node gets pinged by a remote node using the remote-provided .Offset,
- when receiving a ping response as client, using an estimate of the roundtrip
  latency between PingRequest-PingResponse.

The bug above only invalidates the first check. The second check still
occurs. Since all nodes are both client to another node, and server
for the same other node, we still get a check both ways on all pairs of
nodes.

Nonetheless, the half-way broken check reduces robustness overall. So
it's good to fix it.

### Bug fix 2

Release note (bug fix): CLI `cockroach` commands connecting to a
remote server will not produce spurious "latency jump" warnings any
more. This bug had been introduced in CockroachDB v21.2.

Co-authored-by: Raphael 'kena' Poss <[email protected]>

Author: craig[bot]

pkg/rpc/context.go

@@ -577,13 +577,19 @@ func NewContext(ctx context.Context, opts ContextOptions) *Context {
 		breakerClock: breakerClock{
 			clock: opts.Clock,
 		},
-		RemoteClocks: newRemoteClockMonitor(
-			opts.Clock, opts.MaxOffset, 10*opts.Config.RPCHeartbeatInterval, opts.Config.HistogramWindowInterval()),
 		rpcCompression:   enableRPCCompression,
 		MasterCtx:        masterCtx,
 		metrics:          makeMetrics(),
 		heartbeatTimeout: 2 * opts.Config.RPCHeartbeatInterval,
 	}
+
+	// We only monitor remote clocks in server-to-server connections.
+	// CLI commands are exempted.
+	if !opts.ClientOnly {
+		rpcCtx.RemoteClocks = newRemoteClockMonitor(
+			opts.Clock, opts.MaxOffset, 10*opts.Config.RPCHeartbeatInterval, opts.Config.HistogramWindowInterval())
+	}
+
 	if id := opts.Knobs.StorageClusterID; id != nil {
 		rpcCtx.StorageClusterID.Set(masterCtx, *id)
 	}
@@ -1666,6 +1672,16 @@ func (rpcCtx *Context) runHeartbeat(
 	maxOffset := rpcCtx.MaxOffset
 	maxOffsetNanos := maxOffset.Nanoseconds()
 
+	// The request object. Note that we keep the same object from
+	// heartbeat to heartbeat: we compute a new .Offset at the end of
+	// the current heartbeat as input to the next one.
+	request := &PingRequest{
+		OriginAddr:           rpcCtx.Config.Addr,
+		OriginMaxOffsetNanos: maxOffsetNanos,
+		TargetNodeID:         conn.remoteNodeID,
+		ServerVersion:        rpcCtx.Settings.Version.BinaryVersion(),
+	}
+
 	heartbeatClient := NewHeartbeatClient(conn.grpcConn)
 
 	var heartbeatTimer timeutil.Timer
@@ -1692,16 +1708,10 @@ func (rpcCtx *Context) runHeartbeat(
 		}
 
 		if err := rpcCtx.Stopper.RunTaskWithErr(ctx, "rpc heartbeat", func(ctx context.Context) error {
-			// We re-mint the PingRequest to pick up any asynchronous update to clusterID.
+			// Pick up any asynchronous update to clusterID and NodeID.
 			clusterID := rpcCtx.StorageClusterID.Get()
-			request := &PingRequest{
-				OriginNodeID:         rpcCtx.NodeID.Get(),
-				OriginAddr:           rpcCtx.Config.Addr,
-				OriginMaxOffsetNanos: maxOffsetNanos,
-				ClusterID:            &clusterID,
-				TargetNodeID:         conn.remoteNodeID,
-				ServerVersion:        rpcCtx.Settings.Version.BinaryVersion(),
-			}
+			request.ClusterID = &clusterID
+			request.OriginNodeID = rpcCtx.NodeID.Get()
 
 			interceptor := func(context.Context, *PingRequest) error { return nil }
 			if fn := rpcCtx.OnOutgoingPing; fn != nil {
@@ -1760,25 +1770,33 @@ func (rpcCtx *Context) runHeartbeat(
 
 			if err == nil {
 				everSucceeded = true
-				receiveTime := rpcCtx.Clock.Now()
-
-				// Only update the clock offset measurement if we actually got a
-				// successful response from the server.
-				pingDuration := receiveTime.Sub(sendTime)
-				if pingDuration > maximumPingDurationMult*rpcCtx.MaxOffset {
-					request.Offset.Reset()
-				} else {
-					// Offset and error are measured using the remote clock reading
-					// technique described in
-					// http://se.inf.tu-dresden.de/pubs/papers/SRDS1994.pdf, page 6.
-					// However, we assume that drift and min message delay are 0, for
-					// now.
-					request.Offset.MeasuredAt = receiveTime.UnixNano()
-					request.Offset.Uncertainty = (pingDuration / 2).Nanoseconds()
-					remoteTimeNow := timeutil.Unix(0, response.ServerTime).Add(pingDuration / 2)
-					request.Offset.Offset = remoteTimeNow.Sub(receiveTime).Nanoseconds()
+
+				// Only a server connecting to another server needs to check
+				// clock offsets. A CLI command does not need to update its
+				// local HLC, nor does it care that strictly about
+				// client-server latency, nor does it need to track the
+				// offsets.
+				if rpcCtx.RemoteClocks != nil {
+					receiveTime := rpcCtx.Clock.Now()
+
+					// Only update the clock offset measurement if we actually got a
+					// successful response from the server.
+					pingDuration := receiveTime.Sub(sendTime)
+					if pingDuration > maximumPingDurationMult*rpcCtx.MaxOffset {
+						request.Offset.Reset()
+					} else {
+						// Offset and error are measured using the remote clock reading
+						// technique described in
+						// http://se.inf.tu-dresden.de/pubs/papers/SRDS1994.pdf, page 6.
+						// However, we assume that drift and min message delay are 0, for
+						// now.
+						request.Offset.MeasuredAt = receiveTime.UnixNano()
+						request.Offset.Uncertainty = (pingDuration / 2).Nanoseconds()
+						remoteTimeNow := timeutil.Unix(0, response.ServerTime).Add(pingDuration / 2)
+						request.Offset.Offset = remoteTimeNow.Sub(receiveTime).Nanoseconds()
+					}
+					rpcCtx.RemoteClocks.UpdateOffset(ctx, target, request.Offset, pingDuration)
 				}
-				rpcCtx.RemoteClocks.UpdateOffset(ctx, target, request.Offset, pingDuration)
 
 				if cb := rpcCtx.HeartbeatCB; cb != nil {
 					cb()

pkg/rpc/context_test.go

@@ -226,6 +226,96 @@ func TestPingInterceptors(t *testing.T) {
 	}
 }
 
+// TestClockOffsetInPingRequest ensures that all ping requests
+// after the first one have a non-zero offset.
+// (Regression test for issue #84027.)
+func TestClockOffsetInPingRequest(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	testClockOffsetInPingRequestInternal(t, false /* clientOnly */)
+}
+
+// TestClockOffsetInclientPingRequest ensures that all ping requests
+// have a zero offset and there is no Remotelocks to update.
+// (Regression test for issue #84017.)
+func TestClockOffsetInClientPingRequest(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	testClockOffsetInPingRequestInternal(t, true /* clientOnly */)
+}
+
+func testClockOffsetInPingRequestInternal(t *testing.T, clientOnly bool) {
+	ctx := context.Background()
+	stopper := stop.NewStopper()
+	defer stopper.Stop(ctx)
+
+	pings := make(chan PingRequest, 5)
+	done := make(chan struct{})
+	defer func() { close(done) }()
+
+	// Build a minimal server.
+	opts := ContextOptions{
+		TenantID:  roachpb.SystemTenantID,
+		Config:    testutils.NewNodeTestBaseContext(),
+		Clock:     &timeutil.DefaultTimeSource{},
+		MaxOffset: 500 * time.Millisecond,
+		Stopper:   stopper,
+		Settings:  cluster.MakeTestingClusterSettings(),
+	}
+	rpcCtxServer := NewContext(ctx, opts)
+
+	clientOpts := opts
+	clientOpts.Config = testutils.NewNodeTestBaseContext()
+	// Experimentally, values below 50ms seem to incur flakiness.
+	clientOpts.Config.RPCHeartbeatInterval = 100 * time.Millisecond
+	clientOpts.ClientOnly = clientOnly
+	clientOpts.OnOutgoingPing = func(ctx context.Context, req *PingRequest) error {
+		select {
+		case <-done:
+		case pings <- *req:
+		}
+		return nil
+	}
+	rpcCtxClient := NewContext(ctx, clientOpts)
+
+	require.NotNil(t, rpcCtxServer.RemoteClocks)
+	if !clientOnly {
+		require.NotNil(t, rpcCtxClient.RemoteClocks)
+	} else {
+		require.Nil(t, rpcCtxClient.RemoteClocks)
+	}
+
+	t.Logf("server listen")
+	s := newTestServer(t, rpcCtxServer)
+	RegisterHeartbeatServer(s, rpcCtxServer.NewHeartbeatService())
+	rpcCtxServer.NodeID.Set(ctx, 1)
+	ln, err := netutil.ListenAndServeGRPC(stopper, s, util.TestAddr)
+	require.NoError(t, err)
+
+	t.Logf("client dial")
+	// Dial: this causes the heartbeats to start.
+	remoteAddr := ln.Addr().String()
+	_, err = rpcCtxClient.GRPCDialNode(remoteAddr, 1, SystemClass).Connect(ctx)
+	require.NoError(t, err)
+
+	t.Logf("first ping check")
+	firstPing := <-pings
+	require.Zero(t, firstPing.Offset.Offset)
+	require.Zero(t, firstPing.Offset.Uncertainty)
+	require.Zero(t, firstPing.Offset.MeasuredAt)
+	for i := 1; i < 3; i++ {
+		t.Logf("ping %d check", i)
+		nextPing := <-pings
+		if !clientOnly {
+			require.NotZero(t, nextPing.Offset.Offset, i)
+			require.NotZero(t, nextPing.Offset.Uncertainty, i)
+			require.NotZero(t, nextPing.Offset.MeasuredAt, i)
+		} else {
+			require.Zero(t, nextPing.Offset.Offset, i)
+			require.Zero(t, nextPing.Offset.Uncertainty, i)
+			require.Zero(t, nextPing.Offset.MeasuredAt, i)
+		}
+	}
+}
+
 var _ roachpb.InternalServer = &internalServer{}
 
 type internalServer struct {

pkg/rpc/nodedialer/nodedialer.go

@@ -294,7 +294,10 @@ func (n *Dialer) getBreaker(nodeID roachpb.NodeID, class rpc.ConnectionClass) *w
 // samples to compute a reliable average.
 func (n *Dialer) Latency(nodeID roachpb.NodeID) (time.Duration, error) {
 	if n == nil || n.resolver == nil {
-		return 0, errors.New("no node dialer configured")
+		return 0, errors.AssertionFailedf("no node dialer configured")
+	}
+	if n.rpcContext.RemoteClocks == nil {
+		return 0, errors.AssertionFailedf("can't call Latency in a client command")
 	}
 	addr, err := n.resolver(nodeID)
 	if err != nil {