Merge #157825

157825: kvnemesis: add clock skew and drift r=stevendanna a=stevendanna

This injects clock drift between the nodes in a multi-node KVNemesis cluster. To start, the clock uses a Sinusoidal drift centered on a fixed offset from realtime with an amplitude that ensures any two clocks should stay within the MaxOffset. This is perhaps not very realistic, but we can improve this over time.

Epic: none
Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/kv/kvnemesis/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     name = "kvnemesis",
     srcs = [
         "applier.go",
+        "clock.go",
         "doc.go",
         "engine.go",
         "env.go",
@@ -65,6 +66,7 @@ go_test(
     size = "large",
     srcs = [
         "applier_test.go",
+        "clock_test.go",
         "engine_test.go",
         "generator_test.go",
         "kvnemesis_test.go",
@@ -117,6 +119,7 @@ go_test(
         "//pkg/util/randutil",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
+        "//pkg/util/timeutil",
         "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//assert",

pkg/kv/kvnemesis/clock.go

@@ -0,0 +1,90 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package kvnemesis
+
+import (
+	"fmt"
+	"math"
+	"math/rand"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+)
+
+// testClock is an hlc.WallClock that simulates drift around some fixed offset
+// from the given system clock.
+type testClock struct {
+	clock hlc.WallClock
+
+	sign int
+	// freq is the angular frequency in radians/nanosecond.
+	freq        float64
+	amplitude   time.Duration
+	fixedOffset time.Duration
+}
+
+// NewTestClock returns a testClock.
+//
+// The maxOffset is the maximum offset of two clocks created with NewTestClock.
+// The clock will return times in the range
+//
+//	[t-maxOffset/2, t+maxOffset/2]
+//
+// We currently use a sine wave to drift the clock over time. Note that this was
+// chosen without much detailed research. The goal was to have something that
+// drifted the clock over time in one direction to simulate a slow or fast clock
+// and then move it in the opposite direction to simulate ntpd or chronyd
+// correcting the clock.
+//
+// We may instead want something more like a saw-tooth pattern where the
+// correction is rather abrupt.
+func NewTestClock(maxOffset time.Duration, period time.Duration, rng *rand.Rand) *testClock {
+	maxAmplitude := time.Duration(float64(maxOffset / 2))
+	// We don't want all the clocks moving in lock step. We do two things:
+	//
+	// 1. Randomly swap the initial direction of drift, and
+	//
+	// 2. Add a small fixed offset from the center (reducing amplitude
+	// appropriately), in the direction of the swap.
+	sign := 1
+	if rng.Intn(2) == 1 {
+		sign = -1
+	}
+	jitter := 0.10 * rng.Float64()
+	fixedOffset := time.Duration(float64(maxAmplitude) * jitter)
+	amplitude := maxAmplitude - fixedOffset
+	freq := float64(2.0*math.Pi) / float64(period)
+	return &testClock{
+		clock:       timeutil.DefaultTimeSource{},
+		sign:        sign,
+		freq:        freq,
+		amplitude:   amplitude,
+		fixedOffset: fixedOffset,
+	}
+}
+
+var _ hlc.WallClock = (*testClock)(nil)
+
+func (c *testClock) Now() time.Time {
+	//
+	// f(t) = t + sign*(amplitude*sin(freq*t) + offset)
+	//
+	// TODO(ssd): math.Sin on every call to Now() might be expensive. One
+	// alternative might be simple linear growth up to some max which then
+	// reverses direction until the offset is back to zero. Or, if we like the Sin
+	// wave perhaps we can store of the current offset and then only update it
+	// every so many calls.
+	t := c.clock.Now()
+	p := math.Sin(c.freq * float64(t.UnixNano()))
+	adjust := float64(c.amplitude)*p + float64(c.fixedOffset)
+	adjust = float64(c.sign) * adjust
+	return t.Add(time.Duration(adjust))
+}
+
+func (c *testClock) String() string {
+	return fmt.Sprintf("clock[offset=%s,amp=%s]", time.Duration(c.sign)*c.fixedOffset, c.amplitude)
+}

pkg/kv/kvnemesis/clock_test.go

@@ -0,0 +1,71 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package kvnemesis
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTestClock(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	rng, _ := randutil.NewTestRand()
+	maxOffset := 10 * time.Millisecond
+	period := 10 * time.Second
+	tolerance := 50 * time.Microsecond
+
+	t.Run("clock nears amplitude at period", func(t *testing.T) {
+		c := NewTestClock(maxOffset, period, rng)
+		clock := timeutil.NewManualTime(timeutil.Unix(0, 0))
+		c.clock = clock
+		clock.Advance(period)
+		now := c.Now()
+		require.Less(t, tolerance, abs(c.amplitude.Nanoseconds()-abs(period.Nanoseconds()-now.UnixNano())))
+	})
+	t.Run("5000 advances", func(t *testing.T) {
+		clock := timeutil.NewManualTime(timeutil.Unix(0, 0))
+
+		c1 := NewTestClock(maxOffset, period, rng)
+		c2 := NewTestClock(maxOffset, period, rng)
+		c1.clock = clock
+		c2.clock = clock
+
+		for range 5000 {
+			adv := rng.Int31n(100)
+			clock.Advance(time.Duration(adv) * time.Millisecond)
+			now1 := c1.Now()
+			now2 := c2.Now()
+			actual := clock.Now()
+
+			// The two clocks should never be further apart than maxOffset
+			require.LessOrEqual(t, time.Duration(abs(int64(now1.Sub(now2)))), maxOffset)
+
+			// Also just check each clock on its own.
+			clockInsideBounds := func(t *testing.T, computed time.Time, actual time.Time) {
+				maxErr := maxOffset/2 + tolerance
+				max := actual.Add(maxErr)
+				min := actual.Add(-maxErr)
+				require.Less(t, computed, max)
+				require.Greater(t, computed, min)
+			}
+			clockInsideBounds(t, now1, actual)
+			clockInsideBounds(t, now2, actual)
+		}
+	})
+}
+
+func abs(i int64) int64 {
+	if i < 0 {
+		return -1 * i
+	}
+	return i
+}

pkg/kv/kvnemesis/kvnemesis_test.go

@@ -58,7 +58,12 @@ import (
 var defaultNumSteps = envutil.EnvOrDefaultInt("COCKROACH_KVNEMESIS_STEPS", 100)
 
 func (cfg kvnemesisTestCfg) testClusterArgs(
-	ctx context.Context, tr *SeqTracker, partitioner *rpc.Partitioner, mode TestMode,
+	ctx context.Context,
+	t testing.TB,
+	rng *rand.Rand,
+	tr *SeqTracker,
+	partitioner *rpc.Partitioner,
+	mode TestMode,
 ) (base.TestClusterArgs, stop.Closer) {
 	storeKnobs := &kvserver.StoreTestingKnobs{
 		DisableRaftLogQueue:                   true,
@@ -258,15 +263,28 @@ func (cfg kvnemesisTestCfg) testClusterArgs(
 			for i := 0; i < cfg.numNodes; i++ {
 				nodeId := i + 1
 				ctk := rpc.ContextTestingKnobs{}
+				// Test clock injection
+				//
+				// We need to stay away within 80% of the max offset, so we set our
+				// offset to 70% of the configured MaxOffset. Since MaxOffset is
+				// relatively small, this doesn't give us much wiggle room.
+				//
+				// NOTE(ssd): If we see flakes because of untrustworthy remote clocks,
+				// don't hesitate to increase this safety margin for now.
+				maxOffset := time.Duration(float64(storeKnobs.MaxOffset) * 0.70)
+				period := 1 * time.Second
+				clock := NewTestClock(maxOffset, period, rng)
+				t.Logf("n%d using %s", nodeId, clock)
 				partitioner.RegisterTestingKnobs(roachpb.NodeID(nodeId), &ctk)
 				perNodeServerArgs := commonServerArgs
 				perNodeServerArgs.Knobs.Server = &server.TestingKnobs{
+					StickyVFSRegistry:   reg,
+					WallClock:           clock,
 					ContextTestingKnobs: ctk,
 				}
 				perNodeServerArgs.Locality = roachpb.Locality{
 					Tiers: []roachpb.Tier{{Key: "node", Value: fmt.Sprintf("n%d", nodeId)}},
 				}
-				perNodeServerArgs.Knobs.Server = &server.TestingKnobs{StickyVFSRegistry: reg}
 				perNodeServerArgs.StoreSpecs = append(
 					perNodeServerArgs.StoreSpecs,
 					base.StoreSpec{InMemory: true, StickyVFSID: strconv.Itoa(nodeId)},
@@ -608,7 +626,7 @@ func testKVNemesisImpl(t testing.TB, cfg kvnemesisTestCfg) {
 	ctx := context.Background()
 	tr := &SeqTracker{}
 	var partitioner rpc.Partitioner
-	args, closer := cfg.testClusterArgs(ctx, tr, &partitioner, cfg.mode)
+	args, closer := cfg.testClusterArgs(ctx, t, rng, tr, &partitioner, cfg.mode)
 	tc := testcluster.StartTestCluster(t, cfg.numNodes, args)
 	tc.Stopper().AddCloser(closer)
 	defer tc.Stopper().Stop(ctx)