Merge #157029 #158537

157029: admission: compute cpu time token refill rates r=sumeerbhola a=joshimhoff

**admission: compute cpu time token refill rates**
    
This commit introduces a linear model that computes cpu
time token refill rates. cpuTimeTokenFiller uses this
model to determine how many tokens to add per second to
the buckets in cpuTimeTokenGranter.
    
Fixes: #154471
    
Release note: None.

158537: storeliveness: remember when support was withdrawn from a store  r=iskettaneh a=arulajmani

See individual commits.

Co-authored-by: Josh Imhoff <[email protected]>
Co-authored-by: Arul Ajmani <[email protected]>

Author: 3 people

pkg/inspectz/inspectz.go

@@ -59,11 +59,11 @@ func NewServer(
 	mux.Handle("/inspectz/v2/kvflowcontroller", server.makeKVFlowControllerHandler(server.KVFlowControllerV2))
 	mux.Handle(
 		"/inspectz/storeliveness/supportFrom",
-		server.makeStoreLivenessHandler(server.StoreLivenessSupportFrom),
+		server.makeStoreLivenessSupportFromHandler(server.StoreLivenessSupportFrom),
 	)
 	mux.Handle(
 		"/inspectz/storeliveness/supportFor",
-		server.makeStoreLivenessHandler(server.StoreLivenessSupportFor),
+		server.makeStoreLivenessSupportForHandler(server.StoreLivenessSupportFor),
 	)
 
 	return server
@@ -112,9 +112,27 @@ func (s *Server) makeKVFlowControllerHandler(
 	}
 }
 
-func (s *Server) makeStoreLivenessHandler(
+func (s *Server) makeStoreLivenessSupportFromHandler(
 	impl func(ctx context.Context, request *slpb.InspectStoreLivenessRequest) (
-		*slpb.InspectStoreLivenessResponse, error,
+		*slpb.InspectSupportFromResponse, error,
+	),
+) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		ctx := s.AnnotateCtx(context.Background())
+		req := &slpb.InspectStoreLivenessRequest{}
+		resp, err := impl(ctx, req)
+		if err != nil {
+			log.Dev.ErrorfDepth(ctx, 1, "%s", err)
+			http.Error(w, "internal error: check logs for details", http.StatusInternalServerError)
+			return
+		}
+		respond(ctx, w, http.StatusOK, resp)
+	}
+}
+
+func (s *Server) makeStoreLivenessSupportForHandler(
+	impl func(ctx context.Context, request *slpb.InspectStoreLivenessRequest) (
+		*slpb.InspectSupportForResponse, error,
 	),
 ) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
