mmaintegration: improve logging around replace lease changes

This commit improves logging around allocator sync to make it easier
to observe replica lease changes and understand thrashing issues.

Author: wenyihu6

pkg/kv/kvserver/asim/asim.go

@@ -309,7 +309,7 @@ func (s *Simulator) tickWorkload(ctx context.Context, tick time.Time) {
 // each store ticks the pending operations such as relocate range and lease
 // transfers.
 func (s *Simulator) tickStateChanges(ctx context.Context, tick time.Time) {
-	s.changer.Tick(tick, s.state)
+	s.changer.Tick(ctx, tick, s.state)
 	stores := s.state.Stores()
 	s.shuffler(len(stores), func(i, j int) { stores[i], stores[j] = stores[j], stores[i] })
 	for _, store := range stores {
@@ -371,7 +371,7 @@ func (s *Simulator) tickQueues(ctx context.Context, tick time.Time, state state.
 
 		// Tick changes that may have been enqueued with a lower completion
 		// than the current tick, from the queues.
-		s.changer.Tick(tick, state)
+		s.changer.Tick(ctx, tick, state)
 
 		// Try adding suggested load splits that are pending for this store.
 		for _, rangeID := range state.LoadSplitterFor(storeID).ClearSplitKeys() {

pkg/kv/kvserver/asim/metrics/metrics_test.go

@@ -90,7 +90,7 @@ func Example_leaseTransfer() {
 		Author:         1,
 		Wait:           0,
 	})
-	changer.Tick(state.TestingStartTime(), s)
+	changer.Tick(ctx, state.TestingStartTime(), s)
 	m.Tick(ctx, start, s)
 	// Output:
 	//tick,c_ranges,c_write,c_write_b,c_read,c_read_b,s_ranges,s_write,s_write_b,s_read,s_read_b,c_lease_moves,c_replica_moves,c_replica_b_moves
@@ -120,7 +120,7 @@ func Example_rebalance() {
 		})...),
 		Wait: 0,
 	}
-	c.Apply(s)
+	c.Apply(ctx, s)
 
 	m.Tick(ctx, start, s)
 	// Output:

pkg/kv/kvserver/asim/mmaintegration/mma_store_rebalancer.go

@@ -204,7 +204,7 @@ func (msr *MMAStoreRebalancer) Tick(ctx context.Context, tick time.Time, s state
 		}
 		log.KvDistribution.VInfof(ctx, 1, "dispatching operation for pendingChange=%v", curChange)
 		msr.pendingChanges[msr.pendingChangeIdx].syncChangeID =
-			msr.as.MMAPreApply(curChange.usage, curChange.change)
+			msr.as.MMAPreApply(ctx, curChange.usage, curChange.change)
 		msr.pendingTicket = msr.controller.Dispatch(ctx, tick, s, curOp)
 	}
 }

pkg/kv/kvserver/asim/op/controller_test.go

