Merge #153888

153888: kv: move various log lines from Dev to KvExec r=tbg a=arulajmani

See individual commits. 

Informs #152584

`@nicktrav,` `@tbg`  -- these commits migrate all the log lines that need to go into the KvExec channel from the spreadsheet `@kevin-v-ngo` shared with us and then some.

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

Author: craig[bot]

pkg/kv/kvclient/kvcoord/dist_sender.go

@@ -1049,7 +1049,7 @@ func (ds *DistSender) getRoutingInfo(
 			containsFn = (*roachpb.RangeDescriptor).ContainsKeyInverted
 		}
 		if !containsFn(returnToken.Desc(), descKey) {
-			log.Dev.Fatalf(ctx, "programming error: range resolution returning non-matching descriptor: "+
+			log.KvExec.Fatalf(ctx, "programming error: range resolution returning non-matching descriptor: "+
 				"desc: %s, key: %s, reverse: %t", returnToken.Desc(), descKey, redact.Safe(useReverseScan))
 		}
 	}
@@ -1284,7 +1284,7 @@ func (ds *DistSender) Send(
 			if len(parts) != 1 {
 				panic("EndTxn not in last chunk of batch")
 			} else if require1PC {
-				log.Dev.Fatalf(ctx, "required 1PC transaction cannot be split: %s", ba)
+				log.KvExec.Fatalf(ctx, "required 1PC transaction cannot be split: %s", ba)
 			}
 			parts = splitBatchAndCheckForRefreshSpans(ba, true /* split ET */)
 			onePart = false
@@ -2228,7 +2228,7 @@ func (ds *DistSender) sendPartialBatch(
 			{
 				var s redact.StringBuilder
 				slowRangeRPCWarningStr(&s, ba, dur, attempts, routingTok.Desc(), err, reply)
-				log.Dev.Warningf(ctx, "slow range RPC: %v", &s)
+				log.KvExec.Warningf(ctx, "slow range RPC: %v", &s)
 			}
 			// If the RPC wasn't successful, defer the logging of a message once the
 			// RPC is not retried any more.
@@ -2240,7 +2240,7 @@ func (ds *DistSender) sendPartialBatch(
 					ds.metrics.SlowRPCs.Dec(1)
 					var s redact.StringBuilder
 					slowRangeRPCReturnWarningStr(&s, tBegin.Elapsed(), attempts)
-					log.Dev.Warningf(ctx, "slow RPC response: %v", &s)
+					log.KvExec.Warningf(ctx, "slow RPC response: %v", &s)
 				}(tBegin, attempts)
 			}
 			tBegin = 0 // prevent reentering branch for this RPC
@@ -2333,7 +2333,7 @@ func (ds *DistSender) sendPartialBatch(
 	}
 
 	if pErr == nil {
-		log.Dev.Fatal(ctx, "exited retry loop without an error or early exit")
+		log.KvExec.Fatal(ctx, "exited retry loop without an error or early exit")
 	}
 
 	return response{pErr: pErr}
@@ -2563,7 +2563,7 @@ func (ds *DistSender) sendToReplicas(
 	case kvpb.RoutingPolicy_NEAREST:
 		replicaFilter = AllExtantReplicas
 	default:
-		log.Dev.Fatalf(ctx, "unknown routing policy: %s", ba.RoutingPolicy)
+		log.KvExec.Fatalf(ctx, "unknown routing policy: %s", ba.RoutingPolicy)
 	}
 	desc := routing.Desc()
 	replicas, err := NewReplicaSlice(ctx, ds.nodeDescs, desc, routing.Leaseholder(), replicaFilter)
@@ -2603,7 +2603,7 @@ func (ds *DistSender) sendToReplicas(
 		log.VEventf(ctx, 2, "routing to nearest replica; leaseholder not required order=%v", replicas)
 
 	default:
-		log.Dev.Fatalf(ctx, "unknown routing policy: %s", ba.RoutingPolicy)
+		log.KvExec.Fatalf(ctx, "unknown routing policy: %s", ba.RoutingPolicy)
 	}
 
 	// NB: upgrade the connection class to SYSTEM, for critical ranges. Set it to
@@ -2737,7 +2737,7 @@ func (ds *DistSender) sendToReplicas(
 		} else if routing.Leaseholder() == nil {
 			// NB: Normally we don't have both routeToLeaseholder and a nil
 			// leaseholder. This could be changed to an assertion.
-			log.Dev.Errorf(ctx, "attempting %v to route to leaseholder, but the leaseholder is unknown %v", ba, routing)
+			log.KvExec.Errorf(ctx, "attempting %v to route to leaseholder, but the leaseholder is unknown %v", ba, routing)
 		} else if ba.Replica.NodeID == routing.Leaseholder().NodeID {
 			// We are sending this request to the leaseholder, so it doesn't
 			// make sense to attempt to proxy it.
@@ -2780,7 +2780,7 @@ func (ds *DistSender) sendToReplicas(
 				if admissionpb.WorkPriority(ba.AdmissionHeader.Priority) >= admissionpb.NormalPri {
 					// Note that these RPC may or may not have succeeded. Errors are counted separately below.
 					ds.metrics.SlowReplicaRPCs.Inc(1)
-					log.Dev.Warningf(ctx, "slow replica RPC: %v", &s)
+					log.KvExec.Warningf(ctx, "slow replica RPC: %v", &s)
 				} else {
 					log.Eventf(ctx, "slow replica RPC: %v", &s)
 				}

pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go

@@ -66,10 +66,10 @@ func muxRangeFeed(
 	eventCh chan<- RangeFeedMessage,
 ) (retErr error) {
 	if log.V(1) {
-		log.Dev.Infof(ctx, "Establishing MuxRangeFeed (%s...; %d spans)", spans[0], len(spans))
+		log.KvExec.Infof(ctx, "Establishing MuxRangeFeed (%s...; %d spans)", spans[0], len(spans))
 		start := timeutil.Now()
 		defer func() {
-			log.Dev.Infof(ctx, "MuxRangeFeed terminating after %s with err=%v", timeutil.Since(start), retErr)
+			log.KvExec.Infof(ctx, "MuxRangeFeed terminating after %s with err=%v", timeutil.Since(start), retErr)
 		}()
 	}
 
@@ -381,10 +381,10 @@ func (m *rangefeedMuxer) startNodeMuxRangeFeed(
 	defer restore()
 
 	if log.V(1) {
-		log.Dev.Infof(ctx, "Establishing MuxRangeFeed to node %d", nodeID)
+		log.KvExec.Infof(ctx, "Establishing MuxRangeFeed to node %d", nodeID)
 		start := timeutil.Now()
 		defer func() {
-			log.Dev.Infof(ctx, "MuxRangeFeed to node %d terminating after %s with err=%v",
+			log.KvExec.Infof(ctx, "MuxRangeFeed to node %d terminating after %s with err=%v",
 				nodeID, timeutil.Since(start), retErr)
 		}()
 	}
@@ -403,7 +403,7 @@ func (m *rangefeedMuxer) startNodeMuxRangeFeed(
 	maybeCloseClient := func() {
 		if closer, ok := mux.(io.Closer); ok {
 			if err := closer.Close(); err != nil {
-				log.Dev.Warningf(ctx, "error closing mux rangefeed client: %v", err)
+				log.KvExec.Warningf(ctx, "error closing mux rangefeed client: %v", err)
 			}
 		}
 	}
@@ -442,7 +442,7 @@ func (m *rangefeedMuxer) startNodeMuxRangeFeed(
 		}
 
 		if log.V(1) {
-			log.Dev.Infof(ctx, "mux to node %d restarted %d streams", ms.nodeID, len(toRestart))
+			log.KvExec.Infof(ctx, "mux to node %d restarted %d streams", ms.nodeID, len(toRestart))
 		}
 		return m.restartActiveRangeFeeds(ctx, recvErr, toRestart)
 	}
@@ -471,7 +471,7 @@ func (m *rangefeedMuxer) receiveEventsFromNode(
 		// additional event(s) arriving for a stream that is no longer active.
 		if active == nil {
 			if log.V(1) {
-				log.Dev.Infof(ctx, "received stray event stream %d: %v", event.StreamID, event)
+				log.KvExec.Infof(ctx, "received stray event stream %d: %v", event.StreamID, event)
 			}
 			continue
 		}
@@ -566,7 +566,7 @@ func (m *rangefeedMuxer) restartActiveRangeFeed(
 	}
 
 	if log.V(1) {
-		log.Dev.Infof(ctx, "RangeFeed %s@%s (r%d, replica %s) disconnected with last checkpoint %s ago: %v (errInfo %v)",
+		log.KvExec.Infof(ctx, "RangeFeed %s@%s (r%d, replica %s) disconnected with last checkpoint %s ago: %v (errInfo %v)",
 			active.Span, active.StartAfter, active.RangeID, active.ReplicaDescriptor,
 			timeutil.Since(active.Resolved.GoTime()), reason, errInfo)
 	}

pkg/kv/kvclient/kvcoord/txn_coord_sender.go

@@ -239,10 +239,10 @@ func newRootTxnCoordSender(
 	txn.AssertInitialized(context.TODO())
 
 	if txn.Status != roachpb.PENDING {
-		log.Dev.Fatalf(context.TODO(), "unexpected non-pending txn in RootTransactionalSender: %s", txn)
+		log.KvExec.Fatalf(context.TODO(), "unexpected non-pending txn in RootTransactionalSender: %s", txn)
 	}
 	if txn.Sequence != 0 {
-		log.Dev.Fatalf(context.TODO(), "cannot initialize root txn with seq != 0: %s", txn)
+		log.KvExec.Fatalf(context.TODO(), "cannot initialize root txn with seq != 0: %s", txn)
 	}
 
 	tcs := &TxnCoordSender{
@@ -380,7 +380,7 @@ func newLeafTxnCoordSender(
 	txn.AssertInitialized(context.TODO())
 
 	if txn.Status != roachpb.PENDING {
-		log.Dev.Fatalf(context.TODO(), "unexpected non-pending txn in LeafTransactionalSender: %s", tis)
+		log.KvExec.Fatalf(context.TODO(), "unexpected non-pending txn in LeafTransactionalSender: %s", tis)
 	}
 
 	tcs := &TxnCoordSender{
@@ -537,7 +537,7 @@ func (tc *TxnCoordSender) Send(
 
 	// Associate the txnID with the trace.
 	if tc.mu.txn.ID == (uuid.UUID{}) {
-		log.Dev.Fatalf(ctx, "cannot send transactional request through unbound TxnCoordSender")
+		log.KvExec.Fatalf(ctx, "cannot send transactional request through unbound TxnCoordSender")
 	}
 	if sp.IsVerbose() {
 		sp.SetTag("txnID", attribute.StringValue(tc.mu.txn.ID.String()))
@@ -674,7 +674,7 @@ func (tc *TxnCoordSender) Send(
 // docs/RFCS/20200811_non_blocking_txns.md.
 func (tc *TxnCoordSender) maybeCommitWait(ctx context.Context, deferred bool) error {
 	if tc.mu.txn.Status != roachpb.PREPARED && tc.mu.txn.Status != roachpb.COMMITTED {
-		log.Dev.Fatalf(ctx, "maybeCommitWait called when not prepared/committed")
+		log.KvExec.Fatalf(ctx, "maybeCommitWait called when not prepared/committed")
 	}
 	if tc.mu.commitWaitDeferred && !deferred {
 		// If this is an automatic commit-wait call and the user of this
@@ -787,7 +787,7 @@ func (tc *TxnCoordSender) maybeRejectClientLocked(
 			// unexpected for it to find the transaction already in a txnFinalized
 			// state. This may be a bug, so log a stack trace.
 			stack := debugutil.Stack()
-			log.Dev.Errorf(ctx, "%s. stack:\n%s", msg, stack)
+			log.KvExec.Errorf(ctx, "%s. stack:\n%s", msg, stack)
 		}
 		reason := kvpb.TransactionStatusError_REASON_UNKNOWN
 		if tc.mu.txn.Status == roachpb.COMMITTED {
@@ -1028,7 +1028,7 @@ func (tc *TxnCoordSender) updateStateLocked(
 		if errTxnID != txnID {
 			// KV should not return errors for transactions other than the one in
 			// the BatchRequest.
-			log.Dev.Fatalf(ctx, "retryable error for the wrong txn. ba.Txn: %s. pErr: %s",
+			log.KvExec.Fatalf(ctx, "retryable error for the wrong txn. ba.Txn: %s. pErr: %s",
 				ba.Txn, pErr)
 		}
 		return kvpb.NewError(tc.handleRetryableErrLocked(ctx, pErr))
@@ -1095,7 +1095,7 @@ func sanityCheckErrWithTxn(
 			Detail: "you have encountered a known bug in CockroachDB, please consider " +
 				"reporting on the Github issue or reach out via Support.",
 		}))
-	log.Dev.Warningf(ctx, "%v", err)
+	log.KvExec.Warningf(ctx, "%v", err)
 	return err
 }
 
@@ -1479,7 +1479,7 @@ func (tc *TxnCoordSender) UpdateRootWithLeafFinalState(
 	defer tc.mu.Unlock()
 
 	if tc.mu.txn.ID == (uuid.UUID{}) {
-		log.Dev.Fatalf(ctx, "cannot UpdateRootWithLeafFinalState on unbound TxnCoordSender. input id: %s", tfs.Txn.ID)
+		log.KvExec.Fatalf(ctx, "cannot UpdateRootWithLeafFinalState on unbound TxnCoordSender. input id: %s", tfs.Txn.ID)
 	}
 
 	// Sanity check: don't combine if the tfs is for a different txn ID.

pkg/kv/kvclient/kvcoord/txn_interceptor_heartbeater.go

@@ -317,11 +317,11 @@ func (h *txnHeartbeater) closeLocked() {
 // startHeartbeatLoopLocked starts a heartbeat loop in a different goroutine.
 func (h *txnHeartbeater) startHeartbeatLoopLocked(ctx context.Context) {
 	if h.loopInterval < 0 {
-		log.Dev.Infof(ctx, "coordinator heartbeat loop disabled")
+		log.KvExec.Infof(ctx, "coordinator heartbeat loop disabled")
 		return
 	}
 	if h.mu.loopStarted {
-		log.Dev.Fatal(ctx, "attempting to start a second heartbeat loop")
+		log.KvExec.Fatal(ctx, "attempting to start a second heartbeat loop")
 	}
 	log.VEventf(ctx, 2, kvbase.SpawningHeartbeatLoopMsg)
 	h.mu.loopStarted = true
@@ -459,15 +459,15 @@ func (h *txnHeartbeater) heartbeatLocked(ctx context.Context) bool {
 		// client needs to send a rollback.
 		return false
 	case roachpb.COMMITTED:
-		log.Dev.Fatalf(ctx, "txn committed but heartbeat loop hasn't been signaled to stop: %s", h.mu.txn)
+		log.KvExec.Fatalf(ctx, "txn committed but heartbeat loop hasn't been signaled to stop: %s", h.mu.txn)
 	default:
-		log.Dev.Fatalf(ctx, "unexpected txn status in heartbeat loop: %s", h.mu.txn)
+		log.KvExec.Fatalf(ctx, "unexpected txn status in heartbeat loop: %s", h.mu.txn)
 	}
 
 	// Clone the txn in order to put it in the heartbeat request.
 	txn := h.mu.txn.Clone()
 	if txn.Key == nil {
-		log.Dev.Fatalf(ctx, "attempting to heartbeat txn without anchor key: %v", txn)
+		log.KvExec.Fatalf(ctx, "attempting to heartbeat txn without anchor key: %v", txn)
 	}
 	ba := &kvpb.BatchRequest{}
 	ba.Txn = txn
@@ -632,7 +632,7 @@ func (h *txnHeartbeater) abortTxnAsyncLocked(ctx context.Context) {
 		TaskName: taskName,
 	})
 	if err != nil {
-		log.Dev.Warningf(ctx, "%v", err)
+		log.KvExec.Warningf(ctx, "%v", err)
 		h.metrics.AsyncRollbacksFailed.Inc(1)
 		return
 	}

pkg/kv/kvclient/kvcoord/txn_interceptor_pipeliner.go

@@ -490,10 +490,10 @@ func (tp *txnPipeliner) attachLocksToEndTxn(
 
 	if log.V(3) {
 		for _, intent := range et.LockSpans {
-			log.Dev.Infof(ctx, "intent: [%s,%s)", intent.Key, intent.EndKey)
+			log.KvExec.Infof(ctx, "intent: [%s,%s)", intent.Key, intent.EndKey)
 		}
 		for _, write := range et.InFlightWrites {
-			log.Dev.Infof(ctx, "in-flight: %d:%s (%s)", write.Sequence, write.Key, write.Strength)
+			log.KvExec.Infof(ctx, "in-flight: %d:%s (%s)", write.Sequence, write.Key, write.Strength)
 		}
 	}
 	return ba, nil
@@ -736,7 +736,7 @@ func (tp *txnPipeliner) updateLockTracking(
 	// fine for now, but we add some observability to be aware of this happening.
 	if tp.ifWrites.byteSize() > maxBytes {
 		if tp.inflightOverBudgetEveryN.ShouldLog() || log.ExpensiveLogEnabled(ctx, 2) {
-			log.Dev.Warningf(ctx, "a transaction's in-flight writes and locking reads have "+
+			log.KvExec.Warningf(ctx, "a transaction's in-flight writes and locking reads have "+
 				"exceeded the intent tracking limit (kv.transaction.max_intents_bytes). "+
 				"in-flight writes and locking reads size: %d bytes, txn: %s, ba: %s",
 				tp.ifWrites.byteSize(), ba.Txn, ba.Summary())
@@ -748,7 +748,7 @@ func (tp *txnPipeliner) updateLockTracking(
 	// number of ranged locking reads before sending the request.
 	if rejectTxnMaxCount > 0 && tp.writeCount > rejectTxnMaxCount {
 		if tp.inflightOverBudgetEveryN.ShouldLog() || log.ExpensiveLogEnabled(ctx, 2) {
-			log.Dev.Warningf(ctx, "a transaction has exceeded the maximum number of writes "+
+			log.KvExec.Warningf(ctx, "a transaction has exceeded the maximum number of writes "+
 				"allowed by kv.transaction.max_intents_and_locks: "+
 				"count: %d, txn: %s, ba: %s", tp.writeCount, ba.Txn, ba.Summary())
 		}
@@ -782,7 +782,7 @@ func (tp *txnPipeliner) updateLockTracking(
 	condensed := tp.lockFootprint.maybeCondense(ctx, tp.riGen, locksBudget)
 	if condensed && !alreadyCondensed {
 		if tp.condensedIntentsEveryN.ShouldLog() || log.ExpensiveLogEnabled(ctx, 2) {
-			log.Dev.Warningf(ctx,
+			log.KvExec.Warningf(ctx,
 				"a transaction has hit the intent tracking limit (kv.transaction.max_intents_bytes); "+
 					"is it a bulk operation? Intent cleanup will be slower. txn: %s ba: %s",
 				ba.Txn, ba.Summary())
@@ -878,7 +878,7 @@ func (tp *txnPipeliner) updateLockTrackingInner(
 					// Record any writes that were performed asynchronously. We'll
 					// need to prove that these succeeded sometime before we commit.
 					if span.EndKey != nil {
-						log.Dev.Fatalf(ctx, "unexpected multi-key intent pipelined")
+						log.KvExec.Fatalf(ctx, "unexpected multi-key intent pipelined")
 					}
 					tp.ifWrites.insert(span.Key, seq, str)
 				} else {

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -601,7 +601,7 @@ func (twb *txnWriteBuffer) adjustError(
 				// For requests that were not transformed, attributing an error to them
 				// shouldn't confuse the client.
 				if baIdx == pErr.Index.Index && record.transformed {
-					log.Dev.Warningf(ctx, "error index %d is part of a transformed request", pErr.Index.Index)
+					log.KvExec.Warningf(ctx, "error index %d is part of a transformed request", pErr.Index.Index)
 					pErr.Index = nil
 					return pErr
 				} else if baIdx == pErr.Index.Index {
@@ -626,7 +626,7 @@ func (twb *txnWriteBuffer) adjustErrorUponFlush(
 		if pErr.Index.Index < int32(numBuffered) {
 			// If the error belongs to a request because part of the buffer flush, nil
 			// out the index.
-			log.Dev.Warningf(ctx, "error index %d is part of the buffer flush", pErr.Index.Index)
+			log.KvExec.Warningf(ctx, "error index %d is part of the buffer flush", pErr.Index.Index)
 			pErr.Index = nil
 		} else {
 			// Otherwise, adjust the error index to hide the impact of any flushed
@@ -1313,7 +1313,7 @@ func (twb *txnWriteBuffer) mergeResponseWithRequestRecords(
 	ctx context.Context, rr requestRecords, br *kvpb.BatchResponse,
 ) (_ *kvpb.BatchResponse, pErr *kvpb.Error) {
 	if rr.Empty() && br == nil {
-		log.Dev.Fatal(ctx, "unexpectedly found no transformations and no batch response")
+		log.KvExec.Fatal(ctx, "unexpectedly found no transformations and no batch response")
 	} else if rr.Empty() {
 		return br, nil
 	}
@@ -1325,7 +1325,7 @@ func (twb *txnWriteBuffer) mergeResponseWithRequestRecords(
 		brResp := kvpb.ResponseUnion{}
 		if !record.stripped {
 			if len(br.Responses) == 0 {
-				log.Dev.Fatal(ctx, "unexpectedly found a non-stripped request and no batch response")
+				log.KvExec.Fatal(ctx, "unexpectedly found a non-stripped request and no batch response")
 			}
 			// If the request wasn't stripped from the batch we sent to KV, we
 			// received a response for it, which then needs to be combined with

pkg/kv/kvserver/batcheval/cmd_add_sstable.go

@@ -475,7 +475,7 @@ func computeSSTStatsDiffWithFallback(
 	stats, err := storage.ComputeSSTStatsDiff(
 		ctx, sst, readWriter, nowNanos, start, end)
 	if errors.IsAny(err, storage.ComputeSSTStatsDiffReaderHasRangeKeys, storage.ComputeStatsDiffViolation) {
-		log.Dev.Warningf(ctx, "computing SST stats as estimates because of ComputeSSTStatsDiff error: %s", err)
+		log.KvExec.Warningf(ctx, "computing SST stats as estimates because of ComputeSSTStatsDiff error: %s", err)
 		sstStats, err := computeSSTStats(ctx, sst, nowNanos)
 		if err != nil {
 			return enginepb.MVCCStats{}, errors.Wrap(err, "error computing SST stats during fallback")

pkg/kv/kvserver/batcheval/cmd_clear_range.go

@@ -201,7 +201,7 @@ func computeStatsDelta(
 			// We only want to assert the correctness of stats that do not contain
 			// estimates.
 			if delta.ContainsEstimates == 0 && !delta.Equal(computed) {
-				log.Dev.Fatalf(ctx, "fast-path MVCCStats computation gave wrong result: diff(fast, computed) = %s",
+				log.KvExec.Fatalf(ctx, "fast-path MVCCStats computation gave wrong result: diff(fast, computed) = %s",
 					pretty.Diff(delta, computed))
 			}
 		}

pkg/kv/kvserver/batcheval/cmd_end_transaction.go

@@ -833,7 +833,7 @@ func updateFinalizedTxn(
 	opts := storage.MVCCWriteOptions{Stats: ms, Category: fs.BatchEvalReadCategory}
 	if !evalCtx.EvalKnobs().DisableTxnAutoGC && len(externalLocks) == 0 {
 		if log.V(2) {
-			log.Dev.Infof(ctx, "auto-gc'ed %s (%d locks)", txn.Short(), len(args.LockSpans))
+			log.KvExec.Infof(ctx, "auto-gc'ed %s (%d locks)", txn.Short(), len(args.LockSpans))
 		}
 		if !recordAlreadyExisted {
 			// Nothing to delete, so there's no use writing a deletion tombstone. This
@@ -956,7 +956,7 @@ func RunCommitTrigger(
 		return res, nil
 	}
 
-	log.Dev.Fatalf(ctx, "unknown commit trigger: %+v", ct)
+	log.KvExec.Fatalf(ctx, "unknown commit trigger: %+v", ct)
 	return result.Result{}, nil
 }
 
@@ -1400,7 +1400,7 @@ func splitTriggerHelper(
 			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to load lease")
 		}
 		if leftLease.Empty() {
-			log.Dev.Fatalf(ctx, "LHS of split has no lease")
+			log.KvExec.Fatalf(ctx, "LHS of split has no lease")
 		}
 
 		// Copy the lease from the left-hand side of the split over to the