Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

kvserver: fix reproposals test with pipelined writes #113658

Merged
merged 4 commits into from
Nov 3, 2023
Merged

kvserver: fix reproposals test with pipelined writes #113658

Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
832e6f5
kvserver: add missing counter increment
pav-kv Nov 2, 2023
02af5a7
kvserver: add metric checks to reproposals test
pav-kv Nov 2, 2023
7c8be62
kvserver: fix reproposals test with pipelined writes
pav-kv Nov 2, 2023
7ed8c9e
kvserver: use BarchRequest.GetArg
pav-kv Nov 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
30 changes: 23 additions & 7 deletions pkg/kv/kvserver/replica_raft_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,12 @@ func testProposalsWithInjectedLeaseIndexAndReproposalError(t *testing.T, pipelin
if ba == nil {
return "", false // not local proposal
}
inc := ba.Requests[0].GetIncrement()
var inc *kvpb.IncrementRequest
for i := range ba.Requests {
if inc = ba.Requests[i].GetIncrement(); inc != nil {
pav-kv marked this conversation as resolved.
Show resolved Hide resolved
break
}
}
if inc == nil {
return "", false
}
Expand All @@ -191,7 +196,7 @@ func testProposalsWithInjectedLeaseIndexAndReproposalError(t *testing.T, pipelin

cfg := TestStoreConfig(hlc.NewClockForTesting(nil))

var injectedReproposalErrors atomic.Int32
var injectedReproposalErrors atomic.Int64
{
var mu syncutil.Mutex
seen := map[string]int{} // access from proposal buffer under raftMu
Expand All @@ -206,12 +211,13 @@ func testProposalsWithInjectedLeaseIndexAndReproposalError(t *testing.T, pipelin
if !shouldInject(0.2, seen[key]) {
return nil
}
injectedReproposalErrors.Add(1)
t.Logf("inserting reproposal error for %s (seen %d times)", key, seen[key])
return errors.Errorf("injected error")
}
}

var insertedIllegalLeaseIndex atomic.Int32
var insertedIllegalLeaseIndex atomic.Int64
{
var mu syncutil.Mutex
seen := map[string]int{}
Expand Down Expand Up @@ -253,8 +259,8 @@ func testProposalsWithInjectedLeaseIndexAndReproposalError(t *testing.T, pipelin
key = append(key, "-testing"...)
return key
}
var observedAsyncWriteFailures atomic.Int32
var observedReproposalErrors atomic.Int32
var observedAsyncWriteFailures atomic.Int64
var observedReproposalErrors atomic.Int64
const iters = 300
expectations := map[string]int{}
for i := 0; i < iters; i++ {
Expand Down Expand Up @@ -337,19 +343,29 @@ func testProposalsWithInjectedLeaseIndexAndReproposalError(t *testing.T, pipelin
require.NoError(t, err)
require.EqualValues(t, exp, n)
}

t.Logf("observed %d async write restarts, observed %d/%d injected aborts, %d injected illegal lease applied indexes",
observedAsyncWriteFailures.Load(), observedReproposalErrors.Load(), injectedReproposalErrors.Load(), insertedIllegalLeaseIndex.Load())
t.Logf("commands reproposed (unchanged): %d", tc.store.metrics.RaftCommandsReproposed.Count())
t.Logf("commands reproposed (new LAI): %d", tc.store.metrics.RaftCommandsReproposedLAI.Count())

if pipelined {
// If we did pipelined writes, if we needed to repropose and injected an
// error, this should surface as an async write failure instead.
require.Zero(t, observedReproposalErrors.Load())
}
if !pipelined {
require.Equal(t, injectedReproposalErrors.Load(), observedAsyncWriteFailures.Load())
} else {
// If we're not pipelined, we shouldn't be able to get an async write
// failure. This isn't testing anything about reproposals per se, rather
// it's validation that we're truly not doing pipelined writes.
require.Zero(t, observedAsyncWriteFailures.Load())
// All the injected reproposal errors should manifest to the transaction.
require.Equal(t, injectedReproposalErrors.Load(), observedReproposalErrors.Load())
}
// All incorrect lease indices should manifest either as a reproposal, or a
// failed reproposal (when an error is injected).
require.Equal(t, insertedIllegalLeaseIndex.Load(),
pav-kv marked this conversation as resolved.
Show resolved Hide resolved
tc.store.metrics.RaftCommandsReproposedLAI.Count()+injectedReproposalErrors.Load())
}

func checkNoLeakedTraceSpans(t *testing.T, store *Store) {
Expand Down