@@ -95,7 +95,7 @@ func TestLeaseTransferOp(t *testing.T) {
 			results := make([]map[state.RangeID]state.StoreID, len(tc.ticks))
 			pending := []DispatchedTicket{}
 			for i, tick := range tc.ticks {
-				changer.Tick(state.OffsetTick(start, tick), s)
+				changer.Tick(ctx, state.OffsetTick(start, tick), s)
 				controller.Tick(ctx, state.OffsetTick(start, tick), s)
 
 				for _, transfers := range tc.transfers[tick] {
@@ -305,7 +305,7 @@ func TestRelocateRangeOp(t *testing.T) {
 				// range rebalancer will fail if any pending changes that were
 				// set to complete at tick t, still exist at tick t. So we tick
 				// it first here.
-				changer.Tick(state.OffsetTick(start, tick), s)
+				changer.Tick(ctx, state.OffsetTick(start, tick), s)
 				controller.Tick(ctx, state.OffsetTick(start, tick), s)
 
 				relocations := tc.relocations[tick]

pkg/kv/kvserver/asim/queue/lease_queue_test.go

@@ -172,7 +172,7 @@ func TestLeaseQueue(t *testing.T) {
 				s.TickClock(state.OffsetTick(start, tick))
 
 				// Tick state updates that are queued for completion.
-				changer.Tick(state.OffsetTick(start, tick), s)
+				changer.Tick(ctx, state.OffsetTick(start, tick), s)
 
 				// Update the store's view of the cluster, we update all stores
 				// but only care about s1's view.

pkg/kv/kvserver/asim/queue/replicate_queue.go

@@ -188,6 +188,7 @@ func pushReplicateChange(
 		if as != nil {
 			// as may be nil in some tests.
 			changeID = as.NonMMAPreTransferLease(
+				ctx,
 				repl.Desc(),
 				repl.RangeUsageInfo(),
 				op.Source,
@@ -204,6 +205,7 @@ func pushReplicateChange(
 		if as != nil {
 			// as may be nil in some tests.
 			changeID = as.NonMMAPreChangeReplicas(
+				ctx,
 				repl.Desc(),
 				repl.RangeUsageInfo(),
 				op.Chgs,

pkg/kv/kvserver/asim/queue/replicate_queue_test.go

@@ -321,7 +321,7 @@ func TestReplicateQueue(t *testing.T) {
 				s.TickClock(state.OffsetTick(start, tick))
 
 				// Tick state updates that are queued for completion.
-				changer.Tick(state.OffsetTick(start, tick), s)
+				changer.Tick(ctx, state.OffsetTick(start, tick), s)
 
 				// Update the store's view of the cluster, we update all stores
 				// but only care about s1's view.

pkg/kv/kvserver/asim/queue/split_queue_test.go

@@ -162,7 +162,7 @@ func TestSplitQueue(t *testing.T) {
 				sq.Tick(ctx, state.OffsetTick(start, tick), s)
 
 				// Tick state updates that are queued for completion.
-				changer.Tick(state.OffsetTick(start, tick), s)
+				changer.Tick(ctx, state.OffsetTick(start, tick), s)
 
 				// Check every replica on the leaseholder store for enqueuing.
 				for _, repl := range s.Replicas(store.StoreID()) {

pkg/kv/kvserver/asim/state/change.go

@@ -6,19 +6,21 @@
 package state
 
 import (
+	"context"
 	"fmt"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/google/btree"
 )
 
 // Change is a state change for a range, to a target store that has some delay.
 type Change interface {
 	// Apply applies a change to the state.
-	Apply(s State)
+	Apply(ctx context.Context, s State)
 	// Target returns the recipient store of the change.
 	Target() StoreID
 	// Range returns the range id the change is for.
@@ -40,7 +42,7 @@ type Changer interface {
 	Push(tick time.Time, sc Change) (time.Time, bool)
 	// Tick updates state changer to apply any changes that have occurred
 	// between the last tick and this one.
-	Tick(tick time.Time, state State)
+	Tick(ctx context.Context, tick time.Time, state State)
 }
 
 // ReplicaChange contains information necessary to add, remove or move (both) a
@@ -70,9 +72,10 @@ type LeaseTransferChange struct {
 }
 
 // Apply applies a change to the state.
-func (lt *LeaseTransferChange) Apply(s State) {
+func (lt *LeaseTransferChange) Apply(ctx context.Context, s State) {
 	if s.TransferLease(lt.RangeID, lt.TransferTarget) {
 		s.ClusterUsageInfo().storeRef(lt.Author).LeaseTransfers++
+		log.KvDistribution.VEventf(ctx, 3, "r%d: s%d transferred lease to s%d (took %s)", lt.RangeID, lt.Author, lt.TransferTarget, lt.Wait)
 	}
 }
 
@@ -99,9 +102,10 @@ func (lt *LeaseTransferChange) Blocking() bool {
 }
 
 // Apply applies a change to the state.
-func (rsc *RangeSplitChange) Apply(s State) {
+func (rsc *RangeSplitChange) Apply(ctx context.Context, s State) {
 	if _, _, ok := s.SplitRange(rsc.SplitKey); ok {
 		s.ClusterUsageInfo().storeRef(rsc.Author).RangeSplits++
+		log.KvDistribution.VEventf(ctx, 3, "r%d: s%d split range at key=%d (took %s)", rsc.RangeID, rsc.Author, rsc.SplitKey, rsc.Wait)
 	}
 }
 
@@ -183,7 +187,7 @@ func replChangeHasStoreID(storeID StoreID, changes []roachpb.ReplicationTarget)
 // Apply applies a replica change for a range. This is an implementation of the
 // Change interface. It requires that a replica being removed, must not hold
 // the lease unless a replica is also being added in the same change.
-func (rc *ReplicaChange) Apply(s State) {
+func (rc *ReplicaChange) Apply(ctx context.Context, s State) {
 	if len(rc.Changes) == 0 {
 		// Nothing to do.
 		return
@@ -197,6 +201,11 @@ func (rc *ReplicaChange) Apply(s State) {
 	rangeID := rc.RangeID
 
 	defer func() {
+		if rollback != nil {
+			log.KvDistribution.VEventf(ctx, 5, "r%d: s%d failed to apply replica change (after %s)", rangeID, rc.Author, rc.Wait)
+		} else {
+			log.KvDistribution.VEventf(ctx, 5, "r%d: s%d successfully applied replica change (after %s)", rangeID, rc.Author, rc.Wait)
+		}
 		n := len(rollback)
 		for i := n - 1; i > -1; i-- {
 			if rollback[i] != nil {
@@ -275,6 +284,7 @@ func (rc *ReplicaChange) Apply(s State) {
 	for _, nonVoterPromotion := range targets.NonVoterPromotions {
 		ok, revert := promoDemo(
 			s, rangeID, StoreID(nonVoterPromotion.StoreID), roachpb.NON_VOTER, roachpb.VOTER_FULL)
+		log.KvDistribution.VEventf(ctx, 5, "r%d: author s%d promoted non-voter s%s to voter", rangeID, rc.Author, nonVoterPromotion.StoreID)
 		rollback = append(rollback, revert...)
 		if !ok {
 			return
@@ -283,6 +293,7 @@ func (rc *ReplicaChange) Apply(s State) {
 	for _, voterAddition := range targets.VoterAdditions {
 		ok, revert := addReplica(
 			s, rangeID, StoreID(voterAddition.StoreID), roachpb.VOTER_FULL)
+		log.KvDistribution.VEventf(ctx, 5, "r%d: author s%d added voter s%s", rangeID, rc.Author, voterAddition.StoreID)
 		rollback = append(rollback, revert)
 		if !ok {
 			return
@@ -296,6 +307,7 @@ func (rc *ReplicaChange) Apply(s State) {
 		if !s.TransferLease(rangeID, nextLH) {
 			return
 		}
+		log.KvDistribution.VEventf(ctx, 5, "r%d: author s%d transferred lease to s%d", rangeID, rc.Author, nextLH)
 		rollback = append(rollback, func() {
 			if !s.TransferLease(rangeID, lhStore.StoreID()) {
 				panic("unable to rollback lease transfer")
@@ -306,6 +318,7 @@ func (rc *ReplicaChange) Apply(s State) {
 	for _, voterDemotion := range targets.VoterDemotions {
 		ok, revert := promoDemo(
 			s, rangeID, StoreID(voterDemotion.StoreID), roachpb.VOTER_FULL, roachpb.NON_VOTER)
+		log.KvDistribution.VEventf(ctx, 5, "r%d: author s%d demoted voter s%s to non-voter", rangeID, rc.Author, voterDemotion.StoreID)
 		rollback = append(rollback, revert...)
 		if !ok {
 			return
@@ -314,6 +327,7 @@ func (rc *ReplicaChange) Apply(s State) {
 	for _, voterRemoval := range targets.VoterRemovals {
 		ok, revert := removeReplica(
 			s, rangeID, StoreID(voterRemoval.StoreID), roachpb.VOTER_FULL)
+		log.KvDistribution.VEventf(ctx, 5, "r%d: author s%d removed voter s%s", rangeID, rc.Author, voterRemoval.StoreID)
 		rollback = append(rollback, revert)
 		if !ok {
 			return
@@ -322,6 +336,7 @@ func (rc *ReplicaChange) Apply(s State) {
 	for _, nonVoterAddition := range targets.NonVoterAdditions {
 		ok, revert := addReplica(
 			s, rangeID, StoreID(nonVoterAddition.StoreID), roachpb.NON_VOTER)
+		log.KvDistribution.VEventf(ctx, 5, "r%d: author s%d added non-voter s%s", rangeID, rc.Author, nonVoterAddition.StoreID)
 		rollback = append(rollback, revert)
 		if !ok {
 			return
@@ -330,6 +345,7 @@ func (rc *ReplicaChange) Apply(s State) {
 	for _, nonVoterRemoval := range targets.NonVoterRemovals {
 		ok, revert := removeReplica(
 			s, rangeID, StoreID(nonVoterRemoval.StoreID), roachpb.NON_VOTER)
+		log.KvDistribution.VEventf(ctx, 5, "r%d: author s%d removed non-voter s%s", rangeID, rc.Author, nonVoterRemoval.StoreID)
 		rollback = append(rollback, revert)
 		if !ok {
 			return
@@ -345,6 +361,7 @@ func (rc *ReplicaChange) Apply(s State) {
 
 		authorUsageInfo := s.ClusterUsageInfo().storeRef(rc.Author)
 		authorUsageInfo.Rebalances++
+		log.KvDistribution.VEventf(ctx, 5, "r%d: author s%d rebalancing", rangeID, rc.Author)
 		if requiresUpReplication {
 			authorUsageInfo.RebalanceSentBytes += r.Size()
 			s.ClusterUsageInfo().storeRef(storeNeedingSnapshot).RebalanceRcvdBytes += r.Size()
@@ -470,7 +487,7 @@ func (rc *replicaChanger) Push(tick time.Time, change Change) (time.Time, bool)
 
 // Tick updates state changer to apply any changes that have occurred
 // between the last tick and this one.
-func (rc *replicaChanger) Tick(tick time.Time, state State) {
+func (rc *replicaChanger) Tick(ctx context.Context, tick time.Time, state State) {
 	var changeList []*pendingChange
 
 	// NB: Add the smallest unit of time, in order to find all items in
@@ -483,7 +500,7 @@ func (rc *replicaChanger) Tick(tick time.Time, state State) {
 
 	for _, nextChange := range changeList {
 		change := rc.pendingTickets[nextChange.ticket]
-		change.Apply(state)
+		change.Apply(ctx, state)
 
 		// Cleanup the pending trackers for this ticket. This allows another
 		// change to be pushed for Range().

pkg/kv/kvserver/asim/state/change_test.go

@@ -6,6 +6,7 @@
 package state
 
 import (
+	"context"
 	"sort"
 	"testing"
 	"time"
@@ -321,11 +322,12 @@ func TestReplicaChange(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.desc, func(t *testing.T) {
+			ctx := context.Background()
 			state := testMakeRangeState(tc.stores, tc.initVoters, tc.initNonVoters)
 			r, _ := state.Range(1)
 			state.TransferLease(r.RangeID(), StoreID(1))
 			change := tc.change(state)
-			change.Apply(state)
+			change.Apply(ctx, state)
 			voters := testGetReplLocations(state, r, roachpb.VOTER_FULL)
 			nonVoters := testGetReplLocations(state, r, roachpb.NON_VOTER)
 			leaseholder := StoreID(-1)
@@ -686,8 +688,9 @@ func TestReplicaStateChanger(t *testing.T) {
 			tsResults := make([]int64, 0, 1)
 			resultLeaseholders := make(map[int64]map[int64]StoreID)
 
+			ctx := context.Background()
 			for _, tick := range tc.ticks {
-				changer.Tick(OffsetTick(start, tick), state)
+				changer.Tick(ctx, OffsetTick(start, tick), state)
 				if change, ok := tc.pushes[tick]; ok {
 					if ts, ok := changer.Push(OffsetTick(start, tick), change(state)); ok {
 						tsResults = append(tsResults, ReverseOffsetTick(start, ts))