diff --git a/AGENTS.md b/AGENTS.md index 5383403..fefb3cc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,7 +15,9 @@ read it when planning changes to the areas it covers, and keep it current: - [`docs/internal/metrics.md`](docs/internal/metrics.md) — how metrics are split (transport vs producer-state vs warpstream-specific) and the franz-go/`kprom` - drop-in-compatibility contract. Update it when changing metrics. + drop-in-compatibility contract. Update it when that categorization or parity + contract changes. Do not catalog individual custom metrics here; their + Prometheus help strings are the source of truth. - [`docs/internal/tracing.md`](docs/internal/tracing.md) — how the client drives franz-go's produce-record hooks on its own produce path to support tracing (e.g. `kotel`) as a drop-in. Update it when changing hook or tracing behaviour. diff --git a/pkg/wgo/agentpool.go b/pkg/wgo/agentpool.go index a0dfb99..90a1a26 100644 --- a/pkg/wgo/agentpool.go +++ b/pkg/wgo/agentpool.go @@ -224,3 +224,25 @@ func diffRemovedAgents(old []int32, newSet map[int32]struct{}) []int32 { } return removed } + +// diffAgentMembership counts NodeIDs that appeared or disappeared between two +// sorted, unique agent lists. +func diffAgentMembership(old, new []int32) (added, removed int) { + i, j := 0, 0 + for i < len(old) && j < len(new) { + switch { + case old[i] == new[j]: + i++ + j++ + case old[i] < new[j]: + removed++ + i++ + default: + added++ + j++ + } + } + removed += len(old) - i + added += len(new) - j + return added, removed +} diff --git a/pkg/wgo/agentpool_test.go b/pkg/wgo/agentpool_test.go index 35ccd6e..ee58a2b 100644 --- a/pkg/wgo/agentpool_test.go +++ b/pkg/wgo/agentpool_test.go @@ -366,6 +366,41 @@ func TestBuildLeadersAndTopicIDs(t *testing.T) { func stringPtr(s string) *string { return &s } +func TestAgentPool_DiffAgentMembership(t *testing.T) { + tests := map[string]struct { + old []int32 + new []int32 + wantAdded int + wantRemoved int + }{ + "unchanged": { + old: []int32{1, 2, 3}, new: []int32{1, 2, 3}, + }, + "one added": { + old: []int32{1, 2}, new: []int32{1, 2, 3}, wantAdded: 1, + }, + "one removed": { + old: []int32{1, 2, 3}, new: []int32{1, 3}, wantRemoved: 1, + }, + "replace one": { + old: []int32{1, 2, 3}, new: []int32{1, 4}, wantAdded: 1, wantRemoved: 2, + }, + "empty to some": { + old: nil, new: []int32{1, 2}, wantAdded: 2, + }, + "all removed": { + old: []int32{1, 2}, new: nil, wantRemoved: 2, + }, + } + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + added, removed := diffAgentMembership(tc.old, tc.new) + assert.Equal(t, tc.wantAdded, added) + assert.Equal(t, tc.wantRemoved, removed) + }) + } +} + func TestDiffRemovedAgents(t *testing.T) { tests := map[string]struct { old []int32 diff --git a/pkg/wgo/client.go b/pkg/wgo/client.go index c803585..63387fb 100644 --- a/pkg/wgo/client.go +++ b/pkg/wgo/client.go @@ -151,7 +151,7 @@ func NewWarpstreamClient(logger kgo.Logger, reg prometheus.Registerer, opts ...O refreshCancel: refreshCancel, refreshNowCh: make(chan struct{}, 1), } - // Count a dropped leader and do not nudge: the refresh goroutine starts below. + // Publish the excluded-leader count and do not nudge: the refresh goroutine starts below. // The periodic tick fetches the next snapshot. c.noteLeaderDrops(dropped, false) // Demoter sits on top of the lazy pool strategy so refresh-driven @@ -503,12 +503,14 @@ func (c *WarpstreamClient) refreshPool(trigger metadataRefreshTrigger) { c.noteLeaderDrops(dropped, true) } -// noteLeaderDrops counts and logs excluded leaders. nudge asks for another fetch. +// noteLeaderDrops publishes the excluded-leader count and logs it when it is +// above zero. nudge asks for another fetch. The gauge is set even at zero so it +// clears when the exclusion does. func (c *WarpstreamClient) noteLeaderDrops(dropped leaderDrops, nudge bool) { + c.metrics.agentPoolExcludedLeaders.Set(float64(dropped.Count)) if dropped.Count == 0 { return } - c.metrics.agentPoolLeaderDroppedTotal.Add(float64(dropped.Count)) log(c.logger, kgo.LogLevelWarn, "warpstream agentpool: leaders excluded from map", "count", dropped.Count, "first_topic", dropped.Topic, diff --git a/pkg/wgo/client_test.go b/pkg/wgo/client_test.go index 443a180..71931ad 100644 --- a/pkg/wgo/client_test.go +++ b/pkg/wgo/client_test.go @@ -82,6 +82,30 @@ func newTestWarpstreamClient(t *testing.T, topic string, numPartitions int32, op func TestWarpstreamClient_ProduceSync(t *testing.T) { const topic = "test-topic" + t.Run("attempted payload equals the wire counters when every produce succeeds", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c, _, _, _ := newTestWarpstreamClient(t, topic, 1) + + // Three records for one partition go out as one merged batch. + results := c.ProduceSync(t.Context(), []*kgo.Record{ + {Topic: topic, Partition: 0, Value: []byte("a"), Timestamp: time.Now()}, + {Topic: topic, Partition: 0, Value: []byte("b"), Timestamp: time.Now()}, + {Topic: topic, Partition: 0, Value: []byte("c"), Timestamp: time.Now()}, + }) + require.Len(t, results, 3) + for _, res := range results { + require.NoError(t, res.Err) + } + + assert.Equal(t, float64(3), testutil.ToFloat64(c.metrics.produceAttemptRecords[attemptPrimary])) + assert.Equal(t, testutil.ToFloat64(c.metrics.produceWireRecordsTotal), testutil.ToFloat64(c.metrics.produceAttemptRecords[attemptPrimary])) + assert.Equal(t, testutil.ToFloat64(c.metrics.produceWireCompressedBytesTotal), testutil.ToFloat64(c.metrics.produceAttemptBytes[attemptPrimary])) + assert.Positive(t, testutil.ToFloat64(c.metrics.produceAttemptBytes[attemptPrimary])) + assert.Zero(t, testutil.ToFloat64(c.metrics.produceAttemptRecords[attemptHedge])) + assert.Zero(t, testutil.ToFloat64(c.metrics.produceAttemptBytes[attemptHedge])) + }) + }) + t.Run("single record produces and is consumable", func(t *testing.T) { synctest.Test(t, func(t *testing.T) { c, _, clusterAddr, vnet := newTestWarpstreamClient(t, topic, 1) @@ -275,6 +299,11 @@ func TestWarpstreamClient_ProduceSync(t *testing.T) { assert.Equal(t, float64(1), testutil.ToFloat64(c.metrics.produceRecordsFailedTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(c.metrics.produceRecordsRejectedTotal.WithLabelValues(produceRejectedNoAgentAssigned))) assert.Equal(t, float64(0), testutil.ToFloat64(c.metrics.produceRecordsRejectedTotal.WithLabelValues(produceRejectedRecordTooLarge))) + + // The caller cancel detaches this wait. The flush keeps its own + // deadline, and a successful background produce is not a Hedger failure. + synctest.Wait() + assertProduceFailures(t, c.metrics, nil) }) }) @@ -1229,6 +1258,13 @@ func TestWarpstreamClient_WriteTimeoutUnblocksStuckProduce(t *testing.T) { // Returned at the write-timeout ceiling, not hanging. assert.GreaterOrEqual(t, elapsed, writeTimeout) assert.Less(t, elapsed, writeTimeout+time.Second) + assertProduceFailures(t, c.metrics, map[produceFailureReason]float64{ + produceFailureReasonWriteTimeout: 1, + }) + // The stuck attempt was dispatched, so it counts as attempted, + // but nothing was acked on the wire. + assert.Equal(t, float64(1), testutil.ToFloat64(c.metrics.produceAttemptRecords[attemptPrimary])) + assert.Zero(t, testutil.ToFloat64(c.metrics.produceWireRecordsTotal)) }) }) @@ -1246,6 +1282,9 @@ func TestWarpstreamClient_WriteTimeoutUnblocksStuckProduce(t *testing.T) { case <-time.After(5 * time.Second): t.Fatal("Produce promise did not fire within the write-timeout bound") } + assertProduceFailures(t, c.metrics, map[produceFailureReason]float64{ + produceFailureReasonWriteTimeout: 1, + }) }) }) }) @@ -2012,6 +2051,76 @@ func TestWarpstreamClient_OnDemandMetadataRefresh(t *testing.T) { }) } +// TestWarpstreamClient_AgentPoolChurnOnLiveRefresh drives real membership +// changes through the periodic refresh and checks the churn counters: the +// constructor refresh and an unchanged refresh add nothing, an added Agent +// counts once as added, and a removed Agent counts once as removed. +func TestWarpstreamClient_AgentPoolChurnOnLiveRefresh(t *testing.T) { + const topic = "test-topic" + + synctest.Test(t, func(t *testing.T) { + reg := prometheus.NewPedanticRegistry() + vnet := &kfake.VirtualNetwork{} + cluster, clusterAddr := testkafka.CreateCluster(t, 1, topic, testkafka.WithVirtualNetwork(vnet)) + c, err := NewWarpstreamClient(nil, reg, append(testWarpstreamOpts(clusterAddr, topic), WithDialer(vnet.DialContext))...) + require.NoError(t, err) + t.Cleanup(c.Close) + + churn := func() (added, removed float64) { + return testutil.ToFloat64(c.metrics.agentpoolAgentsChanged[agentpoolChurnAdded]), + testutil.ToFloat64(c.metrics.agentpoolAgentsChanged[agentpoolChurnRemoved]) + } + refreshOnce := func() { + time.Sleep(10 * time.Second) + synctest.Wait() + } + + // The constructor refresh is not churn. + added, removed := churn() + assert.Zero(t, added) + assert.Zero(t, removed) + + refreshOnce() + added, removed = churn() + assert.Zero(t, added, "an unchanged refresh adds nothing") + assert.Zero(t, removed) + + const newNode = int32(10) + _, _, err = cluster.AddNode(newNode, 9092+5) + require.NoError(t, err) + refreshOnce() + added, removed = churn() + assert.Equal(t, float64(1), added) + assert.Zero(t, removed) + + // The first refresh after RemoveNode may fail while the client still + // holds a connection to the departed node. A failed refresh keeps the + // previous snapshot, so it must not count as churn. Refresh until the + // pool drops the node, then check the removal is counted exactly once. + require.NoError(t, cluster.RemoveNode(newNode)) + for range 5 { + if !slices.Contains(c.pool.Agents(), newNode) { + break + } + refreshOnce() + if slices.Contains(c.pool.Agents(), newNode) { + _, removed = churn() + assert.Zero(t, removed, "no churn while the pool still holds the node") + } + } + require.NotContains(t, c.pool.Agents(), newNode) + added, removed = churn() + assert.Equal(t, float64(1), added) + assert.Equal(t, float64(1), removed) + + // Further unchanged refreshes add nothing. + refreshOnce() + added, removed = churn() + assert.Equal(t, float64(1), added) + assert.Equal(t, float64(1), removed) + }) +} + func TestWarpstreamClient_IdleClusterStats(t *testing.T) { synctest.Test(t, func(t *testing.T) { reg := prometheus.NewPedanticRegistry() @@ -2048,7 +2157,7 @@ func TestWarpstreamClient_NoteLeaderDrops(t *testing.T) { } c.noteLeaderDrops(leaderDrops{}, true) - assert.Equal(t, float64(0), testutil.ToFloat64(c.metrics.agentPoolLeaderDroppedTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(c.metrics.agentPoolExcludedLeaders)) select { case <-c.refreshNowCh: t.Fatal("clean refresh nudged") @@ -2056,21 +2165,31 @@ func TestWarpstreamClient_NoteLeaderDrops(t *testing.T) { } c.noteLeaderDrops(leaderDrops{Count: 2, Topic: "ingest", Partition: 35, NodeID: 99}, true) - assert.Equal(t, float64(2), testutil.ToFloat64(c.metrics.agentPoolLeaderDroppedTotal)) + assert.Equal(t, float64(2), testutil.ToFloat64(c.metrics.agentPoolExcludedLeaders)) select { case <-c.refreshNowCh: default: t.Fatal("refresh that excluded leaders did not nudge") } - // Constructor path counts the drop and does not fill the nudge channel. + // Constructor path publishes the count and does not fill the nudge channel. + // The gauge holds the latest count; it does not accumulate across refreshes. c.noteLeaderDrops(leaderDrops{Count: 1, Topic: "ingest", Partition: 0, NodeID: 7}, false) - assert.Equal(t, float64(3), testutil.ToFloat64(c.metrics.agentPoolLeaderDroppedTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(c.metrics.agentPoolExcludedLeaders)) select { case <-c.refreshNowCh: t.Fatal("constructor refresh nudged") default: } + + // A clean refresh clears the gauge and does not nudge. + c.noteLeaderDrops(leaderDrops{}, true) + assert.Equal(t, float64(0), testutil.ToFloat64(c.metrics.agentPoolExcludedLeaders)) + select { + case <-c.refreshNowCh: + t.Fatal("clean refresh nudged") + default: + } } func TestWarpstreamClient_WaitRefreshCooldown(t *testing.T) { @@ -2175,7 +2294,7 @@ func TestWarpstreamClient_OnDemandRefreshBackoff(t *testing.T) { c.triggerRefresh() synctest.Wait() assertOnDemandRefresh(t, c, 1, 0) - assert.Equal(t, float64(1), testutil.ToFloat64(c.metrics.agentPoolLeaderDroppedTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(c.metrics.agentPoolExcludedLeaders)) assertMissingLeader(t, c, topic) results := c.ProduceSync(t.Context(), []*kgo.Record{ @@ -2192,7 +2311,9 @@ func TestWarpstreamClient_OnDemandRefreshBackoff(t *testing.T) { synctest.Wait() assertOnDemandRefresh(t, c, float64(i+2), 0) } - assert.Equal(t, float64(6), testutil.ToFloat64(c.metrics.agentPoolLeaderDroppedTotal)) + // Six refreshes later the gauge still reads one excluded leader: it is + // state, not a running total, so the backoff does not move it. + assert.Equal(t, float64(1), testutil.ToFloat64(c.metrics.agentPoolExcludedLeaders)) time.Sleep(time.Second) synctest.Wait() @@ -2202,7 +2323,8 @@ func TestWarpstreamClient_OnDemandRefreshBackoff(t *testing.T) { time.Sleep(9 * time.Second) synctest.Wait() assertOnDemandRefresh(t, c, 7, 0) - assert.Equal(t, float64(6), testutil.ToFloat64(c.metrics.agentPoolLeaderDroppedTotal)) + // The clean fetch clears the gauge. + assert.Equal(t, float64(0), testutil.ToFloat64(c.metrics.agentPoolExcludedLeaders)) assertLeaderRestored(t, c, topic, leader) restored := c.ProduceSync(t.Context(), []*kgo.Record{ @@ -2388,6 +2510,71 @@ func TestWarpstreamClient_OnDemandRefreshBackoff(t *testing.T) { }) } +// TestWarpstreamClient_ExcludedLeadersGaugeAcrossRefreshes checks the gauge +// follows the latest successful snapshot: set while a leader is excluded, +// unchanged by a failed refresh, and back to zero once a clean refresh lands. +func TestWarpstreamClient_ExcludedLeadersGaugeAcrossRefreshes(t *testing.T) { + const topic = "test-topic" + + synctest.Test(t, func(t *testing.T) { + vnet := &kfake.VirtualNetwork{} + cluster, clusterAddr := testkafka.CreateCluster(t, 1, topic, + testkafka.WithVirtualNetwork(vnet), testkafka.WithNumBrokers(3)) + c, err := NewWarpstreamClient(nil, prometheus.NewPedanticRegistry(), append( + testWarpstreamOpts(clusterAddr, topic), WithDialer(vnet.DialContext))...) + require.NoError(t, err) + t.Cleanup(c.Close) + + gauge := func() float64 { return testutil.ToFloat64(c.metrics.agentPoolExcludedLeaders) } + // The constructor refresh saw a healthy cluster. + assert.Zero(t, gauge()) + + raw, err := c.Request(t.Context(), kmsg.NewPtrMetadataRequest()) + require.NoError(t, err) + template := raw.(*kmsg.MetadataResponse) + + var poison, fail atomic.Bool + cluster.ControlKey(int16(kmsg.Metadata), func(req kmsg.Request) (kmsg.Response, error, bool) { + cluster.KeepControl() + switch { + case fail.Load(): + // A response error fails the fetch in one round trip. + resp := cloneMetadataResponse(template, req.GetVersion()) + resp.ErrorCode = kerr.UnknownServerError.Code + return resp, nil, true + case poison.Load(): + return metadataWithMissingLeaders(template, 99, req.GetVersion()), nil, true + } + return nil, nil, false + }) + + // A refresh that excludes a leader publishes it. + poison.Store(true) + time.Sleep(time.Nanosecond) + c.triggerRefresh() + synctest.Wait() + assert.Equal(t, float64(1), gauge()) + assertMissingLeader(t, c, topic) + + // A failed refresh keeps the previous snapshot and the previous value. + fail.Store(true) + c.triggerRefresh() + time.Sleep(3 * time.Second) + synctest.Wait() + assert.GreaterOrEqual(t, failedOnDemandRefreshes(c), float64(1)) + assert.Equal(t, float64(1), gauge(), "a failed refresh must not move the gauge") + assertMissingLeader(t, c, topic) + + // A clean refresh clears it. + fail.Store(false) + poison.Store(false) + c.triggerRefresh() + time.Sleep(10 * time.Second) + synctest.Wait() + assert.Zero(t, gauge()) + }) +} + func assertOnDemandRefresh(t *testing.T, c *WarpstreamClient, onDemand, periodic float64) { t.Helper() assert.Equal(t, onDemand, testutil.ToFloat64(c.metrics.metadataRefreshResultsTotal.WithLabelValues( diff --git a/pkg/wgo/encoded_topic_partition.go b/pkg/wgo/encoded_topic_partition.go index 46f2528..7721642 100644 --- a/pkg/wgo/encoded_topic_partition.go +++ b/pkg/wgo/encoded_topic_partition.go @@ -123,6 +123,14 @@ func unrouteEncodedTopicPartitionRecords(parts []routedEncodedTopicPartitionReco return out } +func sumEncodedStats(parts []routedEncodedTopicPartitionRecords) produceRequestStats { + var s produceRequestStats + for _, p := range parts { + s = s.add(p.encodedStats) + } + return s +} + // agentFromRouted resolves the destination Agent for one wire attempt: nodeID, // plus AgentStateDemoted if any partition was routed there as a probe. A mixed // group only arises when demotion flipped mid-linger, and demoted wins so a diff --git a/pkg/wgo/hedger.go b/pkg/wgo/hedger.go index 5a631a3..7133f25 100644 --- a/pkg/wgo/hedger.go +++ b/pkg/wgo/hedger.go @@ -2,6 +2,7 @@ package wgo import ( "context" + "errors" "fmt" "slices" "sync" @@ -24,6 +25,58 @@ type HedgerConfig struct { MaxHedgeAgents int } +type hedgeTrigger int8 + +const ( + hedgeTriggerLatency hedgeTrigger = iota + hedgeTriggerPrimaryFailure + hedgeTriggerDemotedProbe + hedgeTriggerCount +) + +const ( + hedgeTriggerLabelLatency = "latency" + hedgeTriggerLabelPrimaryFailure = "primary_failure" + hedgeTriggerLabelDemotedProbe = "demoted_probe" +) + +// hedgeCause is why shouldHedge allows a proactive race. A cascade that follows +// a primary failure is not proactive, so it has no cause here. +type hedgeCause int8 + +const ( + hedgeCauseLatency hedgeCause = iota + hedgeCauseDemotedProbe +) + +// hedgeDecision is the result of shouldHedge: whether a proactive race may +// start, how long to wait before it, and why. +type hedgeDecision struct { + delay time.Duration + eligible bool + cause hedgeCause +} + +func (d hedgeDecision) trigger() hedgeTrigger { + if d.cause == hedgeCauseDemotedProbe { + return hedgeTriggerDemotedProbe + } + return hedgeTriggerLatency +} + +// produceStop is why a hedge cascade ended. Cancellation is a stop but not a +// failure reason: the produce still returns the cancel error. +type produceStop int8 + +const ( + produceStopNone produceStop = iota + produceStopCandidatesExhausted + produceStopTerminalError + produceStopWriteTimeout + produceStopCanceled + produceStopInternal +) + // Hedger orchestrates produce attempts across multiple agents for the // same batch of partitions. The whole "retry on a different agent, // possibly fire a hedge to race the primary, give up when @@ -38,10 +91,11 @@ type HedgerConfig struct { // 2. Primary returns first with a failure or partial result → call // runHedgingAttempts, which retries the partitions across other // agents. -// 3. delay is 0 (probe) or the hedge timer fires before primary returns -// → call runHedgingAttemptsAndRaceWithPrimary, which runs the -// fallback waves alongside the still-in-flight primary and merges -// whichever finishes first. +// 3. the hedge decision's delay is 0, or the hedge timer fires before the +// primary returns → call runHedgingAttemptsAndRaceWithPrimary, which +// runs the fallback waves alongside the still-in-flight primary and +// merges whichever finishes first. A zero delay does not imply a probe, +// so the trigger is read from the decision. // // Per-partition fanout lives inside runHedgingAttempt: each wave picks // the next per-partition candidate (so partitions whose strategy @@ -52,10 +106,10 @@ type HedgerConfig struct { // - shouldHedge consults the rolling per-agent stats (same window the // Demoter uses via HealthCheckConfig) for primaryID specifically. A // healthy primary still hedges, just with a longer delay (baseline × -// SlowMultiplier). shouldHedge returns hedge=false only when a +// SlowMultiplier). shouldHedge returns eligible=false only when a // suppression gate trips: no agent stats, no cluster stats, or the // cluster slow/faulty fraction is too high to hedge safely. -// - shouldHedge's inlined leading check forces hedge=true delay=0: when +// - shouldHedge's leading check forces delay=0 with cause demoted_probe: when // routing surfaced primaryID as AgentStateDemoted (a probe), we expect // it to fail, so we don't pay the full delay before the fallback. // @@ -100,6 +154,7 @@ func NewHedger(inner DirectProducer, tracker AgentStatsReader, strategy Partitio // covering multiple partitions). The companion primary counter is // incremented in ProduceSync. m.produceRequestsHedgeTotal.Inc() + m.observeAttempt(attemptHedge, sumEncodedStats(parts)) // DirectProducer takes unrouted partitions (the target Agent is specified // separately), so we strip the routing. @@ -124,27 +179,33 @@ func (h *Hedger) ProduceSync(ctx context.Context, primaryID int32, routedPartiti } // All routed partitions in one call must share primaryID — the buffer - // bins by destination nodeID. A mismatch is a routing bug. + // bins by destination nodeID. A mismatch is a routing bug: fail before + // dispatching or consuming a probe, and keep it out of the attempt + // histogram since no attempt ran. for _, p := range routedPartitions { if p.nodeID != primaryID { + h.metrics.produceRequestsFailed[produceFailureReasonInternalError].Inc() return ProduceResult{err: fmt.Errorf("hedger: partition %s/%d routed to nodeID=%d but primaryID=%d", p.topic, p.partition, p.nodeID, primaryID)} } } + decision := h.shouldHedge(time.Now(), primaryID, routedPartitions) + // observeAttempts records the attempt depth of a resolved produce call, // split by outcome: 1 = resolved on the primary, N = resolved after N-1 // hedge waves. - observeAttempts := func(result ProduceResult, attempts int) { + observeAttempts := func(result ProduceResult, attempts int, stop produceStop) { if result.succeeded() { h.metrics.produceRequestsAttemptsSuccess.Observe(float64(attempts)) - } else { - h.metrics.produceRequestsAttemptsFailure.Observe(float64(attempts)) + return + } + h.metrics.produceRequestsAttemptsFailure.Observe(float64(attempts)) + if stop == produceStopNone { + stop = produceStopInternal } + h.observeProduceFailure(stop) } - // Check the hedging delay to apply to this request. - delay, shouldHedge := h.shouldHedge(time.Now(), primaryID, routedPartitions) - // The rest of the Hedger works with unrouted partitions, because it will be // responsible to route partitions to other candidate agents during hedging // and retries. @@ -163,6 +224,7 @@ func (h *Hedger) ProduceSync(ctx context.Context, primaryID int32, routedPartiti primaryCh := make(chan ProduceResult, 1) go func() { h.metrics.produceRequestsPrimaryTotal.Inc() + h.metrics.observeAttempt(attemptPrimary, sumEncodedStats(routedPartitions)) primaryCh <- h.withCoverageCheck(h.inner.ProduceSync(workCtx, agentFromRouted(primaryID, routedPartitions), partitions), primaryID, partitions) }() @@ -172,53 +234,147 @@ func (h *Hedger) ProduceSync(ctx context.Context, primaryID int32, routedPartiti // primary and returns whichever produces a usable outcome first. Shared by // the delay==0 fast path and the timer.C branch below, so a future edit to // this behavior can't land in only one of them. - raceWithPrimary := func() ProduceResult { - hedged := h.runHedgingAttemptsAndRaceWithPrimary(workCtx, primaryID, partitions, candidates, primaryCh) - observeAttempts(hedged.result, hedged.attempts) + raceWithPrimary := func(trigger hedgeTrigger) ProduceResult { + hedged := h.runHedgingAttemptsAndRaceWithPrimary(workCtx, primaryID, partitions, candidates, primaryCh, trigger) + observeAttempts(hedged.result, hedged.attempts, hedged.stop) return hedged.result } - if shouldHedge { + if decision.eligible { // NewTimer(0) is already fired, so a select against a ready // primaryCh is a coin flip. Skip the timer and always start // the fallback race; either leg can still win. - if delay == 0 { - return raceWithPrimary() + if decision.delay == 0 { + return raceWithPrimary(decision.trigger()) } - timer := time.NewTimer(delay) + timer := time.NewTimer(decision.delay) defer timer.Stop() select { case primaryResult := <-primaryCh: if primaryResult.succeeded() { - observeAttempts(primaryResult, 1) + observeAttempts(primaryResult, 1, produceStopNone) return primaryResult } - hedged := h.runHedgingAttempts(workCtx, primaryID, partitions, candidates) + hedged := h.runHedgingAttempts(workCtx, primaryID, partitions, candidates, hedgeTriggerPrimaryFailure) result := selectProduceResult(primaryResult, hedged.result) - observeAttempts(result, hedged.attempts) + if hedged.result.succeeded() { + h.observeHedgeWin(hedgeTriggerPrimaryFailure) + } + observeAttempts(result, hedged.attempts, finalStop(workCtx, primaryResult, result, hedged.stop)) return result case <-timer.C: - return raceWithPrimary() + return raceWithPrimary(hedgeTriggerLatency) } } - // No hedging — wait for the primary synchronously. + // No proactive race — wait for the primary synchronously. A primary + // failure still runs the fallback cascade. primaryResult := <-primaryCh if primaryResult.succeeded() { - observeAttempts(primaryResult, 1) + observeAttempts(primaryResult, 1, produceStopNone) return primaryResult } - // The primary has failed. Try secondaries. - hedged := h.runHedgingAttempts(workCtx, primaryID, partitions, candidates) + hedged := h.runHedgingAttempts(workCtx, primaryID, partitions, candidates, hedgeTriggerPrimaryFailure) result := selectProduceResult(primaryResult, hedged.result) - observeAttempts(result, hedged.attempts) + if hedged.result.succeeded() { + h.observeHedgeWin(hedgeTriggerPrimaryFailure) + } + observeAttempts(result, hedged.attempts, finalStop(workCtx, primaryResult, result, hedged.stop)) return result } +// observeHedgeWin counts a cascade whose result was the one returned. It runs +// in the owning invocation, after the winner is chosen, because a fallback that +// finishes successfully can still lose the race to the primary. +func (h *Hedger) observeHedgeWin(trigger hedgeTrigger) { + h.metrics.hedgeWinsTotal.Inc() + h.metrics.hedgeTriggerWins[trigger].Inc() +} + +func (h *Hedger) observeProduceFailure(stop produceStop) { + var reason produceFailureReason + switch stop { + case produceStopNone, produceStopCanceled: + return + case produceStopCandidatesExhausted: + reason = produceFailureReasonCandidatesExhausted + case produceStopTerminalError: + reason = produceFailureReasonTerminalError + case produceStopWriteTimeout: + reason = produceFailureReasonWriteTimeout + default: + reason = produceFailureReasonInternalError + } + h.metrics.produceRequestsFailed[reason].Inc() +} + +// classifyProduceStop picks why a cascade ended. A terminal broker or unknown +// error wins over a context that expires afterward, explicit cancellation is +// not a failure reason, and an expired deadline outranks candidate +// exhaustion. +func classifyProduceStop(terminal bool, terminalErr, ctxErr error, exhausted bool) produceStop { + if terminal && !errors.Is(terminalErr, context.Canceled) { + return produceStopTerminalError + } + if ctxErr != nil && errors.Is(ctxErr, context.Canceled) { + return produceStopCanceled + } + if ctxErr != nil && errors.Is(ctxErr, context.DeadlineExceeded) { + return produceStopWriteTimeout + } + if exhausted { + return produceStopCandidatesExhausted + } + if terminal { + return produceStopTerminalError + } + return produceStopNone +} + +// finalStop settles why a failed produce stopped once the primary's result is +// known. The cascade only sees its own attempts, so a terminal primary error +// that no other agent could work around is the cause even when the cascade +// stopped as exhausted or, with the work context already done, as a write +// timeout. A work context that ended while the primary was still running is +// also the cause of an exhausted stop. Other stops are already settled; a +// canceled stop remains omitted. +func finalStop(workCtx context.Context, primary, result ProduceResult, stop produceStop) produceStop { + if (stop != produceStopCandidatesExhausted && stop != produceStopWriteTimeout) || result.succeeded() { + return stop + } + if err := primary.error(); err != nil && !errors.Is(err, context.Canceled) && !getProduceResultErr(err).retriable { + return produceStopTerminalError + } + if stop == produceStopWriteTimeout { + return stop + } + switch err := ctxStopErr(workCtx); { + case errors.Is(err, context.DeadlineExceeded): + return produceStopWriteTimeout + case errors.Is(err, context.Canceled): + return produceStopCanceled + } + return stop +} + +// ctxStopErr is ctx.Err(), except that a deadline already reached counts even +// if its timer has not fired yet. The write timeout and the per-attempt deadline +// are equal by default, so a timed-out attempt can return just before the work +// context reports its own expiry. +func ctxStopErr(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + if deadline, ok := ctx.Deadline(); ok && !time.Now().Before(deadline) { + return context.DeadlineExceeded + } + return nil +} + func (h *Hedger) withCoverageCheck(res ProduceResult, nodeID int32, requested []encodedTopicPartitionRecords) ProduceResult { if res.err != nil { return res @@ -238,10 +394,10 @@ func (h *Hedger) withCoverageCheck(res ProduceResult, nodeID int32, requested [] // cancel as soon as this function returns, which unwinds the losing leg. // The reported attempts is the depth of the winning leg: 1 when the // primary wins, the fallback's own depth when the fallback wins. -func (h *Hedger) runHedgingAttemptsAndRaceWithPrimary(workCtx context.Context, primaryID int32, partitions []encodedTopicPartitionRecords, candidates *hedgerCandidates, primaryCh <-chan ProduceResult) hedgerProduceResult { +func (h *Hedger) runHedgingAttemptsAndRaceWithPrimary(workCtx context.Context, primaryID int32, partitions []encodedTopicPartitionRecords, candidates *hedgerCandidates, primaryCh <-chan ProduceResult, trigger hedgeTrigger) hedgerProduceResult { fallbackCh := make(chan hedgerProduceResult, 1) go func() { - fallbackCh <- h.runHedgingAttempts(workCtx, primaryID, partitions, candidates) + fallbackCh <- h.runHedgingAttempts(workCtx, primaryID, partitions, candidates, trigger) }() // Either side's "wait for the other" branch is bounded: both legs @@ -259,9 +415,18 @@ func (h *Hedger) runHedgingAttemptsAndRaceWithPrimary(workCtx context.Context, p // Primary failed; the fallback's view supersedes primary's // (per-partition errors are equivalent or richer there). fb := <-fallbackCh - return hedgerProduceResult{result: selectProduceResult(primaryResult, fb.result), attempts: fb.attempts} + if fb.result.succeeded() { + h.observeHedgeWin(trigger) + } + result := selectProduceResult(primaryResult, fb.result) + return hedgerProduceResult{ + result: result, + attempts: fb.attempts, + stop: finalStop(workCtx, primaryResult, result, fb.stop), + } case fb := <-fallbackCh: if fb.result.succeeded() { + h.observeHedgeWin(trigger) return fb } // Fallback failed (e.g. exhausted candidates) — the primary may @@ -269,7 +434,9 @@ func (h *Hedger) runHedgingAttemptsAndRaceWithPrimary(workCtx context.Context, p // then succeeds it won on its single attempt (depth 1). When the // fallback failed too the depth-1 label is best-effort: the // fallback's wave count isn't surfaced on this branch. - return hedgerProduceResult{result: selectProduceResult(<-primaryCh, fb.result), attempts: 1} + primaryResult := <-primaryCh + result := selectProduceResult(primaryResult, fb.result) + return hedgerProduceResult{result: result, attempts: 1, stop: finalStop(workCtx, primaryResult, result, fb.stop)} } } @@ -278,16 +445,9 @@ func (h *Hedger) runHedgingAttemptsAndRaceWithPrimary(workCtx context.Context, p // MaxHedgeAgents, or ctx is canceled. The reported attempts is the total // attempt depth: 1 (the primary, which this function models as the first // tried agent) plus one per hedge wave dispatched. -func (h *Hedger) runHedgingAttempts(workCtx context.Context, primaryID int32, partitions []encodedTopicPartitionRecords, candidates *hedgerCandidates) (out hedgerProduceResult) { +func (h *Hedger) runHedgingAttempts(workCtx context.Context, primaryID int32, partitions []encodedTopicPartitionRecords, candidates *hedgerCandidates, trigger hedgeTrigger) (out hedgerProduceResult) { h.metrics.hedgeAttemptsTotal.Inc() - // Count a win only when the result is successful AND we weren't - // preempted by ctx cancellation (e.g. the racing variant aborting - // the fallback because the primary won). - defer func() { - if out.result.succeeded() && workCtx.Err() == nil { - h.metrics.hedgeWinsTotal.Inc() - } - }() + h.metrics.hedgeTriggers[trigger].Inc() acc, err := newProduceResultAccumulator(partitions) if err != nil { @@ -295,6 +455,7 @@ func (h *Hedger) runHedgingAttempts(workCtx context.Context, primaryID int32, pa // cluster buffer) violated its bin-by-partition invariant. // Surface so the bug isn't silenced. out.result = ProduceResult{err: err} + out.stop = produceStopInternal return out } @@ -309,23 +470,30 @@ func (h *Hedger) runHedgingAttempts(workCtx context.Context, primaryID int32, pa // The primary is attempt 1; each dispatched hedge wave adds one. A wave // that bails before dispatching (candidates exhausted) must not count. out.attempts = 1 + exhausted := false for workCtx.Err() == nil { - dispatched, canRetry := h.runHedgingAttempt(workCtx, acc, tried, candidates) + dispatched, canRetry, waveExhausted := h.runHedgingAttempt(workCtx, acc, tried, candidates) if dispatched { out.attempts++ } if !canRetry { + exhausted = waveExhausted break } } // If the caller's ctx was canceled mid-loop, surface that directly: // the work was preempted, not exhausted. We still want the merged - // per-partition state, so we keep acc.result()'s resp. + // per-partition state, so we keep acc.result()'s resp. The stop is + // classified from the accumulator and the context separately so a + // deadline that expires after an abort doesn't hide the abort's cause. out.result = acc.result() - if ctxErr := workCtx.Err(); ctxErr != nil { + ctxErr := workCtx.Err() + if ctxErr != nil { out.result.err = ctxErr } + terminal, terminalErr := acc.terminalErr() + out.stop = classifyProduceStop(terminal, terminalErr, ctxStopErr(workCtx), exhausted) return out } @@ -335,11 +503,12 @@ func (h *Hedger) runHedgingAttempts(workCtx context.Context, primaryID int32, pa // pending or a partition exhausted its candidates). canRetry reports // whether the caller should iterate again — false when no further work // would change the outcome (every partition resolved, accumulator aborted, -// a partition exhausted candidates, or ctx canceled mid-wave). -func (h *Hedger) runHedgingAttempt(workCtx context.Context, acc *produceResultAccumulator, tried map[topicPartition][]int32, candidates *hedgerCandidates) (dispatched, canRetry bool) { +// a partition exhausted candidates, or ctx canceled mid-wave). exhausted is +// set only for the candidate-budget stop. +func (h *Hedger) runHedgingAttempt(workCtx context.Context, acc *produceResultAccumulator, tried map[topicPartition][]int32, candidates *hedgerCandidates) (dispatched, canRetry, exhausted bool) { pending := acc.remaining() if len(pending) == 0 { - return false, false + return false, false, false } // Pick the next candidate per partition; group by chosen agent. @@ -357,7 +526,7 @@ func (h *Hedger) runHedgingAttempt(workCtx context.Context, acc *produceResultAc // termination even if a strategy returns more entries (or // duplicates) than MaxHedgeAgents. if len(tried[tp]) >= h.cfg.MaxHedgeAgents { - return false, false + return false, false, true } var ( @@ -374,7 +543,7 @@ func (h *Hedger) runHedgingAttempt(workCtx context.Context, acc *produceResultAc // We exhausted all candidates for this partition. We give up. if !found { - return false, false + return false, false, true } tried[tp] = append(tried[tp], next) @@ -414,35 +583,36 @@ func (h *Hedger) runHedgingAttempt(workCtx context.Context, acc *produceResultAc // Either every partition resolved or a non-retriable // err aborted the batch — either way, no further wave // would change the outcome. - return true, false + return true, false, false } case <-attemptCtx.Done(): - return true, false + return true, false, false } } - return true, true + return true, true, false } -// shouldHedge returns the hedge delay (and whether to hedge) for primaryID. -func (h *Hedger) shouldHedge(now time.Time, primaryID int32, partitions []routedEncodedTopicPartitionRecords) (time.Duration, bool) { +// shouldHedge returns the hedge decision for primaryID, reading the cause from +// the routing-time AgentState instead of re-querying Candidates. +func (h *Hedger) shouldHedge(now time.Time, primaryID int32, partitions []routedEncodedTopicPartitionRecords) hedgeDecision { // Probe routing: trust the routing-time nodeState. Re-querying the // strategy would consume another probe slot through the Demoter. for _, p := range partitions { if p.nodeID == primaryID && p.nodeState == AgentStateDemoted { - return 0, true + return hedgeDecision{eligible: true, cause: hedgeCauseDemotedProbe} } } primary, ok := h.tracker.AgentStats(now, primaryID) if !ok { h.metrics.hedgeAttemptsSuppressedTotal.WithLabelValues(hedgeSuppressedNoAgentStats).Inc() - return 0, false + return hedgeDecision{} } clusterStats, hasClusterStats := h.tracker.ClusterStats(now, h.health.SlowMultiplier, h.health.FaultyThreshold) if !hasClusterStats { h.metrics.hedgeAttemptsSuppressedTotal.WithLabelValues(hedgeSuppressedNoClusterStats).Inc() - return 0, false + return hedgeDecision{} } // Apply a scale-aware floor of 1/N to each fraction gate so a single @@ -454,11 +624,11 @@ func (h *Hedger) shouldHedge(now time.Time, primaryID int32, partitions []routed // (1/N is tiny) so this is a no-op in production-sized clusters. if clusterStats.SlowFraction > maxFractionFloor(h.health.MaxSlowFraction, clusterStats.SlowContributorsCount) { h.metrics.hedgeAttemptsSuppressedTotal.WithLabelValues(hedgeSuppressedSlowFraction).Inc() - return 0, false + return hedgeDecision{} } if clusterStats.FaultyFraction > maxFractionFloor(h.health.MaxFaultyFraction, clusterStats.FaultyContributorsCount) { h.metrics.hedgeAttemptsSuppressedTotal.WithLabelValues(hedgeSuppressedFaultyFraction).Inc() - return 0, false + return hedgeDecision{} } primarySlow := primary.Latency > clusterStats.SlowThreshold @@ -477,7 +647,11 @@ func (h *Hedger) shouldHedge(now time.Time, primaryID int32, partitions []routed baseline = time.Duration(float64(baseline) * h.health.SlowMultiplier) } - return max(baseline, h.cfg.MinHedgeDelay), true + return hedgeDecision{ + delay: max(baseline, h.cfg.MinHedgeDelay), + eligible: true, + cause: hedgeCauseLatency, + } } // hedgerProduceResult bundles a hedge cascade's result with its attempt @@ -487,6 +661,9 @@ func (h *Hedger) shouldHedge(now time.Time, primaryID int32, partitions []routed type hedgerProduceResult struct { result ProduceResult attempts int + // stop is this cascade's own cause. It can be set even when the merged + // result succeeds through the primary, so read it only for a failed result. + stop produceStop } // hedgerCandidates memoizes strategy.Candidates() per (topic, partition) diff --git a/pkg/wgo/hedger_test.go b/pkg/wgo/hedger_test.go index ad99dac..6b2c26a 100644 --- a/pkg/wgo/hedger_test.go +++ b/pkg/wgo/hedger_test.go @@ -28,6 +28,47 @@ func histogramCountSum(t *testing.T, h prometheus.Histogram) (uint64, float64) { return m.GetHistogram().GetSampleCount(), m.GetHistogram().GetSampleSum() } +// assertProduceFailures checks every failure-reason series. Reasons omitted from +// want are asserted at zero, so a nil want means the invocation emitted none. +func assertProduceFailures(t *testing.T, m *metrics, want map[produceFailureReason]float64) { + t.Helper() + reasons := []produceFailureReason{ + produceFailureReasonCandidatesExhausted, + produceFailureReasonTerminalError, + produceFailureReasonWriteTimeout, + produceFailureReasonInternalError, + } + for _, reason := range reasons { + assert.Equal(t, want[reason], testutil.ToFloat64(m.produceRequestsFailed[reason]), "failure reason %d", reason) + } +} + +// assertTriggerWins checks every hedge-trigger win series. Triggers omitted +// from want are asserted at zero. It also checks the two invariants of the +// family: the series sum to hedgeWinsTotal, and a trigger never wins more +// cascades than it started. +func assertTriggerWins(t *testing.T, m *metrics, want map[hedgeTrigger]float64) { + t.Helper() + var sum float64 + for trigger := hedgeTrigger(0); trigger < hedgeTriggerCount; trigger++ { + got := testutil.ToFloat64(m.hedgeTriggerWins[trigger]) + assert.Equal(t, want[trigger], got, "trigger %d wins", trigger) + assert.LessOrEqual(t, got, testutil.ToFloat64(m.hedgeTriggers[trigger]), "trigger %d wins exceed its triggers", trigger) + sum += got + } + assert.Equal(t, testutil.ToFloat64(m.hedgeWinsTotal), sum, "trigger wins must sum to hedgeWinsTotal") +} + +// assertAttemptPayload checks the attempted-payload counters of both roles +// against the producer-state counts the test expects each role to have sent. +func assertAttemptPayload(t *testing.T, m *metrics, primary, hedge produceRequestStats) { + t.Helper() + for role, want := range map[attemptRole]produceRequestStats{attemptPrimary: primary, attemptHedge: hedge} { + assert.Equal(t, float64(want.records), testutil.ToFloat64(m.produceAttemptRecords[role]), "role %d records", role) + assert.Equal(t, float64(want.compressedBytes), testutil.ToFloat64(m.produceAttemptBytes[role]), "role %d bytes", role) + } +} + // hedgerResult captures one partition's terminal outcome as fired by the // Hedger's per-partition done callback. type hedgerResult struct { @@ -202,6 +243,9 @@ func TestHedger_ProduceSync(t *testing.T) { }, nil } + // reqStats is the payload makeReq sends; makeReq encodes deterministically. + reqStats := sumEncodedStats(unpromise(makeReq(newResultCapture()))) + t.Run("no hedge decision: only primary is called", func(t *testing.T) { producer := newMockDirectProducer() producer.respFn = successResp @@ -218,10 +262,12 @@ func TestHedger_ProduceSync(t *testing.T) { assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assertProduceFailures(t, m, nil) // Primary won → one success observation of attempt depth 1. count, sum := histogramCountSum(t, m.produceRequestsAttemptsSuccess.(prometheus.Histogram)) assert.Equal(t, uint64(1), count) assert.Equal(t, float64(1), sum) + assertAttemptPayload(t, m, reqStats, produceRequestStats{}) }) t.Run("empty input is trivially successful", func(t *testing.T) { @@ -235,6 +281,13 @@ func TestHedger_ProduceSync(t *testing.T) { assert.True(t, res.succeeded()) assert.NoError(t, res.error()) assert.Empty(t, producer.recordedCalls()) + assertProduceFailures(t, m, nil) + successCount, _ := histogramCountSum(t, m.produceRequestsAttemptsSuccess.(prometheus.Histogram)) + failureCount, _ := histogramCountSum(t, m.produceRequestsAttemptsFailure.(prometheus.Histogram)) + assert.Equal(t, uint64(0), successCount) + assert.Equal(t, uint64(0), failureCount) + // Nothing was dispatched, so nothing was attempted. + assertAttemptPayload(t, m, produceRequestStats{}, produceRequestStats{}) }) t.Run("hedging suppressed when primary has no agent stats yet", func(t *testing.T) { @@ -279,8 +332,16 @@ func TestHedger_ProduceSync(t *testing.T) { } assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeWinsTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerDemotedProbe])) assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) assert.GreaterOrEqual(t, testutil.ToFloat64(m.produceRequestsHedgeTotal), float64(1)) + // The fallback won the race; the canceled primary leg emits nothing. + assertProduceFailures(t, m, nil) + assertTriggerWins(t, m, map[hedgeTrigger]float64{hedgeTriggerLatency: 1}) + // The blocked primary was dispatched and lost, and still counts. + assertAttemptPayload(t, m, reqStats, reqStats) }) t.Run("primary fails with retriable error: cascade to secondary", func(t *testing.T) { @@ -300,12 +361,20 @@ func TestHedger_ProduceSync(t *testing.T) { assert.Contains(t, callIDs, secondaryID) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeWinsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerDemotedProbe])) assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) assert.GreaterOrEqual(t, testutil.ToFloat64(m.produceRequestsHedgeTotal), float64(1)) // Primary failed, one hedge wave resolved it → success attempt depth 2. count, sum := histogramCountSum(t, m.produceRequestsAttemptsSuccess.(prometheus.Histogram)) assert.Equal(t, uint64(1), count) assert.Equal(t, float64(2), sum) + // A failed primary that the cascade recovered is not a final failure. + assertProduceFailures(t, m, nil) + assertTriggerWins(t, m, map[hedgeTrigger]float64{hedgeTriggerPrimaryFailure: 1}) + // The failed primary and the retry both count. + assertAttemptPayload(t, m, reqStats, reqStats) }) t.Run("hedge timer fires and secondary wins: hedgeWinsTotal incremented", func(t *testing.T) { @@ -330,6 +399,9 @@ func TestHedger_ProduceSync(t *testing.T) { }, time.Second, 10*time.Millisecond) assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) assert.GreaterOrEqual(t, testutil.ToFloat64(m.produceRequestsHedgeTotal), float64(1)) + assertProduceFailures(t, m, nil) + assertTriggerWins(t, m, map[hedgeTrigger]float64{hedgeTriggerLatency: 1}) + assertAttemptPayload(t, m, reqStats, reqStats) }) t.Run("hedge timer fires but primary wins the race: hedgeAttemptsTotal++ but no hedgeWinsTotal", func(t *testing.T) { @@ -367,6 +439,12 @@ func TestHedger_ProduceSync(t *testing.T) { assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) assert.GreaterOrEqual(t, testutil.ToFloat64(m.produceRequestsHedgeTotal), float64(1)) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assertProduceFailures(t, m, nil) + // The cascade started but lost: a trigger without a win. + assertTriggerWins(t, m, nil) + // The losing hedge leg was dispatched before the primary won. + assertAttemptPayload(t, m, reqStats, reqStats) }) t.Run("hedge decision but primary wins before timer: no hedgeAttemptsTotal increment", func(t *testing.T) { @@ -381,8 +459,12 @@ func TestHedger_ProduceSync(t *testing.T) { require.NoError(t, capture.get(topic, partition).err) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerDemotedProbe])) assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assertAttemptPayload(t, m, reqStats, produceRequestStats{}) }) t.Run("per-partition fanout: each partition resolves independently", func(t *testing.T) { @@ -429,6 +511,12 @@ func TestHedger_ProduceSync(t *testing.T) { assert.Contains(t, callIDs, secondaryB) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeWinsTotal)) + assertProduceFailures(t, m, nil) + assertTriggerWins(t, m, map[hedgeTrigger]float64{hedgeTriggerPrimaryFailure: 1}) + // Two partitions go out together on the primary, then to two different + // fallback agents: the hedge total is the same payload split across flushes. + both := sumEncodedStats(unpromise(req)) + assertAttemptPayload(t, m, both, both) }) t.Run("single-agent cluster (no secondaries): primary error propagates", func(t *testing.T) { @@ -452,11 +540,58 @@ func TestHedger_ProduceSync(t *testing.T) { // fallback candidate. assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonCandidatesExhausted: 1, + }) // No hedge candidate → only the primary attempt was made before // giving up, recorded under the failure outcome. count, sum := histogramCountSum(t, m.produceRequestsAttemptsFailure.(prometheus.Histogram)) assert.Equal(t, uint64(1), count) assert.Equal(t, float64(1), sum) + // The cascade dispatched nothing, so it adds no hedge payload. + assertAttemptPayload(t, m, reqStats, produceRequestStats{}) + }) + + t.Run("hedging suppressed and primary fails: cascade is classified as primary failure", func(t *testing.T) { + producer := newMockDirectProducer() + producer.respFn = successResp + producer.errs[primaryID] = kerr.RequestTimedOut + + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, NewAverageAgentStatsTracker(), stratPrimaryAndSecondary, health, cfg, 0, 1<<20, m, nil) + + capture := newResultCapture() + runHedger(h, context.Background(), makeReq(capture)) + + require.NoError(t, capture.get(topic, partition).err) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerDemotedProbe])) + assertProduceFailures(t, m, nil) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsSuppressedTotal.WithLabelValues(hedgeSuppressedNoAgentStats))) + }) + + t.Run("hedging suppressed and primary fails with no secondary: candidates exhausted", func(t *testing.T) { + emptyStrat := &mockPartitionAssignmentStrategy{ + candidates: map[partitionKey][]Agent{{topic, partition}: healthyAgents(primaryID)}, + } + producer := newMockDirectProducer() + producer.errs[primaryID] = kerr.RequestTimedOut + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, NewAverageAgentStatsTracker(), emptyStrat, health, cfg, 0, 1<<20, m, nil) + + capture := newResultCapture() + runHedger(h, context.Background(), makeReq(capture)) + + require.Error(t, capture.get(topic, partition).err) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsSuppressedTotal.WithLabelValues(hedgeSuppressedNoAgentStats))) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonCandidatesExhausted: 1, + }) }) t.Run("partial fallback coverage: any partition exhausting candidates stops the whole attempt", func(t *testing.T) { @@ -505,6 +640,10 @@ func TestHedger_ProduceSync(t *testing.T) { assert.NotContains(t, callIDs, secondaryID) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonCandidatesExhausted: 1, + }) }) t.Run("primary marked as probe (demoted): hedge fires immediately, ignoring hedge delay", func(t *testing.T) { @@ -555,6 +694,10 @@ func TestHedger_ProduceSync(t *testing.T) { assert.True(t, sawFallback) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeWinsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerDemotedProbe])) + assertTriggerWins(t, m, map[hedgeTrigger]float64{hedgeTriggerDemotedProbe: 1}) }) t.Run("demoted primary is the only candidate: hedge fires but finds no fallback; primary's success wins", func(t *testing.T) { @@ -584,9 +727,12 @@ func TestHedger_ProduceSync(t *testing.T) { // hedgeAttemptsTotal is incremented on the fallback goroutine, which // may not have run yet when the primary already won the race. assert.Eventually(t, func() bool { - return testutil.ToFloat64(m.hedgeAttemptsTotal) == 1 + return testutil.ToFloat64(m.hedgeAttemptsTotal) == 1 && + testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerDemotedProbe]) == 1 }, time.Second, 10*time.Millisecond) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assertProduceFailures(t, m, nil) // Primary fires once; the hedge cascade has no candidates to try // so no hedge wire request is issued. assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) @@ -622,6 +768,10 @@ func TestHedger_ProduceSync(t *testing.T) { require.NoError(t, capture.get(topic, partition).err, "primary's slow but successful probe must be honoured even when the racing-variant fallback returned an empty result") + // The fallback's exhausted stop is dropped because the primary then succeeded. + assertProduceFailures(t, m, nil) + // The probe cascade found nothing to win with; the primary did. + assertTriggerWins(t, m, nil) }) t.Run("both legs fail with retriable errors: surfaces the last error", func(t *testing.T) { @@ -646,6 +796,11 @@ func TestHedger_ProduceSync(t *testing.T) { assert.Len(t, producer.recordedCalls(), 2) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonCandidatesExhausted: 1, + }) + assertTriggerWins(t, m, nil) }) t.Run("non-retriable fallback err aborts: no further waves even with candidates left", func(t *testing.T) { @@ -692,6 +847,9 @@ func TestHedger_ProduceSync(t *testing.T) { assert.Zero(t, seen[agentD]) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonTerminalError: 1, + }) }) t.Run("primary and secondary legs share the tried set: no agent is attempted twice", func(t *testing.T) { @@ -814,6 +972,11 @@ func TestHedger_ProduceSync(t *testing.T) { "got %v", err) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + // Both legs failed: exactly one failure reason, taken from the fallback's stop. + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonCandidatesExhausted: 1, + }) + assertTriggerWins(t, m, nil) }) t.Run("racing path: fallback exhausts before primary returns → fallback kerr preserved", func(t *testing.T) { @@ -840,6 +1003,123 @@ func TestHedger_ProduceSync(t *testing.T) { assert.ErrorIs(t, err, kerr.NotLeaderForPartition, "got %v", err) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonCandidatesExhausted: 1, + }) + assertTriggerWins(t, m, nil) + }) + + t.Run("primary fails non-retriably and the candidate budget is one: terminal error", func(t *testing.T) { + producer := newMockDirectProducer() + producer.errs[primaryID] = kerr.MessageTooLarge + one := cfg + one.MaxHedgeAgents = 1 + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, healthyTracker(), stratPrimaryAndSecondary, health, one, 0, 1<<20, m, nil) + + capture := newResultCapture() + runHedger(h, context.Background(), makeReq(capture)) + + require.Error(t, capture.get(topic, partition).err) + assert.Equal(t, []int32{primaryID}, producer.recordedCallNodeIDs()) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonTerminalError: 1, + }) + }) + + t.Run("primary fails non-retriably and it is the only candidate: terminal error", func(t *testing.T) { + onlyPrimary := &mockPartitionAssignmentStrategy{ + candidates: map[partitionKey][]Agent{{topic, partition}: healthyAgents(primaryID)}, + } + producer := newMockDirectProducer() + producer.errs[primaryID] = kerr.MessageTooLarge + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, healthyTracker(), onlyPrimary, health, cfg, 0, 1<<20, m, nil) + + capture := newResultCapture() + runHedger(h, context.Background(), makeReq(capture)) + + require.Error(t, capture.get(topic, partition).err) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonTerminalError: 1, + }) + }) + + t.Run("racing path: fallback exhausts then the primary fails non-retriably: terminal error", func(t *testing.T) { + producer := newMockDirectProducer() + producer.delays[primaryID] = 50 * time.Millisecond + producer.errs[primaryID] = kerr.MessageTooLarge + producer.errs[secondaryID] = kerr.NotLeaderForPartition + + raceCfg := cfg + raceCfg.MinHedgeDelay = time.Millisecond + raceCfg.MaxHedgeAgents = 2 + + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, slowPrimaryTracker(), stratPrimaryAndSecondary, health, raceCfg, 0, 1<<20, m, nil) + + capture := newResultCapture() + runHedger(h, context.Background(), makeReq(capture)) + + require.Error(t, capture.get(topic, partition).err) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonTerminalError: 1, + }) + assertTriggerWins(t, m, nil) + }) + + t.Run("racing path: fallback exhausts then the primary hits the work deadline is a write timeout", func(t *testing.T) { + // The fallback tries the one secondary and exhausts quickly, leaving + // the primary to run out the clock. The deadline is why this failed. + producer := newMockDirectProducer() + producer.delays[primaryID] = 10 * time.Second + producer.errs[secondaryID] = kerr.NotLeaderForPartition + + raceCfg := cfg + raceCfg.MinHedgeDelay = time.Millisecond + raceCfg.MaxHedgeAgents = 2 + + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, slowPrimaryTracker(), stratPrimaryAndSecondary, health, raceCfg, 0, 1<<20, m, nil) + + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + capture := newResultCapture() + runHedger(h, ctx, makeReq(capture)) + + require.Error(t, capture.get(topic, partition).err) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonWriteTimeout: 1, + }) + assertTriggerWins(t, m, nil) + }) + + t.Run("racing path: fallback exhausts then the caller cancels while the primary runs: no failure counted", func(t *testing.T) { + producer := newMockDirectProducer() + producer.delays[primaryID] = 10 * time.Second + producer.errs[secondaryID] = kerr.NotLeaderForPartition + + raceCfg := cfg + raceCfg.MinHedgeDelay = time.Millisecond + raceCfg.MaxHedgeAgents = 2 + + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, slowPrimaryTracker(), stratPrimaryAndSecondary, health, raceCfg, 0, 1<<20, m, nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // The secondary fails within milliseconds, so the cascade has + // exhausted its candidates well before the cancel. + time.AfterFunc(150*time.Millisecond, cancel) + capture := newResultCapture() + runHedger(h, ctx, makeReq(capture)) + + // The caller still sees the fallback's last error from the merged response. + require.Error(t, capture.get(topic, partition).err) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assertProduceFailures(t, m, nil) + assertTriggerWins(t, m, nil) }) t.Run("ctx canceled mid-fallback: result surfaces ctx err, not synthesized timeout", func(t *testing.T) { @@ -872,6 +1152,33 @@ func TestHedger_ProduceSync(t *testing.T) { assert.ErrorIs(t, err, context.Canceled) assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + assertProduceFailures(t, m, nil) + }) + + t.Run("ctx deadline mid-fallback: failure reason is write timeout", func(t *testing.T) { + producer := newMockDirectProducer() + producer.errs[primaryID] = kerr.RequestTimedOut + gate := make(chan struct{}) + producer.blockCh[secondaryID] = gate + t.Cleanup(func() { close(gate) }) + + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, healthyTracker(), stratPrimaryAndSecondary, health, cfg, 0, 1<<20, m, nil) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + err := h.ProduceSync(ctx, primaryID, []routedEncodedTopicPartitionRecords{{ + encodedTopicPartitionRecords: newEncodedTopicPartitionRecords(topic, partition, []*kgo.Record{{Topic: topic, Partition: partition}}), + nodeID: primaryID, + }}).error() + require.Error(t, err) + assert.ErrorIs(t, err, context.DeadlineExceeded) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeWinsTotal)) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonWriteTimeout: 1, + }) }) t.Run("fallback wins: primary leg is canceled via workCtx instead of running until its per-attempt deadline", func(t *testing.T) { @@ -910,6 +1217,9 @@ func TestHedger_ProduceSync(t *testing.T) { } return false }, time.Second, 10*time.Millisecond, "primary leg was not canceled after fallback won") + // The canceled primary leg is a loser, not a failed produce. + assertProduceFailures(t, m, nil) + assertTriggerWins(t, m, map[hedgeTrigger]float64{hedgeTriggerDemotedProbe: 1}) }) t.Run("returns error when a routed partition's nodeID disagrees with primaryID", func(t *testing.T) { @@ -931,6 +1241,126 @@ func TestHedger_ProduceSync(t *testing.T) { require.Error(t, err) assert.Contains(t, err.Error(), "primaryID=1") assert.Empty(t, producer.recordedCalls()) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonInternalError: 1, + }) + successCount, _ := histogramCountSum(t, m.produceRequestsAttemptsSuccess.(prometheus.Histogram)) + failureCount, _ := histogramCountSum(t, m.produceRequestsAttemptsFailure.(prometheus.Histogram)) + assert.Equal(t, uint64(0), successCount) + assert.Equal(t, uint64(0), failureCount) + // Nothing was dispatched, so nothing was attempted. + assertAttemptPayload(t, m, produceRequestStats{}, produceRequestStats{}) + }) + + t.Run("healthy primary with zero computed delay is a latency trigger", func(t *testing.T) { + // MinHedgeDelay and the baseline are both zero, so the fast path + // races immediately. That delay is not a demoted probe. + tr := NewAverageAgentStatsTracker() + nowNs := time.Now().UnixNano() + for _, id := range []int32{primaryID, secondaryID} { + seedFullWindow(tr, id, nowNs, 20, 0, 0) + } + producer := newMockDirectProducer() + producer.respFn = successResp + gate := make(chan struct{}) + producer.blockCh[primaryID] = gate + t.Cleanup(func() { close(gate) }) + + zeroCfg := cfg + zeroCfg.MinHedgeDelay = 0 + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, tr, stratPrimaryAndSecondary, health, zeroCfg, 0, 1<<20, m, nil) + + capture := newResultCapture() + runHedger(h, context.Background(), makeReq(capture)) + require.NoError(t, capture.get(topic, partition).err) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerDemotedProbe])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.GreaterOrEqual(t, testutil.ToFloat64(m.produceRequestsHedgeTotal), float64(1)) + assertProduceFailures(t, m, nil) + }) + + t.Run("max hedge agents of one exhausts before any fallback dispatch", func(t *testing.T) { + producer := newMockDirectProducer() + producer.errs[primaryID] = kerr.RequestTimedOut + one := cfg + one.MaxHedgeAgents = 1 + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, healthyTracker(), stratPrimaryAndSecondary, health, one, 0, 1<<20, m, nil) + + capture := newResultCapture() + runHedger(h, context.Background(), makeReq(capture)) + require.Error(t, capture.get(topic, partition).err) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assert.Equal(t, []int32{primaryID}, producer.recordedCallNodeIDs()) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonCandidatesExhausted: 1, + }) + assertAttemptPayload(t, m, reqStats, produceRequestStats{}) + }) + + t.Run("canceled before fallback dispatch counts the cascade and emits no failure", func(t *testing.T) { + producer := newMockDirectProducer() + producer.blockCh[primaryID] = make(chan struct{}) + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, healthyTracker(), stratPrimaryAndSecondary, health, cfg, 0, 1<<20, m, nil) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := h.ProduceSync(ctx, primaryID, []routedEncodedTopicPartitionRecords{{ + encodedTopicPartitionRecords: newEncodedTopicPartitionRecords(topic, partition, []*kgo.Record{{Topic: topic, Partition: partition}}), + nodeID: primaryID, + }}).error() + require.ErrorIs(t, err, context.Canceled) + assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerPrimaryFailure])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.hedgeTriggers[hedgeTriggerLatency])) + assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assertProduceFailures(t, m, nil) + }) + + t.Run("unknown fallback error is a terminal failure reason", func(t *testing.T) { + producer := newMockDirectProducer() + producer.errs[primaryID] = kerr.RequestTimedOut + producer.errs[secondaryID] = errors.New("boom") + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, healthyTracker(), stratPrimaryAndSecondary, health, cfg, 0, 1<<20, m, nil) + + capture := newResultCapture() + runHedger(h, context.Background(), makeReq(capture)) + require.Error(t, capture.get(topic, partition).err) + assert.Equal(t, []int32{primaryID, secondaryID}, producer.recordedCallNodeIDs()) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonTerminalError: 1, + }) + }) + + t.Run("duplicate partition is an internal error and dispatches no fallback", func(t *testing.T) { + producer := newMockDirectProducer() + producer.errs[primaryID] = kerr.RequestTimedOut + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, healthyTracker(), stratPrimaryAndSecondary, health, cfg, 0, 1<<20, m, nil) + + part := routedEncodedTopicPartitionRecords{ + encodedTopicPartitionRecords: newEncodedTopicPartitionRecords(topic, partition, []*kgo.Record{{Topic: topic, Partition: partition}}), + nodeID: primaryID, + } + err := h.ProduceSync(context.Background(), primaryID, []routedEncodedTopicPartitionRecords{part, part}).error() + require.Error(t, err) + assert.ErrorContains(t, err, "duplicate") + assert.Equal(t, float64(1), testutil.ToFloat64(m.hedgeAttemptsTotal)) + assert.Equal(t, float64(0), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assert.Equal(t, []int32{primaryID}, producer.recordedCallNodeIDs()) + assertProduceFailures(t, m, map[produceFailureReason]float64{ + produceFailureReasonInternalError: 1, + }) }) t.Run("shared hedge flush does not resolve a partition the leg did not send", func(t *testing.T) { @@ -1204,6 +1634,209 @@ func TestHedger_ProduceSync(t *testing.T) { }) } +// TestHedger_AttemptPayloadCountsMergedHedgeFlush checks the hedge payload is +// measured after buffer merging: two callers' same-partition retries share one +// flush, which is one attempt carrying the merged batch. +func TestHedger_AttemptPayloadCountsMergedHedgeFlush(t *testing.T) { + const ( + topic = "t" + partition = int32(0) + primary = int32(100) + fallback = int32(101) + ) + tracker := NewAverageAgentStatsTracker() + nowNs := time.Now().UnixNano() + for _, id := range []int32{primary, fallback} { + seedFullWindow(tracker, id, nowNs, 20, 1, 0) + } + strategy := &mockPartitionAssignmentStrategy{ + candidates: map[partitionKey][]Agent{{topic, partition}: healthyAgents(primary, fallback)}, + } + producer := newMockDirectProducer() + producer.errs[primary] = kerr.RequestTimedOut + producer.respFn = func(_ int32, partitions []encodedTopicPartitionRecords) (*kmsg.ProduceResponse, error) { + resp := &kmsg.ProduceResponse{} + for _, p := range partitions { + resp.Topics = append(resp.Topics, kmsg.ProduceResponseTopic{ + Topic: p.topic, + Partitions: []kmsg.ProduceResponseTopicPartition{{Partition: p.partition}}, + }) + } + return resp, nil + } + health := HealthCheckConfig{SlowMultiplier: 2.0, MaxSlowFraction: 0.3, FaultyThreshold: 0.05, MaxFaultyFraction: 0.3} + // A failed primary returns long before the hedge delay, so each caller + // cascades without racing the timer. + cfg := HedgerConfig{MinHedgeDelay: time.Hour, MaxHedgeAgents: 2} + + m := newMetrics(prometheus.NewPedanticRegistry()) + // The linger never fires: the test flushes the fallback buffer itself. + h := NewHedger(producer, tracker, strategy, health, cfg, time.Hour, 1<<20, m, nil) + t.Cleanup(h.Close) + + buffer, err := h.hedgeBuffer.agentBufferFor(fallback) + require.NoError(t, err) + + done1 := runHedgerAsync(h, routedHedgeRecord(topic, partition, primary, "caller-1", nil)) + done2 := runHedgerAsync(h, routedHedgeRecord(topic, partition, primary, "caller-2", nil)) + + require.EventuallyWithT(t, func(c *assert.CollectT) { + assert.ElementsMatch(c, []string{"caller-1", "caller-2"}, bufferedRecordValues(buffer)) + }, time.Second, time.Millisecond) + buffer.timerFlush() + for _, done := range []<-chan struct{}{done1, done2} { + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("ProduceSync did not return after the hedge flush") + } + } + + // What the producer actually received, per role. + sent := map[attemptRole]produceRequestStats{} + var hedgeCalls int + for _, call := range producer.recordedCalls() { + role := attemptHedge + if call.nodeID == primary { + role = attemptPrimary + } else { + hedgeCalls++ + } + for _, p := range call.partitions { + sent[role] = sent[role].add(p.encodedStats) + } + } + require.Equal(t, 1, hedgeCalls, "both retries share one hedge flush") + assert.Equal(t, int64(2), sent[attemptHedge].records, "the one flush carries both callers' records") + assert.Equal(t, int64(2), sent[attemptPrimary].records, "each caller's primary carries one record") + + assertAttemptPayload(t, m, sent[attemptPrimary], sent[attemptHedge]) + assert.Equal(t, float64(1), testutil.ToFloat64(m.produceRequestsHedgeTotal)) + assert.Equal(t, float64(2), testutil.ToFloat64(m.produceRequestsPrimaryTotal)) +} + +// deadlineCtx reports a fixed deadline without ever arming its timer, like a +// context whose deadline has been reached but whose timer has not fired yet. +type deadlineCtx struct { + context.Context + deadline time.Time +} + +func (c deadlineCtx) Deadline() (time.Time, bool) { return c.deadline, true } + +func TestHedger_CtxStopErr(t *testing.T) { + canceled, cancel := context.WithCancel(context.Background()) + cancel() + + assert.NoError(t, ctxStopErr(context.Background()), "no deadline, not done") + assert.ErrorIs(t, ctxStopErr(canceled), context.Canceled) + assert.NoError(t, ctxStopErr(deadlineCtx{context.Background(), time.Now().Add(time.Hour)}), "deadline not reached") + assert.ErrorIs(t, + ctxStopErr(deadlineCtx{context.Background(), time.Now().Add(-time.Millisecond)}), + context.DeadlineExceeded, "deadline reached before its timer fired") + assert.ErrorIs(t, ctxStopErr(deadlineCtx{canceled, time.Now().Add(-time.Millisecond)}), context.Canceled, + "an explicit cancel is reported as a cancel, not a deadline") +} + +// TestHedger_WinCountMatchesSelectedResult releases a primary and a fallback +// that both succeed, one after the other in either order, and checks a win is +// counted exactly when the returned result is the fallback's. The loser is held +// inside the producer, ignoring the work context, so it finishes successfully +// after the owner has returned and must not count. +func TestHedger_WinCountMatchesSelectedResult(t *testing.T) { + const ( + topic = "t" + partition = int32(0) + primary = int32(100) + fallback = int32(101) + primaryOffset = int64(1000) + fallbackOff = int64(2000) + ) + strategy := &mockPartitionAssignmentStrategy{ + candidates: map[partitionKey][]Agent{{topic, partition}: { + {NodeID: primary, State: AgentStateDemoted}, + {NodeID: fallback, State: AgentStateHealthy}, + }}, + } + tracker := NewAverageAgentStatsTracker() + nowNs := time.Now().UnixNano() + for _, id := range []int32{primary, fallback} { + seedFullWindow(tracker, id, nowNs, 20, 1, 0) + } + health := HealthCheckConfig{SlowMultiplier: 2.0, MaxSlowFraction: 0.3, FaultyThreshold: 0.05, MaxFaultyFraction: 0.3} + cfg := HedgerConfig{MinHedgeDelay: time.Hour, MaxHedgeAgents: 2} + + for _, fallbackFirst := range []bool{true, false} { + t.Run(fmt.Sprintf("fallbackFirst=%v", fallbackFirst), func(t *testing.T) { + gates := map[int32]chan struct{}{primary: make(chan struct{}), fallback: make(chan struct{})} + entered := map[int32]chan struct{}{primary: make(chan struct{}, 1), fallback: make(chan struct{}, 1)} + + producer := newMockDirectProducer() + // respFn takes no context, so a held call completes successfully + // even after the work context is canceled. + producer.respFn = func(nodeID int32, _ []encodedTopicPartitionRecords) (*kmsg.ProduceResponse, error) { + entered[nodeID] <- struct{}{} + <-gates[nodeID] + offset := primaryOffset + if nodeID == fallback { + offset = fallbackOff + } + return &kmsg.ProduceResponse{Topics: []kmsg.ProduceResponseTopic{{ + Topic: topic, + Partitions: []kmsg.ProduceResponseTopicPartition{{Partition: partition, BaseOffset: offset}}, + }}}, nil + } + + m := newMetrics(prometheus.NewPedanticRegistry()) + h := NewHedger(producer, tracker, strategy, health, cfg, 0, 1<<20, m, nil) + t.Cleanup(h.Close) + + // A demoted primary starts the fallback immediately. + part := routedHedgeRecord(topic, partition, primary, "v", nil) + part.item.nodeState = AgentStateDemoted + resCh := make(chan ProduceResult, 1) + go func() { + resCh <- h.ProduceSync(context.Background(), primary, unpromise([]promised[routedEncodedTopicPartitionRecords]{part})) + }() + + // Both legs are in flight before either is released. + for _, id := range []int32{primary, fallback} { + select { + case <-entered[id]: + case <-time.After(time.Second): + t.Fatalf("agent %d was never called", id) + } + } + + winner, loser := primary, fallback + wantOffset, wantWins := primaryOffset, float64(0) + if fallbackFirst { + winner, loser = fallback, primary + wantOffset, wantWins = fallbackOff, 1 + } + close(gates[winner]) + res := <-resCh + require.NoError(t, res.error()) + assert.Equal(t, wantOffset, res.resp.Topics[0].Partitions[0].BaseOffset) + + // The loser now completes successfully, after the owner returned. + close(gates[loser]) + require.Eventually(t, func() bool { + for _, call := range producer.recordedCalls() { + if call.nodeID == loser && call.err == nil { + return true + } + } + return false + }, time.Second, time.Millisecond, "the losing leg must finish successfully") + h.Close() + + assert.Equal(t, wantWins, testutil.ToFloat64(m.hedgeWinsTotal)) + assertTriggerWins(t, m, map[hedgeTrigger]float64{hedgeTriggerDemotedProbe: wantWins}) + }) + } +} + func TestMaxFractionFloor(t *testing.T) { tests := map[string]struct { configured float64 @@ -1276,3 +1909,69 @@ func TestHedgerCandidates_fetch(t *testing.T) { assert.Equal(t, 2, strategy.lastMaxCandidates(topic, 0)) }) } + +func TestHedger_FinalStop(t *testing.T) { + canceled, cancel := context.WithCancel(context.Background()) + cancel() + expired := deadlineCtx{context.Background(), time.Now().Add(-time.Millisecond)} + + failed := ProduceResult{err: kerr.RequestTimedOut} + succeeded := ProduceResult{resp: &kmsg.ProduceResponse{}} + + tests := []struct { + name string + ctx context.Context + primary ProduceResult + result ProduceResult + stop produceStop + want produceStop + }{ + {"a successful result is left alone", context.Background(), failed, succeeded, produceStopCandidatesExhausted, produceStopCandidatesExhausted}, + {"a settled terminal stop is left alone", expired, failed, failed, produceStopTerminalError, produceStopTerminalError}, + {"a settled write timeout is left alone", context.Background(), failed, failed, produceStopWriteTimeout, produceStopWriteTimeout}, + {"a write timeout after a non-retriable primary error is terminal", expired, ProduceResult{err: kerr.MessageTooLarge}, failed, produceStopWriteTimeout, produceStopTerminalError}, + {"a write timeout after a canceled primary stays a write timeout", expired, ProduceResult{err: context.Canceled}, failed, produceStopWriteTimeout, produceStopWriteTimeout}, + {"exhausted stays exhausted while the context is live", context.Background(), failed, failed, produceStopCandidatesExhausted, produceStopCandidatesExhausted}, + {"exhausted after a non-retriable primary error is terminal", context.Background(), ProduceResult{err: kerr.MessageTooLarge}, failed, produceStopCandidatesExhausted, produceStopTerminalError}, + {"a non-retriable primary error outranks an expired deadline", expired, ProduceResult{err: kerr.MessageTooLarge}, failed, produceStopCandidatesExhausted, produceStopTerminalError}, + {"exhausted with an expired deadline is a write timeout", expired, failed, failed, produceStopCandidatesExhausted, produceStopWriteTimeout}, + {"exhausted with a canceled context emits no failure", canceled, failed, failed, produceStopCandidatesExhausted, produceStopCanceled}, + {"a canceled primary is a cancel, not a terminal error", canceled, ProduceResult{err: context.Canceled}, failed, produceStopCandidatesExhausted, produceStopCanceled}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, finalStop(tt.ctx, tt.primary, tt.result, tt.stop)) + }) + } +} + +func TestHedger_ClassifyProduceStop(t *testing.T) { + brokerErr := kerr.MessageTooLarge + unknownErr := errors.New("boom") + + tests := []struct { + name string + terminal bool + terminalErr error + ctxErr error + exhausted bool + want produceStop + }{ + {name: "success", want: produceStopNone}, + {name: "candidates exhausted", exhausted: true, want: produceStopCandidatesExhausted}, + {name: "terminal broker error", terminal: true, terminalErr: brokerErr, want: produceStopTerminalError}, + {name: "terminal unknown error", terminal: true, terminalErr: unknownErr, want: produceStopTerminalError}, + {name: "deadline is the returned failure", ctxErr: context.DeadlineExceeded, want: produceStopWriteTimeout}, + {name: "deadline after exhaustion is the returned failure", ctxErr: context.DeadlineExceeded, exhausted: true, want: produceStopWriteTimeout}, + {name: "deadline does not relabel a terminal broker error", terminal: true, terminalErr: brokerErr, ctxErr: context.DeadlineExceeded, want: produceStopTerminalError}, + {name: "explicit cancel", ctxErr: context.Canceled, want: produceStopCanceled}, + {name: "explicit cancel after exhaustion", ctxErr: context.Canceled, exhausted: true, want: produceStopCanceled}, + {name: "cancel that aborted the accumulator", terminal: true, terminalErr: context.Canceled, ctxErr: context.Canceled, want: produceStopCanceled}, + {name: "cancel does not relabel a terminal broker error", terminal: true, terminalErr: brokerErr, ctxErr: context.Canceled, want: produceStopTerminalError}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, classifyProduceStop(tt.terminal, tt.terminalErr, tt.ctxErr, tt.exhausted)) + }) + } +} diff --git a/pkg/wgo/metrics.go b/pkg/wgo/metrics.go index 55e74d8..1e55e36 100644 --- a/pkg/wgo/metrics.go +++ b/pkg/wgo/metrics.go @@ -26,6 +26,10 @@ type metrics struct { hedgeAttemptsTotal prometheus.Counter hedgeWinsTotal prometheus.Counter hedgeAttemptsSuppressedTotal *prometheus.CounterVec + hedgeTriggers [hedgeTriggerCount]prometheus.Counter + hedgeTriggerWins [hedgeTriggerCount]prometheus.Counter + produceRequestsFailed [produceFailureReasonCount]prometheus.Counter + agentpoolAgentsChanged [agentpoolChurnCount]prometheus.Counter lingerFlushesTotal prometheus.Counter @@ -46,11 +50,14 @@ type metrics struct { produceRequestsPrimaryTotal prometheus.Counter produceRequestsHedgeTotal prometheus.Counter + produceAttemptRecords [attemptRoleCount]prometheus.Counter + produceAttemptBytes [attemptRoleCount]prometheus.Counter + produceRecordsTotal prometheus.Counter produceRecordsFailedTotal prometheus.Counter produceRecordsRejectedTotal *prometheus.CounterVec - agentPoolLeaderDroppedTotal prometheus.Counter + agentPoolExcludedLeaders prometheus.Gauge metadataRefreshResultsTotal *prometheus.CounterVec clusterStatsAvailable prometheus.Gauge @@ -102,6 +109,52 @@ const ( metadataRefreshResultFailed = "failed" ) +// attemptRole is which side of the hedge a Produce attempt is on. +type attemptRole int8 + +const ( + attemptPrimary attemptRole = iota + attemptHedge + attemptRoleCount +) + +const ( + attemptLabelPrimary = "primary" + attemptLabelHedge = "hedge" +) + +// produceFailureReason is why a Hedger produce failed. One increment is one +// failed invocation, not a record, partition or wire attempt. +type produceFailureReason int8 + +const ( + produceFailureReasonCandidatesExhausted produceFailureReason = iota + produceFailureReasonTerminalError + produceFailureReasonWriteTimeout + produceFailureReasonInternalError + produceFailureReasonCount +) + +const ( + produceFailureReasonLabelCandidatesExhausted = "candidates_exhausted" + produceFailureReasonLabelTerminalError = "terminal_error" + produceFailureReasonLabelWriteTimeout = "write_timeout" + produceFailureReasonLabelInternalError = "internal_error" +) + +type agentpoolChurnDirection int8 + +const ( + agentpoolChurnAdded agentpoolChurnDirection = iota + agentpoolChurnRemoved + agentpoolChurnCount +) + +const ( + agentpoolChurnLabelAdded = "added" + agentpoolChurnLabelRemoved = "removed" +) + func agentStateLabel(state AgentState) string { if state == AgentStateDemoted { return agentStateDemoted @@ -178,6 +231,36 @@ func newMetrics(reg prometheus.Registerer) *metrics { NativeHistogramMinResetDuration: time.Hour, }, []string{"outcome"}) + hedgeTriggers := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "warpstream_produce_hedge_triggers_total", + Help: "Why a logical fallback cascade started: latency (hedge timer, or a healthy primary whose computed delay is already zero), primary_failure (the primary failed before the race), or demoted_probe (the routing-time primary was demoted). One increment per cascade entry, including a cascade that dispatches no request. Not a wire request or a hedge wave.", + }, []string{"trigger"}) + + hedgeTriggerWins := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "warpstream_produce_hedge_trigger_wins_total", + Help: "Logical fallback cascades whose result won, by the trigger that started the cascade (latency, primary_failure, demoted_probe). Counted at the same point as warpstream_hedge_wins_total, so the series sum to it. Divide by warpstream_produce_hedge_triggers_total for the win rate of each trigger.", + }, []string{"trigger"}) + + produceAttemptRecords := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "warpstream_produce_attempt_records_total", + Help: "Records handed to the direct producer, by attempt role (primary, hedge). Counted at dispatch whether or not the attempt succeeds, including a losing leg that is canceled afterwards. This is attempted work, not records confirmed on the wire; compare with produce_records_total, which counts only acked requests.", + }, []string{"attempt"}) + + produceAttemptBytes := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "warpstream_produce_attempt_bytes_total", + Help: "Compressed record bytes handed to the direct producer, by attempt role (primary, hedge). Same boundary as warpstream_produce_attempt_records_total; compare with produce_compressed_bytes_total, which counts only acked requests.", + }, []string{"attempt"}) + + produceRequestsFailed := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "warpstream_produce_requests_failed_total", + Help: "Why a Hedger produce failed: candidates_exhausted (a partition hit its candidate budget or had no unused candidate), terminal_error (a non-retriable or unknown error from the primary or a retry), write_timeout (the work deadline expired; it outranks candidate exhaustion but not a terminal error), or internal_error (routing mismatch, duplicate partition, or an unclassifiable result). The reason is why the retry cascade stopped. One increment per failed invocation, not per public call, record, partition, or wire attempt. Success and caller cancellation are omitted. The routing-mismatch guard is counted here but not in warpstream_produce_requests_attempts.", + }, []string{"reason"}) + + agentpoolAgentsChanged := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "warpstream_agentpool_agents_changed_total", + Help: "NodeIDs added to or removed from the AgentPool on a successful live Metadata refresh, by direction. Constructor initialization is excluded. An address-only or leader-only change is not membership churn.", + }, []string{"direction"}) + version, franzGoVersion := clientBuildInfo() promauto.With(reg).NewGauge(prometheus.GaugeOpts{ Name: "warpstream_client_build_info", @@ -198,6 +281,34 @@ func newMetrics(reg prometheus.Registerer) *metrics { Name: "warpstream_hedge_attempts_suppressed_total", Help: "Total number of produce requests where hedging was suppressed.", }, []string{"reason"}), + hedgeTriggers: [hedgeTriggerCount]prometheus.Counter{ + hedgeTriggerLatency: hedgeTriggers.WithLabelValues(hedgeTriggerLabelLatency), + hedgeTriggerPrimaryFailure: hedgeTriggers.WithLabelValues(hedgeTriggerLabelPrimaryFailure), + hedgeTriggerDemotedProbe: hedgeTriggers.WithLabelValues(hedgeTriggerLabelDemotedProbe), + }, + produceAttemptRecords: [attemptRoleCount]prometheus.Counter{ + attemptPrimary: produceAttemptRecords.WithLabelValues(attemptLabelPrimary), + attemptHedge: produceAttemptRecords.WithLabelValues(attemptLabelHedge), + }, + produceAttemptBytes: [attemptRoleCount]prometheus.Counter{ + attemptPrimary: produceAttemptBytes.WithLabelValues(attemptLabelPrimary), + attemptHedge: produceAttemptBytes.WithLabelValues(attemptLabelHedge), + }, + hedgeTriggerWins: [hedgeTriggerCount]prometheus.Counter{ + hedgeTriggerLatency: hedgeTriggerWins.WithLabelValues(hedgeTriggerLabelLatency), + hedgeTriggerPrimaryFailure: hedgeTriggerWins.WithLabelValues(hedgeTriggerLabelPrimaryFailure), + hedgeTriggerDemotedProbe: hedgeTriggerWins.WithLabelValues(hedgeTriggerLabelDemotedProbe), + }, + produceRequestsFailed: [produceFailureReasonCount]prometheus.Counter{ + produceFailureReasonCandidatesExhausted: produceRequestsFailed.WithLabelValues(produceFailureReasonLabelCandidatesExhausted), + produceFailureReasonTerminalError: produceRequestsFailed.WithLabelValues(produceFailureReasonLabelTerminalError), + produceFailureReasonWriteTimeout: produceRequestsFailed.WithLabelValues(produceFailureReasonLabelWriteTimeout), + produceFailureReasonInternalError: produceRequestsFailed.WithLabelValues(produceFailureReasonLabelInternalError), + }, + agentpoolAgentsChanged: [agentpoolChurnCount]prometheus.Counter{ + agentpoolChurnAdded: agentpoolAgentsChanged.WithLabelValues(agentpoolChurnLabelAdded), + agentpoolChurnRemoved: agentpoolAgentsChanged.WithLabelValues(agentpoolChurnLabelRemoved), + }, lingerFlushesTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{ Name: "warpstream_linger_flushes_total", Help: "Total number of partition batch flushes triggered by the linger buffer.", @@ -237,9 +348,9 @@ func newMetrics(reg prometheus.Registerer) *metrics { Name: "warpstream_produce_records_rejected_total", Help: "Total number of records rejected by the client before any wire dispatch, by reason (record_too_large, no_agent_assigned).", }, []string{"reason"}), - agentPoolLeaderDroppedTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{ - Name: "warpstream_agentpool_leader_dropped_total", - Help: "Partition leaders excluded from the assignment map because their NodeID was absent from that Metadata response's broker list. One increment per excluded leader, including the constructor refresh. Topic-level Metadata errors are not counted. A partition Leader below 0 is not counted.", + agentPoolExcludedLeaders: promauto.With(reg).NewGauge(prometheus.GaugeOpts{ + Name: "warpstream_agentpool_excluded_leaders", + Help: "Number of partitions whose named leader NodeID was absent from the broker list of the last successful Metadata refresh, so the AgentPool excluded that leader. Produces to these partitions go to a stand-in while the pool has any agent; with an empty broker list they are rejected instead. 0 when no leader is excluded. Set on every successful refresh, including the constructor refresh; a failed refresh keeps the previous value. Topic-level Metadata errors and a partition Leader below 0 are not counted.", }), metadataRefreshResultsTotal: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ Name: "warpstream_metadata_refresh_results_total", @@ -299,6 +410,21 @@ func (m *metrics) observeMetadataRefresh(trigger metadataRefreshTrigger, before, result = metadataRefreshResultMembershipChanged } m.metadataRefreshResultsTotal.WithLabelValues(string(trigger), result).Inc() + if err != nil { + return + } + added, removed := diffAgentMembership(before, after) + if added > 0 { + m.agentpoolAgentsChanged[agentpoolChurnAdded].Add(float64(added)) + } + if removed > 0 { + m.agentpoolAgentsChanged[agentpoolChurnRemoved].Add(float64(removed)) + } +} + +func (m *metrics) observeAttempt(role attemptRole, stats produceRequestStats) { + m.produceAttemptRecords[role].Add(float64(stats.records)) + m.produceAttemptBytes[role].Add(float64(stats.compressedBytes)) } // observeClusterStats records one ClusterStats compute. Without a view the diff --git a/pkg/wgo/metrics_test.go b/pkg/wgo/metrics_test.go index cc08fe2..6f03e71 100644 --- a/pkg/wgo/metrics_test.go +++ b/pkg/wgo/metrics_test.go @@ -1,6 +1,7 @@ package wgo import ( + "fmt" "runtime/debug" "strings" "testing" @@ -239,6 +240,144 @@ func TestMetrics_ObserveClusterStats(t *testing.T) { require.InDelta(t, 5, gaugeValue(t, reg, "warpstream_cluster_faulty_contributors"), 0) } +func TestNewMetrics_HedgeTriggers(t *testing.T) { + reg := prometheus.NewPedanticRegistry() + m := newMetrics(reg) + + m.hedgeTriggers[hedgeTriggerLatency].Inc() + m.hedgeTriggers[hedgeTriggerPrimaryFailure].Add(2) + m.hedgeTriggers[hedgeTriggerDemotedProbe].Add(3) + + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP warpstream_produce_hedge_triggers_total Why a logical fallback cascade started: latency (hedge timer, or a healthy primary whose computed delay is already zero), primary_failure (the primary failed before the race), or demoted_probe (the routing-time primary was demoted). One increment per cascade entry, including a cascade that dispatches no request. Not a wire request or a hedge wave. + # TYPE warpstream_produce_hedge_triggers_total counter + warpstream_produce_hedge_triggers_total{trigger="demoted_probe"} 3 + warpstream_produce_hedge_triggers_total{trigger="latency"} 1 + warpstream_produce_hedge_triggers_total{trigger="primary_failure"} 2 + `), "warpstream_produce_hedge_triggers_total")) +} + +func TestNewMetrics_AgentPoolExcludedLeaders(t *testing.T) { + reg := prometheus.NewPedanticRegistry() + m := newMetrics(reg) + + // 0 before any refresh publishes a value. + require.InDelta(t, 0.0, gaugeValue(t, reg, "warpstream_agentpool_excluded_leaders"), 0) + + m.agentPoolExcludedLeaders.Set(3) + require.InDelta(t, 3.0, gaugeValue(t, reg, "warpstream_agentpool_excluded_leaders"), 0) + + // The old counter is gone: it was never deployed. + families, err := reg.Gather() + require.NoError(t, err) + for _, f := range families { + assert.NotEqual(t, "warpstream_agentpool_leader_dropped_total", f.GetName()) + } +} + +func TestNewMetrics_ProduceAttemptPayload(t *testing.T) { + reg := prometheus.NewPedanticRegistry() + m := newMetrics(reg) + + m.observeAttempt(attemptPrimary, produceRequestStats{records: 3, batches: 2, uncompressedBytes: 900, compressedBytes: 300}) + m.observeAttempt(attemptHedge, produceRequestStats{records: 1, batches: 1, uncompressedBytes: 90, compressedBytes: 40}) + m.observeAttempt(attemptHedge, produceRequestStats{records: 2, batches: 1, uncompressedBytes: 60, compressedBytes: 20}) + + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP warpstream_produce_attempt_records_total Records handed to the direct producer, by attempt role (primary, hedge). Counted at dispatch whether or not the attempt succeeds, including a losing leg that is canceled afterwards. This is attempted work, not records confirmed on the wire; compare with produce_records_total, which counts only acked requests. + # TYPE warpstream_produce_attempt_records_total counter + warpstream_produce_attempt_records_total{attempt="hedge"} 3 + warpstream_produce_attempt_records_total{attempt="primary"} 3 + # HELP warpstream_produce_attempt_bytes_total Compressed record bytes handed to the direct producer, by attempt role (primary, hedge). Same boundary as warpstream_produce_attempt_records_total; compare with produce_compressed_bytes_total, which counts only acked requests. + # TYPE warpstream_produce_attempt_bytes_total counter + warpstream_produce_attempt_bytes_total{attempt="hedge"} 60 + warpstream_produce_attempt_bytes_total{attempt="primary"} 300 + `), "warpstream_produce_attempt_records_total", "warpstream_produce_attempt_bytes_total")) +} + +func TestNewMetrics_HedgeTriggerWins(t *testing.T) { + reg := prometheus.NewPedanticRegistry() + m := newMetrics(reg) + + m.hedgeTriggerWins[hedgeTriggerLatency].Inc() + m.hedgeTriggerWins[hedgeTriggerPrimaryFailure].Add(2) + m.hedgeTriggerWins[hedgeTriggerDemotedProbe].Add(3) + + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP warpstream_produce_hedge_trigger_wins_total Logical fallback cascades whose result won, by the trigger that started the cascade (latency, primary_failure, demoted_probe). Counted at the same point as warpstream_hedge_wins_total, so the series sum to it. Divide by warpstream_produce_hedge_triggers_total for the win rate of each trigger. + # TYPE warpstream_produce_hedge_trigger_wins_total counter + warpstream_produce_hedge_trigger_wins_total{trigger="demoted_probe"} 3 + warpstream_produce_hedge_trigger_wins_total{trigger="latency"} 1 + warpstream_produce_hedge_trigger_wins_total{trigger="primary_failure"} 2 + `), "warpstream_produce_hedge_trigger_wins_total")) +} + +func TestNewMetrics_ProduceRequestsFailed(t *testing.T) { + reg := prometheus.NewPedanticRegistry() + m := newMetrics(reg) + + m.produceRequestsFailed[produceFailureReasonCandidatesExhausted].Inc() + m.produceRequestsFailed[produceFailureReasonTerminalError].Add(2) + m.produceRequestsFailed[produceFailureReasonWriteTimeout].Add(3) + m.produceRequestsFailed[produceFailureReasonInternalError].Add(4) + + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP warpstream_produce_requests_failed_total Why a Hedger produce failed: candidates_exhausted (a partition hit its candidate budget or had no unused candidate), terminal_error (a non-retriable or unknown error from the primary or a retry), write_timeout (the work deadline expired; it outranks candidate exhaustion but not a terminal error), or internal_error (routing mismatch, duplicate partition, or an unclassifiable result). The reason is why the retry cascade stopped. One increment per failed invocation, not per public call, record, partition, or wire attempt. Success and caller cancellation are omitted. The routing-mismatch guard is counted here but not in warpstream_produce_requests_attempts. + # TYPE warpstream_produce_requests_failed_total counter + warpstream_produce_requests_failed_total{reason="candidates_exhausted"} 1 + warpstream_produce_requests_failed_total{reason="internal_error"} 4 + warpstream_produce_requests_failed_total{reason="terminal_error"} 2 + warpstream_produce_requests_failed_total{reason="write_timeout"} 3 + `), "warpstream_produce_requests_failed_total")) +} + +func TestMetrics_ObserveAgentPoolChurn(t *testing.T) { + reg := prometheus.NewPedanticRegistry() + m := newMetrics(reg) + + m.observeMetadataRefresh(metadataRefreshTriggerOnDemand, nil, []int32{1, 2}, nil) + m.observeMetadataRefresh(metadataRefreshTriggerPeriodic, []int32{1, 2}, []int32{1, 2}, nil) + // A failed refresh reports no churn even if the sets happen to differ. + m.observeMetadataRefresh(metadataRefreshTriggerOnDemand, []int32{1}, []int32{1, 9}, assert.AnError) + m.observeMetadataRefresh(metadataRefreshTriggerPeriodic, []int32{1, 2, 3}, []int32{1, 4}, nil) + + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(` + # HELP warpstream_agentpool_agents_changed_total NodeIDs added to or removed from the AgentPool on a successful live Metadata refresh, by direction. Constructor initialization is excluded. An address-only or leader-only change is not membership churn. + # TYPE warpstream_agentpool_agents_changed_total counter + warpstream_agentpool_agents_changed_total{direction="added"} 3 + warpstream_agentpool_agents_changed_total{direction="removed"} 2 + `), "warpstream_agentpool_agents_changed_total")) +} + +// BenchmarkMetrics_ObserveAttempt measures the per-dispatch cost of the +// attempted-payload counters for a many-partition payload. +func BenchmarkMetrics_ObserveAttempt(b *testing.B) { + for _, partitions := range []int{1, 32, 256, 1024} { + b.Run(fmt.Sprintf("partitions=%d", partitions), func(b *testing.B) { + m := newMetrics(prometheus.NewRegistry()) + parts := make([]routedEncodedTopicPartitionRecords, partitions) + for i := range parts { + parts[i].encodedStats = produceRequestStats{records: 10, batches: 1, uncompressedBytes: 1000, compressedBytes: 400} + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + m.observeAttempt(attemptPrimary, sumEncodedStats(parts)) + } + }) + } +} + +func BenchmarkMetrics_HedgeTriggerInc(b *testing.B) { + m := newMetrics(prometheus.NewRegistry()) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + m.hedgeAttemptsTotal.Inc() + m.hedgeTriggers[hedgeTriggerLatency].Inc() + } +} + func BenchmarkMetrics_DirectRequestAccounting(b *testing.B) { m := newMetrics(prometheus.NewRegistry()) b.ReportAllocs() diff --git a/pkg/wgo/produce_result.go b/pkg/wgo/produce_result.go index 0dda80f..77b4c68 100644 --- a/pkg/wgo/produce_result.go +++ b/pkg/wgo/produce_result.go @@ -276,6 +276,17 @@ func (a *produceResultAccumulator) accumulate(res scopedProduceResult) { } } +// terminalErr reports whether the abort policy stopped retries, and the error +// that tripped it. +func (a *produceResultAccumulator) terminalErr() (bool, error) { + a.mu.Lock() + defer a.mu.Unlock() + if !a.aborted { + return false, nil + } + return true, a.lastErr.err +} + // done reports whether the caller should stop retrying. func (a *produceResultAccumulator) done() bool { a.mu.Lock()