kvserver/closedts: move from log.KvDistribution to log.KvExec

Author: wenyihu6

pkg/kv/kvserver/closedts/policyrefresher/policy_refresher.go

@@ -69,7 +69,7 @@ func NewPolicyRefresher(
 	knobs *TestingKnobs,
 ) *PolicyRefresher {
 	if getLeaseholderReplicas == nil || getNodeLatencies == nil {
-		log.KvDistribution.Fatalf(context.Background(), "getLeaseholderReplicas and getNodeLatencies must be non-nil")
+		log.KvExec.Fatalf(context.Background(), "getLeaseholderReplicas and getNodeLatencies must be non-nil")
 		return nil
 	}
 	refresher := &PolicyRefresher{

pkg/kv/kvserver/closedts/sidetransport/receiver.go

@@ -147,7 +147,7 @@ func (s *Receiver) onRecvErr(ctx context.Context, nodeID roachpb.NodeID, err err
 	defer s.mu.Unlock()
 
 	if err != io.EOF {
-		log.KvDistribution.Warningf(ctx, "closed timestamps side-transport connection dropped from node: %d (%s)", nodeID, err)
+		log.KvExec.Warningf(ctx, "closed timestamps side-transport connection dropped from node: %d (%s)", nodeID, err)
 	} else {
 		log.VEventf(ctx, 2, "closed timestamps side-transport connection dropped from node: %d (%s)", nodeID, err)
 	}
@@ -236,11 +236,11 @@ func (r *incomingStream) processUpdate(ctx context.Context, msg *ctpb.Update) {
 	log.VEventf(ctx, 4, "received side-transport update: %v", msg)
 
 	if msg.NodeID == 0 {
-		log.KvDistribution.Fatalf(ctx, "missing NodeID in message: %s", msg)
+		log.KvExec.Fatalf(ctx, "missing NodeID in message: %s", msg)
 	}
 
 	if msg.NodeID != r.nodeID {
-		log.KvDistribution.Fatalf(ctx, "wrong NodeID; expected %d, got %d", r.nodeID, msg.NodeID)
+		log.KvExec.Fatalf(ctx, "wrong NodeID; expected %d, got %d", r.nodeID, msg.NodeID)
 	}
 
 	// Handle the removed ranges. In order to not lose closed ts info, before we
@@ -259,11 +259,11 @@ func (r *incomingStream) processUpdate(ctx context.Context, msg *ctpb.Update) {
 		for _, rangeID := range msg.Removed {
 			info, ok := r.mu.tracked[rangeID]
 			if !ok {
-				log.KvDistribution.Fatalf(ctx, "attempting to unregister a missing range: r%d", rangeID)
+				log.KvExec.Fatalf(ctx, "attempting to unregister a missing range: r%d", rangeID)
 			}
 			ts, ok := r.mu.lastClosed[info.policy]
 			if !ok {
-				log.KvDistribution.Fatalf(ctx, "missing closed timestamp policy %v for range r%d", info.policy, rangeID)
+				log.KvExec.Fatalf(ctx, "missing closed timestamp policy %v for range r%d", info.policy, rangeID)
 			}
 			r.stores.ForwardSideTransportClosedTimestampForRange(ctx, rangeID, ts, info.lai)
 		}
@@ -279,7 +279,7 @@ func (r *incomingStream) processUpdate(ctx context.Context, msg *ctpb.Update) {
 		r.mu.lastClosed = make(map[ctpb.RangeClosedTimestampPolicy]hlc.Timestamp, len(r.mu.lastClosed))
 		r.mu.tracked = make(map[roachpb.RangeID]trackedRange, len(r.mu.tracked))
 	} else if msg.SeqNum != r.mu.lastSeqNum+1 {
-		log.KvDistribution.Fatalf(ctx, "expected closed timestamp side-transport message with sequence number "+
+		log.KvExec.Fatalf(ctx, "expected closed timestamp side-transport message with sequence number "+
 			"%d, got %d", r.mu.lastSeqNum+1, msg.SeqNum)
 	}
 	r.mu.lastSeqNum = msg.SeqNum
@@ -328,13 +328,13 @@ func (r *incomingStream) Run(
 				r.nodeID = msg.NodeID
 
 				if err := r.server.onFirstMsg(ctx, r, r.nodeID); err != nil {
-					log.KvDistribution.Warningf(ctx, "%s", err.Error())
+					log.KvExec.Warningf(ctx, "%s", err.Error())
 					return
 				} else if ch := r.testingKnobs.onFirstMsg; ch != nil {
 					ch <- struct{}{}
 				}
 				if !msg.Snapshot {
-					log.KvDistribution.Fatal(ctx, "expected the first message to be a snapshot")
+					log.KvExec.Fatal(ctx, "expected the first message to be a snapshot")
 				}
 				r.AddLogTag("remote", r.nodeID)
 				ctx = r.AnnotateCtx(ctx)

pkg/kv/kvserver/closedts/sidetransport/sender.go

@@ -653,7 +653,7 @@ func (b *updatesBuf) Push(ctx context.Context, update *ctpb.Update) {
 	if b.sizeLocked() != 0 {
 		lastIdx := b.lastIdxLocked()
 		if prevSeq := b.mu.data[lastIdx].SeqNum; prevSeq != update.SeqNum-1 {
-			log.KvDistribution.Fatalf(ctx, "bad sequence number; expected %d, got %d", prevSeq+1, update.SeqNum)
+			log.KvExec.Fatalf(ctx, "bad sequence number; expected %d, got %d", prevSeq+1, update.SeqNum)
 		}
 	}
 
@@ -762,7 +762,7 @@ func (b *updatesBuf) GetBySeq(ctx context.Context, seqNum ctpb.SeqNum) (*ctpb.Up
 			continue
 		}
 		if seqNum > lastSeq+1 {
-			log.KvDistribution.Fatalf(ctx, "skipping sequence numbers; requested: %d, last: %d", seqNum, lastSeq)
+			log.KvExec.Fatalf(ctx, "skipping sequence numbers; requested: %d, last: %d", seqNum, lastSeq)
 		}
 		idx := (b.mu.head + (int)(seqNum-firstSeq)) % len(b.mu.data)
 		return b.mu.data[idx], true
@@ -963,7 +963,7 @@ func (r *rpcConn) run(ctx context.Context, stopper *stop.Stopper) {
 				}
 				if err := r.maybeConnect(ctx, stopper); err != nil {
 					if !errors.HasType(err, (*netutil.InitialHeartbeatFailedError)(nil)) && everyN.ShouldLog() {
-						log.KvDistribution.Infof(ctx, "side-transport failed to connect to n%d: %s", r.nodeID, err)
+						log.KvExec.Infof(ctx, "side-transport failed to connect to n%d: %s", r.nodeID, err)
 					}
 					time.Sleep(errSleepTime)
 					continue
@@ -993,7 +993,7 @@ func (r *rpcConn) run(ctx context.Context, stopper *stop.Stopper) {
 				}
 				if err := r.stream.Send(msg); err != nil {
 					if err != io.EOF && everyN.ShouldLog() {
-						log.KvDistribution.Warningf(ctx, "failed to send closed timestamp message %d to n%d: %s",
+						log.KvExec.Warningf(ctx, "failed to send closed timestamp message %d to n%d: %s",
 							r.lastSent, r.nodeID, err)
 					}
 					// Keep track of the fact that we need a new connection.

pkg/kv/kvserver/closedts/tracker/heap_tracker.go

@@ -103,7 +103,7 @@ func (h *heapTracker) Track(ctx context.Context, ts hlc.Timestamp) RemovalToken
 func (h *heapTracker) Untrack(ctx context.Context, tok RemovalToken) {
 	idx := tok.(heapToken).index
 	if idx == -1 {
-		log.KvDistribution.Fatalf(ctx, "attempting to untrack already-untracked item")
+		log.KvExec.Fatalf(ctx, "attempting to untrack already-untracked item")
 	}
 	h.mu.Lock()
 	defer h.mu.Unlock()

pkg/kv/kvserver/closedts/tracker/lockfree_tracker.go

@@ -168,7 +168,7 @@ func (t *lockfreeTracker) Untrack(ctx context.Context, tok RemovalToken) {
 	// types.
 	refcnt := b.refcnt.Add(-1)
 	if refcnt < 0 {
-		log.KvDistribution.Fatalf(ctx, "negative bucket refcount: %d", refcnt)
+		log.KvExec.Fatalf(ctx, "negative bucket refcount: %d", refcnt)
 	}
 	if refcnt == 0 {
 		// Reset the bucket, so that future Track() calls can create a new one.

pkg/kv/kvserver/closedts/tracker/tracker_test.go

@@ -167,7 +167,7 @@ func TestLockfreeTrackerRandomStress(t *testing.T) {
 	// hours of stressrace on GCE worker (and it failed once on CI stress too).
 	// Figure out why.
 	maxToleratedErrorMillis := 3 * c.maxEvaluationTime.Milliseconds()
-	log.KvDistribution.Infof(ctx, "maximum lower bound error: %dms. maximum request evaluation time: %s",
+	log.KvExec.Infof(ctx, "maximum lower bound error: %dms. maximum request evaluation time: %s",
 		maxOvershotMillis, c.maxEvaluationTime)
 	require.Lessf(t, maxOvershotMillis, maxToleratedErrorMillis,
 		"maximum tracker lowerbound error was %dms, above maximum tolerated %dms",
@@ -446,7 +446,7 @@ func (c *trackerChecker) run(ctx context.Context) error {
 		if c.maxOvershotNanos < overshotNanos {
 			c.maxOvershotNanos = overshotNanos
 		}
-		log.KvDistribution.VInfof(ctx, 1, "lower bound error: %dms", overshotNanos/1000000)
+		log.KvExec.VInfof(ctx, 1, "lower bound error: %dms", overshotNanos/1000000)
 	}
 }
 
@@ -517,7 +517,7 @@ func benchmarkTracker(ctx context.Context, b *testing.B, t Tracker) {
 				toks[i] = toks[i][:0]
 			}
 			mu.Unlock()
-			log.KvDistribution.VInfof(ctx, 1, "cleared %d reqs", n)
+			log.KvExec.VInfof(ctx, 1, "cleared %d reqs", n)
 		}
 	}()
 

pkg/kv/kvserver/replica_closedts.go

@@ -293,7 +293,7 @@ func (st *sidetransportAccess) forward(
 
 func (st *sidetransportAccess) assertNoRegression(ctx context.Context, cur, up closedTimestamp) {
 	if cur.regression(up) {
-		log.KvDistribution.Fatalf(ctx, "side-transport update saw closed timestamp regression on r%d: "+
+		log.KvExec.Fatalf(ctx, "side-transport update saw closed timestamp regression on r%d: "+
 			"(lai=%d, ts=%s) -> (lai=%d, ts=%s)", st.rangeID, cur.lai, cur.ts, up.lai, up.ts)
 	}
 }