@@ -147,20 +165,20 @@ func (s *Server) KVFlowHandlesV2(
 // StoreLivenessSupportFrom implements the InspectzServer interface.
 func (s *Server) StoreLivenessSupportFrom(
 	_ context.Context, _ *slpb.InspectStoreLivenessRequest,
-) (*slpb.InspectStoreLivenessResponse, error) {
-	resp := &slpb.InspectStoreLivenessResponse{}
+) (*slpb.InspectSupportFromResponse, error) {
+	resp := &slpb.InspectSupportFromResponse{}
 	support, err := s.storeLiveness.InspectAllSupportFrom()
-	resp.SupportStatesPerStore = support
+	resp.SupportFromStatesPerStore = support
 	return resp, err
 }
 
 // StoreLivenessSupportFor implements the InspectzServer interface.
 func (s *Server) StoreLivenessSupportFor(
 	_ context.Context, _ *slpb.InspectStoreLivenessRequest,
-) (*slpb.InspectStoreLivenessResponse, error) {
-	resp := &slpb.InspectStoreLivenessResponse{}
+) (*slpb.InspectSupportForResponse, error) {
+	resp := &slpb.InspectSupportForResponse{}
 	support, err := s.storeLiveness.InspectAllSupportFor()
-	resp.SupportStatesPerStore = support
+	resp.SupportForStatesPerStore = support
 	return resp, err
 }
 

pkg/inspectz/inspectzpb/inspectz.proto

@@ -32,13 +32,13 @@ service Inspectz {
   // storeliveness.SupportManagers' views of support provided from other stores.
   // It's housed under /inspectz/storeliveness/supportFrom.
   rpc StoreLivenessSupportFrom(kv.kvserver.storeliveness.storelivenesspb.InspectStoreLivenessRequest)
-      returns (kv.kvserver.storeliveness.storelivenesspb.InspectStoreLivenessResponse) {}
+      returns (kv.kvserver.storeliveness.storelivenesspb.InspectSupportFromResponse) {}
 
   // StoreLivenessSupportFor exposes the in-memory state of all stores'
   // storeliveness.SupportManagers' views of support provided for other stores.
   // It's housed under /inspectz/storeliveness/supportFor.
   rpc StoreLivenessSupportFor(kv.kvserver.storeliveness.storelivenesspb.InspectStoreLivenessRequest)
-      returns (kv.kvserver.storeliveness.storelivenesspb.InspectStoreLivenessResponse) {}
+      returns (kv.kvserver.storeliveness.storelivenesspb.InspectSupportForResponse) {}
 
 }
 

pkg/inspectz/unsupported.go

@@ -51,13 +51,13 @@ func (u Unsupported) KVFlowHandlesV2(
 // StoreLivenessSupportFrom is part of the inspectzpb.InspectzServer interface.
 func (u Unsupported) StoreLivenessSupportFrom(
 	_ context.Context, _ *slpb.InspectStoreLivenessRequest,
-) (*slpb.InspectStoreLivenessResponse, error) {
+) (*slpb.InspectSupportFromResponse, error) {
 	return nil, errorutil.UnsupportedUnderClusterVirtualization(errorutil.FeatureNotAvailableToNonSystemTenantsIssue)
 }
 
 // StoreLivenessSupportFor is part of the inspectzpb.InspectzServer interface.
 func (u Unsupported) StoreLivenessSupportFor(
 	_ context.Context, _ *slpb.InspectStoreLivenessRequest,
-) (*slpb.InspectStoreLivenessResponse, error) {
+) (*slpb.InspectSupportForResponse, error) {
 	return nil, errorutil.UnsupportedUnderClusterVirtualization(errorutil.FeatureNotAvailableToNonSystemTenantsIssue)
 }

pkg/kv/kvserver/storeliveness/fabric.go

@@ -73,6 +73,6 @@ type Fabric interface {
 // InspectFabric is an interface that exposes all in-memory support state for a
 // given store. It is used to power the Store Liveness /inspectz functionality.
 type InspectFabric interface {
-	InspectSupportFrom() slpb.SupportStatesPerStore
-	InspectSupportFor() slpb.SupportStatesPerStore
+	InspectSupportFrom() slpb.InspectSupportFromStatesPerStore
+	InspectSupportFor() slpb.InspectSupportForStatesPerStore
 }

pkg/kv/kvserver/storeliveness/store_liveness_test.go

@@ -120,7 +120,11 @@ func TestStoreLiveness(t *testing.T) {
 					case "debug-supporter-state":
 						var sortedSupportMap []string
 						for _, support := range sm.supporterStateHandler.supporterState.supportFor {
-							sortedSupportMap = append(sortedSupportMap, fmt.Sprintf("%+v", support))
+							sortedSupportMap = append(sortedSupportMap, fmt.Sprintf(
+								"%+v lastSupportWithdrawnTime:%s",
+								support.SupportState,
+								support.lastSupportWithdrawnTime,
+							))
 						}
 						slices.Sort(sortedSupportMap)
 						return fmt.Sprintf(

pkg/kv/kvserver/storeliveness/storelivenesspb/service.proto

@@ -155,19 +155,42 @@ message SupportState {
   util.hlc.Timestamp expiration = 3 [(gogoproto.nullable) = false];
 }
 
+// InspectSupportForState is used for observability via inspectz. It wraps
+// SupportState and includes additional non-persisted metadata like
+// LastSupportWithdrawnTime.
+message InspectSupportForState {
+  SupportState       support_state                = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+  // LastSupportWithdrawnTime is the timestamp at which support was last
+  // withdrawn from the remote store. This field is not persisted to disk.
+  util.hlc.Timestamp last_support_withdrawn_time  = 2 [(gogoproto.nullable) = false];
+}
+
 // InspectStoreLivenessRequest is used to power the Store Liveness /inspectz
 // functionality. The request doesn't take any parameters.
 message InspectStoreLivenessRequest {}
 
-// InspectStoreLivenessRequest is used to power the Store Liveness /inspectz
-// functionality. The response is a list of SupportStatesPerStore.
-message InspectStoreLivenessResponse {
-  repeated SupportStatesPerStore support_states_per_store = 1 [(gogoproto.nullable) = false];
+// InspectSupportFromResponse is used to power the Store Liveness /inspectz
+// supportFrom functionality.
+message InspectSupportFromResponse {
+  repeated InspectSupportFromStatesPerStore support_from_states_per_store = 1 [(gogoproto.nullable) = false];
+}
+
+// InspectSupportForResponse is used to power the Store Liveness /inspectz
+// supportFor functionality.
+message InspectSupportForResponse {
+  repeated InspectSupportForStatesPerStore support_for_states_per_store = 1 [(gogoproto.nullable) = false];
+}
+
+// InspectSupportFromStatesPerStore includes all SupportStates for a given
+// store; they correspond to the support-from map of a given store.
+message InspectSupportFromStatesPerStore {
+  StoreIdent store_id = 1 [(gogoproto.nullable) = false, (gogoproto.customname) = "StoreID"];
+  repeated SupportState support_from_states = 2 [(gogoproto.nullable) = false];
 }
 
-// SupportStatesPerStore includes all SupportStates for a given store; they
-// correspond to either the support-from or support-for map of a given store.
-message SupportStatesPerStore {
+// InspectSupportForStatesPerStore includes all InspectSupportForStates for a
+// given store; they correspond to the support-for map of a given store.
+message InspectSupportForStatesPerStore {
   StoreIdent store_id = 1 [(gogoproto.nullable) = false, (gogoproto.customname) = "StoreID"];
-  repeated SupportState support_states = 2 [(gogoproto.nullable) = false];
+  repeated InspectSupportForState support_for_states = 2 [(gogoproto.nullable) = false];
 }

pkg/kv/kvserver/storeliveness/support_manager.go

@@ -121,15 +121,15 @@ func (sm *SupportManager) SupportFor(id slpb.StoreIdent) (slpb.Epoch, bool) {
 }
 
 // InspectSupportFrom implements the InspectFabric interface.
-func (sm *SupportManager) InspectSupportFrom() slpb.SupportStatesPerStore {
-	supportStates := sm.requesterStateHandler.exportAllSupportFrom()
-	return slpb.SupportStatesPerStore{StoreID: sm.storeID, SupportStates: supportStates}
+func (sm *SupportManager) InspectSupportFrom() slpb.InspectSupportFromStatesPerStore {
+	supportFromStates := sm.requesterStateHandler.exportAllSupportFrom()
+	return slpb.InspectSupportFromStatesPerStore{StoreID: sm.storeID, SupportFromStates: supportFromStates}
 }
 
 // InspectSupportFor implements the InspectFabric interface.
-func (sm *SupportManager) InspectSupportFor() slpb.SupportStatesPerStore {
-	supportStates := sm.supporterStateHandler.exportAllSupportFor()
-	return slpb.SupportStatesPerStore{StoreID: sm.storeID, SupportStates: supportStates}
+func (sm *SupportManager) InspectSupportFor() slpb.InspectSupportForStatesPerStore {
+	supportForStates := sm.supporterStateHandler.exportAllSupportFor()
+	return slpb.InspectSupportForStatesPerStore{StoreID: sm.storeID, SupportForStates: supportForStates}
 }
 
 // SupportFrom implements the Fabric interface. It delegates the response to the

pkg/kv/kvserver/storeliveness/supporter_state.go

@@ -17,14 +17,25 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 )
 
+// supportState contains state pertaining to support provided to a single remote
+// store.
+type supportState struct {
+	// SupportState contains all fields that must be persisted to disk.
+	slpb.SupportState
+
+	// lastSupportWithdrawnTime is the timestamp at which support was last
+	// withdrawn for the remote store.
+	lastSupportWithdrawnTime hlc.Timestamp
+}
+
 // supporterState stores the core data structures for providing support.
 type supporterState struct {
 	// meta stores the SupporterMeta, including the max timestamp at which this
 	// store has withdrawn support.
 	meta slpb.SupporterMeta
-	// supportFor stores the SupportState for each remote store for which this
+	// supportFor stores the supportState for each remote store for which this
 	// store has provided support.
-	supportFor map[slpb.StoreIdent]slpb.SupportState
+	supportFor map[slpb.StoreIdent]supportState
 }
 
 // supporterStateHandler is the main interface for handling support for other
@@ -64,15 +75,15 @@ func newSupporterStateHandler() *supporterStateHandler {
 	ssh := &supporterStateHandler{
 		supporterState: supporterState{
 			meta:       slpb.SupporterMeta{},
-			supportFor: make(map[slpb.StoreIdent]slpb.SupportState),
+			supportFor: make(map[slpb.StoreIdent]supportState),
 		},
 	}
 	ssh.update.Store(
 		&supporterStateForUpdate{
 			checkedIn: &ssh.supporterState,
 			inProgress: supporterState{
 				meta:       slpb.SupporterMeta{},
-				supportFor: make(map[slpb.StoreIdent]slpb.SupportState),
+				supportFor: make(map[slpb.StoreIdent]supportState),
 			},
 		},
 	)
@@ -101,7 +112,7 @@ type supporterStateForUpdate struct {
 func (ssh *supporterStateHandler) getSupportFor(id slpb.StoreIdent) slpb.SupportState {
 	ssh.mu.RLock()
 	defer ssh.mu.RUnlock()
-	return ssh.supporterState.supportFor[id]
+	return ssh.supporterState.supportFor[id].SupportState
 }
 
 // getNumSupportFor returns the size of the supporterState.supportFor map.
@@ -111,16 +122,19 @@ func (ssh *supporterStateHandler) getNumSupportFor() int {
 	return len(ssh.supporterState.supportFor)
 }
 
-// exportAllSupportFor exports a copy of all SupportStates from the
+// exportAllSupportFor exports a copy of all InspectSupportForStates from the
 // supporterState.supportFor map.
-func (ssh *supporterStateHandler) exportAllSupportFor() []slpb.SupportState {
+func (ssh *supporterStateHandler) exportAllSupportFor() []slpb.InspectSupportForState {
 	ssh.mu.RLock()
 	defer ssh.mu.RUnlock()
-	supportStates := make([]slpb.SupportState, len(ssh.supporterState.supportFor))
+	supportForStates := make([]slpb.InspectSupportForState, 0, len(ssh.supporterState.supportFor))
 	for _, ss := range ssh.supporterState.supportFor {
-		supportStates = append(supportStates, ss)
+		supportForStates = append(supportForStates, slpb.InspectSupportForState{
+			SupportState:             ss.SupportState,
+			LastSupportWithdrawnTime: ss.lastSupportWithdrawnTime,
+		})
 	}
-	return supportStates
+	return supportForStates
 }
 
 // Functions for handling supporterState updates.
@@ -146,13 +160,11 @@ func (ssfu *supporterStateForUpdate) getMeta() slpb.SupporterMeta {
 	return ssfu.checkedIn.meta
 }
 
-// getSupportFor returns the SupportState from the inProgress view; if not
-// present, it falls back to the SupportState from the checkedIn view.
+// getSupportFor returns the supportState from the inProgress view; if not
+// present, it falls back to the supportState from the checkedIn view.
 // The returned boolean indicates whether the store is present in the supportFor
 // map; it does NOT indicate whether support is provided.
-func (ssfu *supporterStateForUpdate) getSupportFor(
-	storeID slpb.StoreIdent,
-) (slpb.SupportState, bool) {
+func (ssfu *supporterStateForUpdate) getSupportFor(storeID slpb.StoreIdent) (supportState, bool) {
 	ss, ok := ssfu.inProgress.supportFor[storeID]
 	if !ok {
 		ss, ok = ssfu.checkedIn.supportFor[storeID]
@@ -180,7 +192,8 @@ func (ssfu *supporterStateForUpdate) write(ctx context.Context, b storage.Batch)
 		}
 	}
 	for _, ss := range ssfu.inProgress.supportFor {
-		if err := writeSupportForState(ctx, b, ss); err != nil {
+		// Only write the persistent SupportState to disk.
+		if err := writeSupportForState(ctx, b, ss.SupportState); err != nil {
 			return err
 		}
 	}
@@ -201,9 +214,9 @@ func (ssh *supporterStateHandler) read(ctx context.Context, r storage.Reader) er
 	ssh.mu.Lock()
 	defer ssh.mu.Unlock()
 	ssh.supporterState.meta = meta
-	ssh.supporterState.supportFor = make(map[slpb.StoreIdent]slpb.SupportState, len(supportFor))
+	ssh.supporterState.supportFor = make(map[slpb.StoreIdent]supportState, len(supportFor))
 	for _, s := range supportFor {
-		ssh.supporterState.supportFor[s.Target] = s
+		ssh.supporterState.supportFor[s.Target] = supportState{SupportState: s}
 	}
 	return nil
 }
@@ -256,12 +269,15 @@ func (ssfu *supporterStateForUpdate) handleHeartbeat(
 	from := msg.From
 	ss, ok := ssfu.getSupportFor(from)
 	if !ok {
-		ss = slpb.SupportState{Target: from}
+		ss = supportState{SupportState: slpb.SupportState{Target: from}}
 	}
 	ssNew := handleHeartbeat(ss, msg)
+	assert(ss.lastSupportWithdrawnTime == ssNew.lastSupportWithdrawnTime,
+		"lastSupportWithdrawnTime changed on successful heartbeat",
+	)
 	if ss != ssNew {
 		ssfu.inProgress.supportFor[from] = ssNew
-		logSupportForChange(ctx, ss, ssNew)
+		logSupportForChange(ctx, ss.SupportState, ssNew.SupportState)
 	}
 	return slpb.Message{
 		Type:       slpb.MsgHeartbeatResp,
@@ -274,7 +290,7 @@ func (ssfu *supporterStateForUpdate) handleHeartbeat(
 
 // handleHeartbeat contains the core logic for updating the epoch and expiration
 // of a support requester upon receiving a heartbeat.
-func handleHeartbeat(ss slpb.SupportState, msg *slpb.Message) slpb.SupportState {
+func handleHeartbeat(ss supportState, msg *slpb.Message) supportState {
 	assert(!msg.Expiration.IsEmpty(), "requested support with zero expiration")
 	if ss.Epoch == msg.Epoch {
 		ss.Expiration.Forward(msg.Expiration)
@@ -315,10 +331,10 @@ func (ssfu *supporterStateForUpdate) withdrawSupport(
 	)
 	supportWithdrawnForStoreIDs = make(map[roachpb.StoreID]struct{})
 	for id, ss := range ssfu.checkedIn.supportFor {
-		ssNew := maybeWithdrawSupport(ss, now)
-		if ss != ssNew {
+		ssNew, withdrawn := maybeWithdrawSupport(ss, now)
+		if withdrawn {
 			ssfu.inProgress.supportFor[id] = ssNew
-			log.KvExec.Infof(ctx, "withdrew support for %s", supportChangeStr(ss, ssNew))
+			log.KvExec.Infof(ctx, "withdrew support for %s", supportChangeStr(ss.SupportState, ssNew.SupportState))
 			meta := ssfu.getMeta()
 			if meta.MaxWithdrawn.Forward(now) {
 				ssfu.inProgress.meta = meta
@@ -330,11 +346,15 @@ func (ssfu *supporterStateForUpdate) withdrawSupport(
 }
 
 // maybeWithdrawSupport contains the core logic for updating the epoch and
-// expiration of a support requester when withdrawing support.
-func maybeWithdrawSupport(ss slpb.SupportState, now hlc.ClockTimestamp) slpb.SupportState {
+// expiration of a support requester when withdrawing support. If support is
+// withdrawn, lastSupportWithdrawnTime is updated to the current timestamp and
+// a boolean indicating as such is returned.
+func maybeWithdrawSupport(ss supportState, now hlc.ClockTimestamp) (supportState, bool) {
 	if !ss.Expiration.IsEmpty() && ss.Expiration.LessEq(now.ToTimestamp()) {
 		ss.Epoch++
 		ss.Expiration = hlc.Timestamp{}
+		ss.lastSupportWithdrawnTime = now.ToTimestamp()
+		return ss, true
 	}
-	return ss
+	return ss, false
 }

pkg/kv/kvserver/storeliveness/testdata/basic

@@ -49,7 +49,7 @@ debug-supporter-state
 meta:
 {MaxWithdrawn:201.000000000,0}
 support for:
-{Target:{NodeID:2 StoreID:2} Epoch:3 Expiration:0,0}
+{Target:{NodeID:2 StoreID:2} Epoch:3 Expiration:0,0} lastSupportWithdrawnTime:201.000000000,0
 
 debug-metrics
 ----