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

release-23.2: kvserver: fix reproposals test with pipelined writes #113769

Merged
merged 4 commits into from
Nov 6, 2023
Merged

release-23.2: kvserver: fix reproposals test with pipelined writes #113769

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
85112c0
kvserver: add missing counter increment
pav-kv Nov 2, 2023
cefa999
kvserver: add metric checks to reproposals test
pav-kv Nov 2, 2023
1d00234
kvserver: fix reproposals test with pipelined writes
pav-kv Nov 2, 2023
f06aa83
kvserver: use BarchRequest.GetArg
pav-kv Nov 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 21 additions & 14 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 @@ -164,16 +164,12 @@ func testProposalsWithInjectedLeaseIndexAndReproposalError(t *testing.T, pipelin
isOurCommand := func(ba *kvpb.BatchRequest) (_ string, ok bool) {
if ba == nil {
return "", false // not local proposal
}
inc := ba.Requests[0].GetIncrement()
if inc == nil {
return "", false
}
key := string(inc.Key)
if !strings.HasSuffix(key, "-testing") {
} else if inc, found := ba.GetArg(kvpb.Increment); !found {
return "", false
} else if key := string(inc.(*kvpb.IncrementRequest).Key); strings.HasSuffix(key, "-testing") {
return key, true
}
return key, true
return "", false
}

rnd, seed := randutil.NewPseudoRand()
Expand All @@ -191,7 +187,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 +202,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 +250,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 +334,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(),
tc.store.metrics.RaftCommandsReproposedLAI.Count()+injectedReproposalErrors.Load())
}

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