kvserver: remove snapshot CanDecline flag

tbg · tbg · commit 2a1857f11b93 · 2021-10-29T23:10:36.000+02:00
Only preemptive snapshots specified CanDecline, and we haven't been
using those since 19.2. This also removes the corresponding rejection
enum reason `SnapshotResponse_DECLINED`, and retires the
`server.declined_reservation_timeout` cluster setting.

Release note: None
diff --git a/pkg/kv/kvserver/client_raft_test.go b/pkg/kv/kvserver/client_raft_test.go
@@ -1327,9 +1327,8 @@ func TestFailedSnapshotFillsReservation(t *testing.T) {
 	rep2Desc, found := desc.GetReplicaDescriptor(2)
 	require.True(t, found)
 	header := kvserver.SnapshotRequest_Header{
-		CanDecline: true,
-		RangeSize:  100,
-		State:      kvserverpb.ReplicaState{Desc: desc},
+		RangeSize: 100,
+		State:     kvserverpb.ReplicaState{Desc: desc},
 		RaftMessageRequest: kvserver.RaftMessageRequest{
 			RangeID:     rep.RangeID,
 			FromReplica: repDesc,
diff --git a/pkg/kv/kvserver/raft.pb.go b/pkg/kv/kvserver/raft.pb.go
diff --git a/pkg/kv/kvserver/raft.proto b/pkg/kv/kvserver/raft.proto
@@ -153,8 +153,6 @@ message SnapshotRequest {
   }
 
   message Header {
-    reserved 1;
-
     // The replica state at the time the snapshot was generated. Note
     // that ReplicaState.Desc differs from the above range_descriptor
     // field which holds the updated descriptor after the new replica
@@ -168,11 +166,6 @@ message SnapshotRequest {
     // The estimated size of the range, to be used in reservation decisions.
     int64 range_size = 3;
 
-    // can_decline is set on preemptive snapshots, but not those generated
-    // by raft because at that point it is better to queue up the stream
-    // than to cancel it.
-    bool can_decline = 4;
-
     // The priority of the snapshot.
     Priority priority = 6;
 
@@ -193,6 +186,8 @@ message SnapshotRequest {
     //
     // TODO(irfansharif): Remove this in v22.1.
     bool deprecated_unreplicated_truncated_state = 8;
+
+    reserved 1, 4;
   }
 
   Header header = 1;
@@ -214,7 +209,7 @@ message SnapshotResponse {
     ACCEPTED = 1;
     APPLIED = 2;
     ERROR = 3;
-    DECLINED = 4;
+    reserved 4;
   }
   Status status = 1;
   string message = 2;
diff --git a/pkg/kv/kvserver/replica_command.go b/pkg/kv/kvserver/replica_command.go
@@ -2467,14 +2467,9 @@ func (r *Replica) sendSnapshot(
 			},
 		},
 		RangeSize: r.GetMVCCStats().Total(),
-		// Recipients currently cannot choose to decline any snapshots.
-		// In 19.2 and earlier versions pre-emptive snapshots could be declined.
-		//
-		// TODO(ajwerner): Consider removing the CanDecline flag.
-		CanDecline: false,
-		Priority:   priority,
-		Strategy:   SnapshotRequest_KV_BATCH,
-		Type:       snapType,
+		Priority:  priority,
+		Strategy:  SnapshotRequest_KV_BATCH,
+		Type:      snapType,
 	}
 	newBatchFn := func() storage.Batch {
 		return r.store.Engine().NewUnindexedBatch(true /* writeOnly */)
diff --git a/pkg/kv/kvserver/store_pool.go b/pkg/kv/kvserver/store_pool.go
@@ -41,16 +41,6 @@ const (
 	TestTimeUntilStoreDeadOff = 24 * time.Hour
 )
 
-// DeclinedReservationsTimeout specifies a duration during which the local
-// replicate queue will not consider stores which have rejected a reservation a
-// viable target.
-var DeclinedReservationsTimeout = settings.RegisterDurationSetting(
-	"server.declined_reservation_timeout",
-	"the amount of time to consider the store throttled for up-replication after a reservation was declined",
-	1*time.Second,
-	settings.NonNegativeDuration,
-)
-
 // FailedReservationsTimeout specifies a duration during which the local
 // replicate queue will not consider stores which have failed a reservation a
 // viable target.
@@ -914,7 +904,6 @@ type throttleReason int
 
 const (
 	_ throttleReason = iota
-	throttleDeclined
 	throttleFailed
 )
 
@@ -929,19 +918,10 @@ func (sp *StorePool) throttle(reason throttleReason, why string, storeID roachpb
 	detail := sp.getStoreDetailLocked(storeID)
 	detail.throttledBecause = why
 
-	// If a snapshot is declined, be it due to an error or because it was
-	// rejected, we mark the store detail as having been declined so it won't
-	// be considered as a candidate for new replicas until after the configured
-	// timeout period has passed.
+	// If a snapshot is declined, we mark the store detail as having been declined
+	// so it won't be considered as a candidate for new replicas until after the
+	// configured timeout period has passed.
 	switch reason {
-	case throttleDeclined:
-		timeout := DeclinedReservationsTimeout.Get(&sp.st.SV)
-		detail.throttledUntil = sp.clock.PhysicalTime().Add(timeout)
-		if log.V(2) {
-			ctx := sp.AnnotateCtx(context.TODO())
-			log.Infof(ctx, "snapshot declined (%s), s%d will be throttled for %s until %s",
-				why, storeID, timeout, detail.throttledUntil)
-		}
 	case throttleFailed:
 		timeout := FailedReservationsTimeout.Get(&sp.st.SV)
 		detail.throttledUntil = sp.clock.PhysicalTime().Add(timeout)
@@ -950,6 +930,8 @@ func (sp *StorePool) throttle(reason throttleReason, why string, storeID roachpb
 			log.Infof(ctx, "snapshot failed (%s), s%d will be throttled for %s until %s",
 				why, storeID, timeout, detail.throttledUntil)
 		}
+	default:
+		log.Warningf(sp.AnnotateCtx(context.TODO()), "unknown throttle reason %v", reason)
 	}
 }
 
diff --git a/pkg/kv/kvserver/store_pool_test.go b/pkg/kv/kvserver/store_pool_test.go
@@ -835,30 +835,15 @@ func TestStorePoolThrottle(t *testing.T) {
 	sg := gossiputil.NewStoreGossiper(g)
 	sg.GossipStores(uniqueStore, t)
 
-	{
-		expected := sp.clock.Now().GoTime().Add(DeclinedReservationsTimeout.Get(&sp.st.SV))
-		sp.throttle(throttleDeclined, "", 1)
-
-		sp.detailsMu.Lock()
-		detail := sp.getStoreDetailLocked(1)
-		sp.detailsMu.Unlock()
-		if !detail.throttledUntil.Equal(expected) {
-			t.Errorf("expected store to have been throttled to %v, found %v",
-				expected, detail.throttledUntil)
-		}
-	}
+	expected := sp.clock.Now().GoTime().Add(FailedReservationsTimeout.Get(&sp.st.SV))
+	sp.throttle(throttleFailed, "", 1)
 
-	{
-		expected := sp.clock.Now().GoTime().Add(FailedReservationsTimeout.Get(&sp.st.SV))
-		sp.throttle(throttleFailed, "", 1)
-
-		sp.detailsMu.Lock()
-		detail := sp.getStoreDetailLocked(1)
-		sp.detailsMu.Unlock()
-		if !detail.throttledUntil.Equal(expected) {
-			t.Errorf("expected store to have been throttled to %v, found %v",
-				expected, detail.throttledUntil)
-		}
+	sp.detailsMu.Lock()
+	detail := sp.getStoreDetailLocked(1)
+	sp.detailsMu.Unlock()
+	if !detail.throttledUntil.Equal(expected) {
+		t.Errorf("expected store to have been throttled to %v, found %v",
+			expected, detail.throttledUntil)
 	}
 }
 
diff --git a/pkg/kv/kvserver/store_raft.go b/pkg/kv/kvserver/store_raft.go
@@ -74,7 +74,7 @@ func (s *Store) HandleSnapshot(
 
 		if s.IsDraining() {
 			return stream.Send(&SnapshotResponse{
-				Status:  SnapshotResponse_DECLINED,
+				Status:  SnapshotResponse_ERROR,
 				Message: storeDrainingMsg,
 			})
 		}
diff --git a/pkg/kv/kvserver/store_snapshot.go b/pkg/kv/kvserver/store_snapshot.go
@@ -40,9 +40,7 @@ import (
 
 const (
 	// Messages that provide detail about why a snapshot was rejected.
-	snapshotStoreTooFullMsg = "store almost out of disk space"
-	snapshotApplySemBusyMsg = "store busy applying snapshots"
-	storeDrainingMsg        = "store is draining"
+	storeDrainingMsg = "store is draining"
 
 	// IntersectingSnapshotMsg is part of the error message returned from
 	// canAcceptSnapshotLocked and is exposed here so testing can rely on it.
@@ -500,38 +498,23 @@ func (kvSS *kvBatchSnapshotStrategy) Close(ctx context.Context) {
 }
 
 // reserveSnapshot throttles incoming snapshots. The returned closure is used
-// to cleanup the reservation and release its resources. A nil cleanup function
-// and a non-empty rejectionMessage indicates the reservation was declined.
+// to cleanup the reservation and release its resources.
 func (s *Store) reserveSnapshot(
 	ctx context.Context, header *SnapshotRequest_Header,
-) (_cleanup func(), _rejectionMsg string, _err error) {
+) (_cleanup func(), _err error) {
 	tBegin := timeutil.Now()
-	if header.RangeSize == 0 {
-		// Empty snapshots are exempt from rate limits because they're so cheap to
-		// apply. This vastly speeds up rebalancing any empty ranges created by a
-		// RESTORE or manual SPLIT AT, since it prevents these empty snapshots from
-		// getting stuck behind large snapshots managed by the replicate queue.
-	} else if header.CanDecline {
-		storeDesc, ok := s.cfg.StorePool.getStoreDescriptor(s.StoreID())
-		if ok && (!maxCapacityCheck(storeDesc) || header.RangeSize > storeDesc.Capacity.Available) {
-			return nil, snapshotStoreTooFullMsg, nil
-		}
-		select {
-		case s.snapshotApplySem <- struct{}{}:
-		case <-ctx.Done():
-			return nil, "", ctx.Err()
-		case <-s.stopper.ShouldQuiesce():
-			return nil, "", errors.Errorf("stopped")
-		default:
-			return nil, snapshotApplySemBusyMsg, nil
-		}
-	} else {
+
+	// Empty snapshots are exempt from rate limits because they're so cheap to
+	// apply. This vastly speeds up rebalancing any empty ranges created by a
+	// RESTORE or manual SPLIT AT, since it prevents these empty snapshots from
+	// getting stuck behind large snapshots managed by the replicate queue.
+	if header.RangeSize != 0 {
 		select {
 		case s.snapshotApplySem <- struct{}{}:
 		case <-ctx.Done():
-			return nil, "", ctx.Err()
+			return nil, ctx.Err()
 		case <-s.stopper.ShouldQuiesce():
-			return nil, "", errors.Errorf("stopped")
+			return nil, errors.Errorf("stopped")
 		}
 	}
 
@@ -560,7 +543,7 @@ func (s *Store) reserveSnapshot(
 		if header.RangeSize != 0 {
 			<-s.snapshotApplySem
 		}
-	}, "", nil
+	}, nil
 }
 
 // canAcceptSnapshotLocked returns (_, nil) if the snapshot can be applied to
@@ -716,16 +699,10 @@ func (s *Store) receiveSnapshot(
 			header.Type, storeID, header.State.Desc.Replicas())
 	}
 
-	cleanup, rejectionMsg, err := s.reserveSnapshot(ctx, header)
+	cleanup, err := s.reserveSnapshot(ctx, header)
 	if err != nil {
 		return err
 	}
-	if cleanup == nil {
-		return stream.Send(&SnapshotResponse{
-			Status:  SnapshotResponse_DECLINED,
-			Message: rejectionMsg,
-		})
-	}
 	defer cleanup()
 
 	// The comment on ReplicaPlaceholder motivates and documents
@@ -995,7 +972,6 @@ func SendEmptySnapshot(
 		State:                                state,
 		RaftMessageRequest:                   req,
 		RangeSize:                            ms.Total(),
-		CanDecline:                           false,
 		Priority:                             SnapshotRequest_RECOVERY,
 		Strategy:                             SnapshotRequest_KV_BATCH,
 		Type:                                 SnapshotRequest_VIA_SNAPSHOT_QUEUE,
@@ -1056,20 +1032,6 @@ func sendSnapshot(
 		return err
 	}
 	switch resp.Status {
-	case SnapshotResponse_DECLINED:
-		if header.CanDecline {
-			declinedMsg := "reservation rejected"
-			if len(resp.Message) > 0 {
-				declinedMsg = resp.Message
-			}
-			err := &benignError{errors.Errorf("%s: remote declined %s: %s", to, snap, declinedMsg)}
-			storePool.throttle(throttleDeclined, err.Error(), to.StoreID)
-			return err
-		}
-		err := errors.Errorf("%s: programming error: remote declined required %s: %s",
-			to, snap, resp.Message)
-		storePool.throttle(throttleFailed, err.Error(), to.StoreID)
-		return err
 	case SnapshotResponse_ERROR:
 		storePool.throttle(throttleFailed, resp.Message, to.StoreID)
 		return errors.Errorf("%s: remote couldn't accept %s with error: %s",
diff --git a/pkg/kv/kvserver/store_test.go b/pkg/kv/kvserver/store_test.go
diff --git a/pkg/settings/registry.go b/pkg/settings/registry.go

Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,7 @@ func (s *Store) HandleSnapshot(`
`74`	`74`
`75`	`75`	`if s.IsDraining() {`
`76`	`76`	`return stream.Send(&SnapshotResponse{`
`77`		`- Status: SnapshotResponse_DECLINED,`
	`77`	`+ Status: SnapshotResponse_ERROR,`
`78`	`78`	`Message: storeDrainingMsg,`
`79`	`79`	`})`
`80`	`80`	`}`