Skip to content

Commit

Permalink
storage: default learners to on
Browse files Browse the repository at this point in the history 
Closes cockroachdb#38902

Release note: None
  • Loading branch information
@danhhz
danhhz committed Aug 7, 2019
1 parent 11e4b35 commit c4dcc42
Show file tree
Hide file tree
Showing 17 changed files with 333 additions and 427 deletions.
2 changes: 1 addition & 1 deletion docs/generated/settings/settings.html
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
<tr><td><code>kv.closed_timestamp.target_duration</code></td><td>duration</td><td><code>30s</code></td><td>if nonzero, attempt to provide closed timestamp notifications for timestamps trailing cluster time by approximately this duration</td></tr>
<tr><td><code>kv.follower_read.target_multiple</code></td><td>float</td><td><code>3</code></td><td>if above 1, encourages the distsender to perform a read against the closest replica if a request is older than kv.closed_timestamp.target_duration * (1 + kv.closed_timestamp.close_fraction * this) less a clock uncertainty interval. This value also is used to create follower_timestamp(). (WARNING: may compromise cluster stability or correctness; do not edit without supervision)</td></tr>
<tr><td><code>kv.import.batch_size</code></td><td>byte size</td><td><code>32 MiB</code></td><td>the maximum size of the payload in an AddSSTable request (WARNING: may compromise cluster stability or correctness; do not edit without supervision)</td></tr>
<tr><td><code>kv.learner_replicas.enabled</code></td><td>boolean</td><td><code>false</code></td><td>use learner replicas for replica addition</td></tr>
<tr><td><code>kv.learner_replicas.enabled</code></td><td>boolean</td><td><code>true</code></td><td>use learner replicas for replica addition</td></tr>
<tr><td><code>kv.raft.command.max_size</code></td><td>byte size</td><td><code>64 MiB</code></td><td>maximum size of a raft command</td></tr>
<tr><td><code>kv.raft_log.disable_synchronization_unsafe</code></td><td>boolean</td><td><code>false</code></td><td>set to true to disable synchronization on Raft log writes to persistent storage. Setting to true risks data loss or data corruption on server crashes. The setting is meant for internal testing only and SHOULD NOT be used in production.</td></tr>
<tr><td><code>kv.range.backpressure_range_size_multiplier</code></td><td>float</td><td><code>2</code></td><td>multiple of range_max_bytes that a range is allowed to grow to without splitting before writes to that range are blocked, or 0 to disable</td></tr>
Expand Down
2 changes: 1 addition & 1 deletion pkg/sql/ambiguous_commit_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ func (t *interceptingTransport) SendNext(
// where the primary key is specified in advance, it can result in violated
// uniqueness constraints, or duplicate key violations. See #6053, #7604, and
// #10023.
func TestAmbiguousCommit(t *testing.T) {
func TestDanStressAmbiguousCommit(t *testing.T) {
defer leaktest.AfterTest(t)()

testutils.RunTrueAndFalse(t, "ambiguousSuccess", func(t *testing.T, ambiguousSuccess bool) {
Expand Down
304 changes: 68 additions & 236 deletions pkg/storage/client_merge_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,22 +38,22 @@ import (
"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
"github.com/cockroachdb/cockroach/pkg/storage/stateloader"
"github.com/cockroachdb/cockroach/pkg/storage/storagebase"
"github.com/cockroachdb/cockroach/pkg/storage/storagepb"
"github.com/cockroachdb/cockroach/pkg/storage/txnwait"
"github.com/cockroachdb/cockroach/pkg/testutils"
"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
"github.com/cockroachdb/cockroach/pkg/util/hlc"
"github.com/cockroachdb/cockroach/pkg/util/leaktest"
"github.com/cockroachdb/cockroach/pkg/util/log"
"github.com/cockroachdb/cockroach/pkg/util/protoutil"
"github.com/cockroachdb/cockroach/pkg/util/randutil"
"github.com/cockroachdb/cockroach/pkg/util/retry"
"github.com/cockroachdb/cockroach/pkg/util/syncutil"
"github.com/cockroachdb/cockroach/pkg/util/timeutil"
"github.com/gogo/protobuf/proto"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.etcd.io/etcd/raft"
"go.etcd.io/etcd/raft/raftpb"
)
Expand Down Expand Up @@ -1749,146 +1749,87 @@ func TestStoreReplicaGCAfterMerge(t *testing.T) {
}
}

// TestStoreRangeMergeResplitAddReplicaRace verifies that an add replica request
// that occurs concurrently with a merge is aborted.
//
// To see why aborting the add replica request is necessary, consider two
// adjacent and collocated ranges, Q and R. Say the replicate queue decides to
// rebalance Q onto store S4. It will initiate a ChangeReplicas command that
// will send S4 a preemptive snapshot, then launch a replica-change transaction
// to update R's range descriptor with the new replica. Now say the merge queue
// decides to merge Q and R after the preemptive snapshot of Q has been sent to
// S4 but before the replica-change transaction has started. The merge can
// succeed because the ranges are still collocated. (The new replica of Q is
// only considered added once the replica-change transaction commits.) If the
// replica-change transaction were to commit, the new replica of Q on S4 would
// have a snapshot of Q that predated the merge. In order to catch up, it would
// need to apply the merge trigger, but the merge trigger will explode because
// S4 does not have a replica of R.
//
// To avoid this scenario, ChangeReplicas commands will abort if they discover
// the range descriptor has changed between when the snapshot is sent and when
// the replica-change transaction starts.
func TestStoreRangeMergeAddReplicaRace(t *testing.T) {
// TestStoreRangeMergeAddReplicaRace verifies that when an add replica request
// occurs concurrently with a merge, one of them is aborted with a "descriptor
// changed" CPut error.
func TestDanStressStoreRangeMergeAddReplicaRace(t *testing.T) {
defer leaktest.AfterTest(t)()

ctx := context.Background()
storeCfg := storage.TestStoreConfig(nil)
storeCfg.TestingKnobs.DisableSplitQueue = true
storeCfg.TestingKnobs.DisableMergeQueue = true
storeCfg.TestingKnobs.DisableReplicateQueue = true

mtc := &multiTestContext{
storeConfig: &storeCfg,
// This test was written before the multiTestContext started creating many
// system ranges at startup, and hasn't been update to take that into
// account.
startWithSingleRange: true,
}

mtc.Start(t, 2)
defer mtc.Stop()
store0, store1 := mtc.Store(0), mtc.Store(1)

lhsDesc, rhsDesc, err := createSplitRanges(ctx, store0)
if err != nil {
t.Fatal(err)
}

// Arrange to block all snapshots until we're ready.
snapshotSentCh := make(chan struct{})
allowSnapshotCh := make(chan struct{})
mtc.transport.Listen(store1.StoreID(), RaftMessageHandlerInterceptor{
RaftMessageHandler: store1,
handleSnapshotFilter: func(header *storage.SnapshotRequest_Header) {
snapshotSentCh <- struct{}{}
<-allowSnapshotCh
},
tc := testcluster.StartTestCluster(t, 2, base.TestClusterArgs{
ReplicationMode: base.ReplicationManual,
})
defer tc.Stopper().Stop(ctx)

// Launch the add replicas request. This will get stuck sending a preemptive
// snapshot thanks to the handler we installed above.
errCh := make(chan error)
scratchStartKey := tc.ScratchRange(t)
origDesc := tc.LookupRangeOrFatal(t, scratchStartKey)
splitKey := scratchStartKey.Next()
beforeDesc, _ := tc.SplitRangeOrFatal(t, splitKey)

mergeErrCh, addErrCh := make(chan error, 1), make(chan error, 1)
go func() {
errCh <- mtc.replicateRangeNonFatal(rhsDesc.RangeID, 1)
mergeErrCh <- tc.Server(0).DB().AdminMerge(ctx, scratchStartKey)
}()

// Wait for the add replicas request to send a snapshot, then merge the ranges
// together.
<-snapshotSentCh
mergeArgs := adminMergeArgs(lhsDesc.StartKey.AsRawKey())
if _, pErr := client.SendWrapped(ctx, store0.TestSender(), mergeArgs); pErr != nil {
t.Fatal(pErr)
}

// Allow snapshots, which will allow the add replicas to complete.
close(allowSnapshotCh)
expErr := `change replicas of r2 failed: descriptor changed: .* \(range subsumed\)`
if err := <-errCh; !testutils.IsError(err, expErr) {
t.Fatalf("expected error %q, got %v", expErr, err)
go func() {
targets := []roachpb.ReplicationTarget{tc.Target(1)}
_, err := tc.Server(0).DB().AdminChangeReplicas(
ctx, scratchStartKey, roachpb.ADD_REPLICA, targets, beforeDesc)
addErrCh <- err
}()
mergeErr := <-mergeErrCh
addErr := <-addErrCh
afterDesc := tc.LookupRangeOrFatal(t, scratchStartKey)

const acceptableMergeErr = `unexpected value: raw_bytes|ranges not collocated` +
`|cannot merge range with non-voter replicas`
if mergeErr == nil && testutils.IsError(addErr, `descriptor changed: \[expected\]`) {
// Merge won the race, no add happened.
require.Len(t, afterDesc.Replicas().Voters(), 1)
require.Equal(t, origDesc.EndKey, afterDesc.EndKey)
} else if addErr == nil && testutils.IsError(mergeErr, acceptableMergeErr) {
// Add won the race, no merge happened.
require.Len(t, afterDesc.Replicas().Voters(), 2)
require.Equal(t, beforeDesc.EndKey, afterDesc.EndKey)
} else {
t.Fatalf(`expected exactly one of merge or add to succeed got: [merge] %v [add] %v`,
mergeErr, addErr)
}
}

// TestStoreRangeMergeResplitAddReplicaRace tests a diabolical edge case in the
// merge/add replica race.
//
// Consider the same situation as described in the comment on
// TestStoreRangeMergeAddReplicaRace, except that Q and R merge together and
// then split at exactly the same key, all before the replica-change transaction
// starts. Q's range descriptor will have the same start key, end key, and next
// replica ID that it did when the preemptive snapshot started. That is, it will
// look unchanged! To protect against this, range descriptors contain a
// generation counter, which is incremented on every split or merge. The
// presence of this counter means that ChangeReplicas commands can detect and
// abort if any merges have occurred since the preemptive snapshot, even if the
// sequence of splits or merges left the keyspan of the range unchanged. This
// diabolical edge case is tested here.
// merge/add replica race. If two replicas merge and then split at the previous
// boundary, the descriptor will look unchanged and our usual CPut protection
// would fail. For this reason, we introduced RangeDescriptor.Generation.
//
// Note that splits will not increment the generation counter until the cluster
// version includes VersionRangeMerges. That's ok, because a sequence of splits
// alone will always result in a descriptor with a smaller end key. Only a
// sequence of splits AND merges can result in an unchanged end key, and merges
// always increment the generation counter.
func TestStoreRangeMergeResplitAddReplicaRace(t *testing.T) {
func TestDanStressStoreRangeMergeResplitAddReplicaRace(t *testing.T) {
defer leaktest.AfterTest(t)()

ctx := context.Background()
storeCfg := storage.TestStoreConfig(nil)
storeCfg.TestingKnobs.DisableSplitQueue = true
storeCfg.TestingKnobs.DisableMergeQueue = true
storeCfg.TestingKnobs.DisableReplicateQueue = true

mtc := &multiTestContext{storeConfig: &storeCfg}
mtc.Start(t, 2)
defer mtc.Stop()
store0, store1 := mtc.Store(0), mtc.Store(1)

lhsDesc, rhsDesc, err := createSplitRanges(ctx, store0)
if err != nil {
t.Fatal(err)
}

mtc.transport.Listen(store1.StoreID(), RaftMessageHandlerInterceptor{
RaftMessageHandler: store1,
handleSnapshotFilter: func(header *storage.SnapshotRequest_Header) {
mergeArgs := adminMergeArgs(lhsDesc.StartKey.AsRawKey())
if _, pErr := client.SendWrapped(ctx, store0.TestSender(), mergeArgs); pErr != nil {
// The filter is invoked by a different goroutine, so can't use t.Fatal.
t.Error(pErr)
return
}
splitArgs := adminSplitArgs(rhsDesc.StartKey.AsRawKey())
if _, pErr := client.SendWrapped(ctx, store0.TestSender(), splitArgs); pErr != nil {
// The filter is invoked by a different goroutine, so can't use t.Fatal.
t.Error(pErr)
return
}
},
tc := testcluster.StartTestCluster(t, 2, base.TestClusterArgs{
ReplicationMode: base.ReplicationManual,
})

err = mtc.replicateRangeNonFatal(lhsDesc.RangeID, 1)
if exp := "change replicas of r.* failed: descriptor changed"; !testutils.IsError(err, exp) {
t.Fatalf("expected error %q, got %v", exp, err)
defer tc.Stopper().Stop(ctx)

scratchStartKey := tc.ScratchRange(t)
splitKey := scratchStartKey.Next()
origDesc, _ := tc.SplitRangeOrFatal(t, splitKey)
require.NoError(t, tc.Server(0).DB().AdminMerge(ctx, scratchStartKey))
resplitDesc, _ := tc.SplitRangeOrFatal(t, splitKey)

assert.Equal(t, origDesc.RangeID, resplitDesc.RangeID)
assert.Equal(t, origDesc.StartKey, resplitDesc.StartKey)
assert.Equal(t, origDesc.EndKey, resplitDesc.EndKey)
assert.Equal(t, origDesc.Replicas().All(), resplitDesc.Replicas().All())
assert.NotEqual(t, origDesc.Generation, resplitDesc.Generation)

targets := []roachpb.ReplicationTarget{tc.Target(1)}
_, err := tc.Server(0).DB().AdminChangeReplicas(
ctx, scratchStartKey, roachpb.ADD_REPLICA, targets, origDesc)
if !testutils.IsError(err, `descriptor changed`) {
t.Fatalf(`expected "descriptor changed" error got: %+v`, err)
}
}

Expand Down Expand Up @@ -2323,117 +2264,7 @@ func TestStoreRangeMergeDeadFollowerDuringTxn(t *testing.T) {
}
}

func TestStoreRangeMergeReadoptedBothFollowers(t *testing.T) {
defer leaktest.AfterTest(t)()

ctx := context.Background()
storeCfg := storage.TestStoreConfig(nil)
storeCfg.TestingKnobs.DisableReplicateQueue = true
storeCfg.TestingKnobs.DisableReplicaGCQueue = true
storeCfg.TestingKnobs.DisableMergeQueue = true
mtc := &multiTestContext{storeConfig: &storeCfg}
mtc.Start(t, 3)
defer mtc.Stop()
store0, store2 := mtc.Store(0), mtc.Store(2)
distSender := mtc.distSenders[0]

// Create two ranges on all nodes.
rngID := store0.LookupReplica(roachpb.RKey("a")).Desc().RangeID
mtc.replicateRange(rngID, 1, 2)
lhsDesc, rhsDesc, err := createSplitRanges(ctx, store0)
if err != nil {
t.Fatal(err)
}

// Wait for all stores to have fully processed the split.
for _, key := range []roachpb.Key{roachpb.Key("a"), roachpb.Key("b")} {
if _, pErr := client.SendWrapped(ctx, distSender, incrementArgs(key, 1)); pErr != nil {
t.Fatal(pErr)
}
mtc.waitForValues(key, []int64{1, 1, 1})
}

lhsRepl0, err := store0.GetReplica(lhsDesc.RangeID)
if err != nil {
t.Fatal(err)
}
lhsRepl2, err := store2.GetReplica(lhsDesc.RangeID)
if err != nil {
t.Fatal(err)
}
rhsRepl2, err := store2.GetReplica(rhsDesc.RangeID)
if err != nil {
t.Fatal(err)
}

// Abandon the two ranges on store2, but do not GC them.
mtc.unreplicateRange(lhsDesc.RangeID, 2)
mtc.unreplicateRange(rhsDesc.RangeID, 2)

// Merge the two ranges together.
args := adminMergeArgs(lhsDesc.StartKey.AsRawKey())
_, pErr := client.SendWrapped(ctx, store0.TestSender(), args)
if pErr != nil {
t.Fatal(pErr)
}

addLHSRepl2 := func() error {
for r := retry.StartWithCtx(ctx, retry.Options{}); r.Next(); {
_, err := lhsRepl0.ChangeReplicas(ctx, roachpb.ADD_REPLICA, roachpb.ReplicationTarget{
NodeID: store2.Ident.NodeID,
StoreID: store2.Ident.StoreID,
}, lhsRepl0.Desc(), storagepb.ReasonUnknown, t.Name())
if !testutils.IsError(err, "store busy applying snapshots") {
return err
}
}
t.Fatal("unreachable")
return nil
}

// Attempt to re-add the merged range to store2. The operation should fail
// because store2's LHS and RHS replicas intersect the merged range.
err = addLHSRepl2()
if exp := "cannot apply snapshot: snapshot intersects existing range"; !testutils.IsError(err, exp) {
t.Fatalf("expected %q error, but got %v", exp, err)
}

// GC the replica of the LHS on store2.
if err := store2.ManualReplicaGC(lhsRepl2); err != nil {
t.Fatal(err)
}
if _, err := store2.GetReplica(lhsDesc.RangeID); err == nil {
t.Fatal("lhs replica not destroyed on store2")
}

// Attempt to re-add the merged range to store2. The operation should fail
// again because store2's RHS still intersects the merged range.
err = addLHSRepl2()
if exp := "cannot apply snapshot: snapshot intersects existing range"; !testutils.IsError(err, exp) {
t.Fatalf("expected %q error, but got %v", exp, err)
}

// GC the replica of the RHS on store2.
if err := store2.ManualReplicaGC(rhsRepl2); err != nil {
t.Fatal(err)
}
if _, err := store2.GetReplica(rhsDesc.RangeID); err == nil {
t.Fatal("rhs replica not destroyed on store2")
}

// Attempt to re-add the merged range to store2 one last time. This time the
// operation should succeed because there are no remaining intersecting
// replicas.
if err := addLHSRepl2(); err != nil {
t.Fatal(err)
}

// Give store2 the lease to force all commands to be applied, including the
// ChangeReplicas.
mtc.transferLease(ctx, lhsDesc.RangeID, 0, 2)
}

func TestStoreRangeReadoptedLHSFollower(t *testing.T) {
func TestDanStressStoreRangeReadoptedLHSFollower(t *testing.T) {
defer leaktest.AfterTest(t)()

run := func(t *testing.T, withMerge bool) {
Expand Down Expand Up @@ -2492,7 +2323,7 @@ func TestStoreRangeReadoptedLHSFollower(t *testing.T) {
StoreID: mtc.idents[2].StoreID,
}},
*lhsDesc,
); !testutils.IsError(err, "cannot apply snapshot: snapshot intersects existing range") {
); !testutils.IsError(err, "descriptor changed") {
t.Fatal(err)
}

Expand Down Expand Up @@ -3268,8 +3099,9 @@ func TestStoreRangeMergeDuringShutdown(t *testing.T) {
}
}

func TestMergeQueue(t *testing.T) {
func TestDanStressMergeQueue(t *testing.T) {
defer leaktest.AfterTest(t)()
t.Skip(`WIP still not sure what's going on here`)

ctx := context.Background()
manualClock := hlc.NewManualClock(123)
Expand Down
Loading

0 comments on commit c4dcc42

Please sign in to comment.