diff --git a/Makefile b/Makefile index e667ab61..06daf350 100644 --- a/Makefile +++ b/Makefile @@ -252,6 +252,14 @@ build-storage-bench: proto @mkdir -p "$(dir $(BENCH_OUTPUT))" @cd storage && CGO_ENABLED=1 CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" go test $(LDFLAGS) -c -o "$(BENCH_OUTPUT)" . +# Compile the ring heartbeat benchmark once so paired runs do not include Go +# compilation. PERFLOOP_BUILD_OUTPUT_DIR is supplied by the benchmark runner. +.PHONY: build-bench-ring-heartbeat +build-bench-ring-heartbeat: + @test -n "$(PERFLOOP_BUILD_OUTPUT_DIR)" || { echo "PERFLOOP_BUILD_OUTPUT_DIR is required"; exit 1; } + @mkdir -p "$(PERFLOOP_BUILD_OUTPUT_DIR)" + @cd coordinator && go test $(LDFLAGS) -c -o "$(PERFLOOP_BUILD_OUTPUT_DIR)/ring-heartbeat.test" ./ring + # Compile the topology RPC benchmark once so paired runs do not include Go # compilation. PERFLOOP_BUILD_OUTPUT_DIR is supplied by the benchmark runner. .PHONY: build-bench-cache-topology-rpc @@ -615,6 +623,7 @@ help: @echo " bench - Run benchmarks" @echo " bench-integration - Compile integration benchmarks (set INTEGRATION_BENCH_BINARY)" @echo " build-bench-compaction-serving-reads - Compile the compaction/serving benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)" + @echo " build-bench-ring-heartbeat - Compile the ring heartbeat benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)" @echo " build-bench-cache-topology-rpc - Compile the CacheService topology RPC benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)" @echo " build-bench-cache-service-delete - Compile the CacheService delete benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)" @echo " build-bench-cache-service-list-with-values - Compile the CacheService ListWithValues benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)" diff --git a/coordinator/ring/config.go b/coordinator/ring/config.go index 1a9ad7d4..46a99ee6 100644 --- a/coordinator/ring/config.go +++ b/coordinator/ring/config.go @@ -13,8 +13,12 @@ import ( ) const ( - // DefaultHeartbeatPeriod is the default interval for heartbeat updates - DefaultHeartbeatPeriod = 500 * time.Millisecond + // DefaultHeartbeatPeriod is the default interval for heartbeat updates. + // Ring timestamps have one-second precision, so sub-second periods are coalesced. + DefaultHeartbeatPeriod = time.Second + + // MinHeartbeatPeriod is the smallest effective interval for ring heartbeats. + MinHeartbeatPeriod = time.Second // MinHeartbeatTimeout is the default timeout before marking a node unhealthy. // Set long enough to handle rolling updates (60s allows for typical pod restarts). @@ -53,9 +57,8 @@ type Config struct { // ApplyDefaults applies default values for any unset or invalid configuration fields. // This should be called before using the configuration. func (c *Config) ApplyDefaults() { - if c.HeartbeatPeriod <= 0 { - c.HeartbeatPeriod = DefaultHeartbeatPeriod - } + c.HeartbeatPeriod = effectiveHeartbeatPeriod(c.HeartbeatPeriod) + // HeartbeatTimeout must be at least the maximum of: // - The configured value // - MinHeartbeatTimeout (60s) to prevent flaky health checks @@ -156,7 +159,19 @@ func (c *LifecyclerConfig) ApplyDefaults() { } } -// ToBasicLifecyclerConfig converts to dskit ring.BasicLifecyclerConfig +func effectiveHeartbeatPeriod(period time.Duration) time.Duration { + if period <= 0 { + return DefaultHeartbeatPeriod + } + if period < MinHeartbeatPeriod { + return MinHeartbeatPeriod + } + return period +} + +// ToBasicLifecyclerConfig converts to dskit ring.BasicLifecyclerConfig. +// The heartbeat floor is applied here as well as in ApplyDefaults so direct +// callers cannot pass a sub-second period to dskit's ticker. func (c *LifecyclerConfig) ToBasicLifecyclerConfig() ring.BasicLifecyclerConfig { addr := c.InstanceAddr if c.InstancePort > 0 { @@ -166,7 +181,7 @@ func (c *LifecyclerConfig) ToBasicLifecyclerConfig() ring.BasicLifecyclerConfig return ring.BasicLifecyclerConfig{ ID: c.InstanceID, Addr: addr, - HeartbeatPeriod: c.RingConfig.HeartbeatPeriod, + HeartbeatPeriod: effectiveHeartbeatPeriod(c.RingConfig.HeartbeatPeriod), HeartbeatTimeout: c.RingConfig.HeartbeatTimeout, TokensObservePeriod: c.ObservePeriod, NumTokens: c.NumTokens, diff --git a/coordinator/ring/config_test.go b/coordinator/ring/config_test.go index 7f3f732f..f36b7c5f 100644 --- a/coordinator/ring/config_test.go +++ b/coordinator/ring/config_test.go @@ -25,6 +25,18 @@ func TestConfig_ApplyDefaults(t *testing.T) { ReplicationFactor: 1, }, }, + { + name: "sub-second heartbeat is floored", + input: Config{ + HeartbeatPeriod: 500 * time.Millisecond, + HeartbeatTimeout: 5 * time.Second, + }, + expected: Config{ + HeartbeatPeriod: MinHeartbeatPeriod, + HeartbeatTimeout: MinHeartbeatTimeout, + ReplicationFactor: 1, + }, + }, { name: "short timeout is upgraded to minimum", input: Config{ @@ -198,3 +210,17 @@ func TestLifecyclerConfig_ToBasicLifecyclerConfig(t *testing.T) { }) } } + +func TestLifecyclerConfig_ToBasicLifecyclerConfig_FloorsSubSecondHeartbeat(t *testing.T) { + cfg := LifecyclerConfig{ + RingConfig: Config{ + HeartbeatPeriod: 500 * time.Millisecond, + }, + } + + basicCfg := cfg.ToBasicLifecyclerConfig() + + assert.Equal(t, MinHeartbeatPeriod, basicCfg.HeartbeatPeriod) + assert.Equal(t, 500*time.Millisecond, cfg.RingConfig.HeartbeatPeriod, + "conversion should not mutate an unnormalised config") +} diff --git a/coordinator/ring/heartbeat_benchmark_test.go b/coordinator/ring/heartbeat_benchmark_test.go new file mode 100644 index 00000000..59dfbba0 --- /dev/null +++ b/coordinator/ring/heartbeat_benchmark_test.go @@ -0,0 +1,226 @@ +// Copyright 2026 Tigris Data, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package ring + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/go-kit/log" + dskitring "github.com/grafana/dskit/ring" + "github.com/grafana/dskit/services" + "github.com/prometheus/client_golang/prometheus" + "github.com/tigrisdata/ocache/coordinator/gossip" +) + +const ( + heartbeatBenchmarkMembers = 8 + heartbeatBenchmarkTokensPerNode = 1 + heartbeatBenchmarkStep = 100 * time.Millisecond +) + +type heartbeatBenchmarkDelegate struct { + base dskitring.InstanceRegisterDelegate + callbacks atomic.Uint64 + production ringDelegate +} + +func (d *heartbeatBenchmarkDelegate) OnRingInstanceRegister(l *dskitring.BasicLifecycler, desc dskitring.Desc, exists bool, id string, instance dskitring.InstanceDesc) (dskitring.InstanceState, dskitring.Tokens) { + return d.base.OnRingInstanceRegister(l, desc, exists, id, instance) +} + +func (d *heartbeatBenchmarkDelegate) OnRingInstanceTokens(*dskitring.BasicLifecycler, dskitring.Tokens) { +} + +func (d *heartbeatBenchmarkDelegate) OnRingInstanceStopping(*dskitring.BasicLifecycler) {} + +func (d *heartbeatBenchmarkDelegate) OnRingInstanceHeartbeat(l *dskitring.BasicLifecycler, desc *dskitring.Desc, instance *dskitring.InstanceDesc) { + d.callbacks.Add(1) + d.production.OnRingInstanceHeartbeat(l, desc, instance) +} + +type heartbeatBenchmarkFixture struct { + delegate *heartbeatBenchmarkDelegate + reg *prometheus.Registry + lifecycler *dskitring.BasicLifecycler + memberlistKV *gossip.Memberlist + cancel context.CancelFunc + stopOnce sync.Once +} + +func newHeartbeatBenchmarkFixture(b *testing.B) *heartbeatBenchmarkFixture { + b.Helper() + + ctx, cancel := context.WithCancel(context.Background()) + logger := log.NewNopLogger() + reg := prometheus.NewRegistry() + nodeID := "heartbeat-benchmark-member-0" + memberlistKV, err := gossip.NewMemberlist(nodeID, "127.0.0.1:0", nil, logger, reg) + if err != nil { + cancel() + b.Fatalf("create memberlist: %v", err) + } + if err := memberlistKV.Start(ctx); err != nil { + cancel() + b.Fatalf("start memberlist: %v", err) + } + + desc := dskitring.NewDesc() + registeredAt := time.Now() + for member := 0; member < heartbeatBenchmarkMembers; member++ { + desc.AddIngester( + fmt.Sprintf("heartbeat-benchmark-member-%d", member), + fmt.Sprintf("127.0.0.1:%d", 10000+member), + "", + []uint32{uint32(member + 1)}, + dskitring.ACTIVE, + registeredAt, + ) + } + if err := memberlistKV.Client().CAS(ctx, RingKey, func(interface{}) (interface{}, bool, error) { + return desc, true, nil + }); err != nil { + _ = memberlistKV.Stop(context.Background()) + cancel() + b.Fatalf("seed ring: %v", err) + } + + cfg := LifecyclerConfig{ + InstanceID: nodeID, + InstanceAddr: "127.0.0.1:10000", + NumTokens: heartbeatBenchmarkTokensPerNode, + } + cfg.ApplyDefaults() + + delegate := &heartbeatBenchmarkDelegate{ + base: dskitring.NewInstanceRegisterDelegate(dskitring.ACTIVE, heartbeatBenchmarkTokensPerNode), + production: ringDelegate{rm: &RingManager{logger: logger}}, + } + basicCfg := cfg.ToBasicLifecyclerConfig() + // Keep teardown from unregistering the seeded instance with a non-heartbeat CAS. + basicCfg.KeepInstanceInTheRingOnShutdown = true + lifecycler, err := dskitring.NewBasicLifecycler( + basicCfg, + RingName, + RingKey, + memberlistKV.Client(), + delegate, + logger, + reg, + ) + if err != nil { + _ = memberlistKV.Stop(context.Background()) + cancel() + b.Fatalf("create lifecycler: %v", err) + } + if err := services.StartAndAwaitRunning(ctx, lifecycler); err != nil { + _ = memberlistKV.Stop(context.Background()) + cancel() + b.Fatalf("start lifecycler: %v", err) + } + + fixture := &heartbeatBenchmarkFixture{ + delegate: delegate, + reg: reg, + lifecycler: lifecycler, + memberlistKV: memberlistKV, + cancel: cancel, + } + b.Cleanup(fixture.stop) + return fixture +} + +func (f *heartbeatBenchmarkFixture) stop() { + f.stopOnce.Do(func() { + _ = services.StopAndAwaitTerminated(context.Background(), f.lifecycler) + _ = f.memberlistKV.Stop(context.Background()) + f.cancel() + }) +} + +type heartbeatBenchmarkCounts struct { + casAttempts float64 + casSuccesses float64 + heartbeats float64 + callbacks uint64 +} + +func heartbeatBenchmarkSnapshot(b *testing.B, fixture *heartbeatBenchmarkFixture) heartbeatBenchmarkCounts { + b.Helper() + + families, err := fixture.reg.Gather() + if err != nil { + b.Fatalf("gather heartbeat metrics: %v", err) + } + + var counts heartbeatBenchmarkCounts + found := make(map[string]bool, 3) + for _, family := range families { + var target *float64 + switch family.GetName() { + case "memberlist_client_cas_attempt_total": + target = &counts.casAttempts + case "memberlist_client_cas_success_total": + target = &counts.casSuccesses + case "ring_member_heartbeats_total": + target = &counts.heartbeats + default: + continue + } + found[family.GetName()] = true + for _, metric := range family.GetMetric() { + *target += metric.GetCounter().GetValue() + } + } + + for _, name := range []string{ + "memberlist_client_cas_attempt_total", + "memberlist_client_cas_success_total", + "ring_member_heartbeats_total", + } { + if !found[name] { + b.Fatalf("metric %s not found", name) + } + } + counts.callbacks = fixture.delegate.callbacks.Load() + return counts +} + +func BenchmarkRingHeartbeat(b *testing.B) { + fixture := newHeartbeatBenchmarkFixture(b) + before := heartbeatBenchmarkSnapshot(b, fixture) + + b.ResetTimer() + for b.Loop() { + time.Sleep(heartbeatBenchmarkStep) + } + b.StopTimer() + + // Stop-time lifecycle work is outside the heartbeat workload. Capture the + // endpoint before stopping so failed teardown CAS attempts cannot be folded + // into the heartbeat-attempt calculation. + after := heartbeatBenchmarkSnapshot(b, fixture) + fixture.stop() + + heartbeats := after.heartbeats - before.heartbeats + callbacks := float64(after.callbacks - before.callbacks) + casAttempts := after.casAttempts - before.casAttempts + casSuccesses := after.casSuccesses - before.casSuccesses + nonHeartbeatCAS := casSuccesses - heartbeats + heartbeatAttempts := casAttempts - nonHeartbeatCAS + if heartbeats <= 0 { + b.Fatalf("heartbeat benchmark produced no successful heartbeats (attempts=%.0f callbacks=%.0f)", casAttempts, callbacks) + } + if heartbeatAttempts < heartbeats || callbacks < heartbeats { + b.Fatalf("heartbeat counters are inconsistent (attempts=%.0f successes=%.0f callbacks=%.0f heartbeats=%.0f)", heartbeatAttempts, casSuccesses, callbacks, heartbeats) + } + + b.ReportMetric(0, "ns/op") + b.ReportMetric(heartbeatAttempts/heartbeats, "cas-attempts/heartbeat") + b.ReportMetric(callbacks/heartbeats, "heartbeat-callbacks/heartbeat") +} diff --git a/docs/cluster.md b/docs/cluster.md index bd8a399f..f612f25f 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -191,7 +191,7 @@ Planned improvements: Each node monitors others via the gossip protocol: -- Regular heartbeat messages (default: every 5 seconds) +- Regular ring heartbeat messages (default: every 1 second; sub-second configured values are coalesced) - Configurable heartbeat timeout (default: 60 seconds) - Automatic topology updates propagated via memberlist - Epoch tracking for efficient client topology cache invalidation diff --git a/docs/rfcs/RFC-007-sharded-cluster.md b/docs/rfcs/RFC-007-sharded-cluster.md index f201d699..22a310a4 100644 --- a/docs/rfcs/RFC-007-sharded-cluster.md +++ b/docs/rfcs/RFC-007-sharded-cluster.md @@ -231,7 +231,7 @@ dskit BasicLifecycler manages instance state transitions: #### 4.3.3 Ring Heartbeats -- **Heartbeat Period**: 500ms (DefaultHeartbeatPeriod) +- **Heartbeat Period**: 1s minimum (DefaultHeartbeatPeriod); sub-second values are coalesced because ring timestamps use Unix-second precision - **Heartbeat Timeout**: 60s minimum (MinHeartbeatTimeout) - Ring heartbeats update instance state in the KV store - Other nodes receive updates via memberlist gossip (not direct peer-to-peer) @@ -263,7 +263,7 @@ Ring state is stored in a distributed KV backed by memberlist: | ------------------- | -------------------------------------- | --------------------------------- | | **Gossip Messages** | Propagate state changes between nodes | 200ms interval, 3 nodes per cycle | | **Push/Pull Sync** | Full state reconciliation | Every 30 seconds | -| **Ring Heartbeats** | Update instance liveness in KV | Every 500ms | +| **Ring Heartbeats** | Update instance liveness in KV | Every 1s minimum; sub-second values are coalesced | | **KV Watcher** | Immediate notification of ring changes | Event-driven | #### 4.4.2 Gossip Configuration