Merge #155795 #155798 #155850 #155853

155795: workload/schemachange: remove FK violation prediction for INSERT operations r=rafiss a=rafiss


Previously, the random schema workload attempted to predict whether INSERT operations would violate foreign key constraints by analyzing the database state and constraint definitions. This prediction logic involved:
- Querying the database to check if FK violations would occur
- Checking for ongoing FK constraint mutations
- Distinguishing between expected and potential FK violations

This commit removes all FK violation prediction logic and makes ForeignKeyViolation unconditionally a potential execution and commit error for INSERT operations. This aligns with the existing pattern used for adding FK constraints, where FK violations are always treated as potential errors due to the asynchronous validation job.

This change simplifies the code by removing ~200 lines of complex prediction logic and deflakes tests that depend on accurate error handling during concurrent schema changes.

Resolves: #155686
Resolves: #154715 
Resolves: #152402

Epic: None

Release note: None

🤖 Generated with [Claude Code](https://claude.com/claude-code)

155798: kvnemesis: log a metrics report r=stevendanna a=miraradeva

This commit adds logging of some key metrics at the end of each kvnemesis run.

Fixes: #153793

Release note: None

----

This is what the report looks like for a run of `TestKVNemesisMultiNode`:
```
Metric                              | Node 1               | Node 2               | Node 3               | Node 4
------------------------------------+----------------------+----------------------+----------------------+---------------------
raft.commands.proposed              | 2906                 | 117                  | 273                  | 118
raft.commands.reproposed.new-lai    | 0                    | 1                    | 12                   | 0
raft.commands.reproposed.unchanged  | 78                   | 0                    | 0                    | 0
txn.server_side.1PC.success         | 84                   | 0                    | 1                    | 0
txnrecovery.successes.committed     | 0                    | 0                    | 0                    | 0
txnwaitqueue.deadlocks_total        | 0                    | 0                    | 4                    | 0
txn.aborts                          | 13                   | 10                   | 8                    | 12
txn.durations                       | μ=25ms p99=260ms     | μ=82ms p99=671ms     | μ=81ms p99=1.476s    | μ=224ms p99=4.832s
txn.restarts.writetooold            | 8                    | 1                    | 2                    | 1
txn.restarts.serializable           | 1                    | 0                    | 0                    | 0
txn.restarts.readwithinuncertainty  | 0                    | 0                    | 0                    | 0
```

155850: kvserver/rangefeed: move from log.KvDistribution to log.KvExec r=stevendanna a=wenyihu6

Epic: none 
Release note: none 

155853: roachprod: fix label propagation on workload machines r=DarrylWong,golgeek a=herkolategan

Previously, label propagation did not take the boot-disk-only parameter of workload machines into account. This change fixes the propagation to skip trying to add labels to non-existent persistent disks if the boot-disk-only flag has been passed to the provider.

Epic: None
Release note: None

Co-authored-by: Rafi Shamim <[email protected]>
Co-authored-by: Mira Radeva <[email protected]>
Co-authored-by: wenyihu6 <[email protected]>
Co-authored-by: Herko Lategan <[email protected]>

Author: 5 people

pkg/kv/kvnemesis/BUILD.bazel

@@ -113,6 +113,7 @@ go_test(
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
+        "//pkg/util/metric",
         "//pkg/util/randutil",
         "//pkg/util/stop",
         "//pkg/util/syncutil",

pkg/kv/kvnemesis/kvnemesis_test.go

@@ -13,6 +13,7 @@ import (
 	"os"
 	"path"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -46,6 +47,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
@@ -698,19 +700,91 @@ func testKVNemesisImpl(t testing.TB, cfg kvnemesisTestCfg) {
 	env := &Env{SQLDBs: sqlDBs, Tracker: tr, L: logger, Partitioner: &partitioner, Restarter: tc}
 	failures, err := RunNemesis(ctx, rng, env, config, cfg.concurrency, cfg.numSteps, cfg.mode, dbs...)
 
-	for i := 0; i < cfg.numNodes; i++ {
-		t.Logf("[%d] proposed: %d", i,
-			tc.GetFirstStoreFromServer(t, i).Metrics().RaftCommandsProposed.Count())
-		t.Logf("[%d] reproposed unchanged: %d", i,
-			tc.GetFirstStoreFromServer(t, i).Metrics().RaftCommandsReproposed.Count())
-		t.Logf("[%d] reproposed with new LAI: %d", i,
-			tc.GetFirstStoreFromServer(t, i).Metrics().RaftCommandsReproposedLAI.Count())
-	}
+	logMetricsReport(t, tc)
 
 	require.NoError(t, err, `%+v`, err)
 	require.Zero(t, len(failures), "kvnemesis detected failures") // they've been logged already
 }
 
+func logMetricsReport(t testing.TB, tc *testcluster.TestCluster) {
+	metricsOfInterest := []string{
+		// Raft command metrics
+		"raft.commands.proposed",
+		"raft.commands.reproposed.new-lai",
+		"raft.commands.reproposed.unchanged",
+		// Transaction metrics
+		"txn.aborts",
+		"txn.durations",
+		"txn.restarts.writetooold",
+		"txn.restarts.serializable",
+		"txn.restarts.readwithinuncertainty",
+		"txn.server_side.1PC.success",
+		"txnrecovery.failures",
+		"txnrecovery.successes.aborted",
+		"txnrecovery.successes.committed",
+		"txnwaitqueue.deadlocks_total",
+	}
+
+	numNodes := tc.NumServers()
+	nodeMetrics := make([]map[string]string, numNodes)
+	metricsMap := make(map[string]struct{}, len(metricsOfInterest))
+	for _, m := range metricsOfInterest {
+		metricsMap[m] = struct{}{}
+	}
+	for i := 0; i < numNodes; i++ {
+		nodeMetrics[i] = make(map[string]string)
+		processMetric := func(name string, v interface{}) {
+			switch val := v.(type) {
+			case *metric.Counter:
+				nodeMetrics[i][name] = fmt.Sprintf("%d", val.Count())
+			case *metric.Gauge:
+				nodeMetrics[i][name] = fmt.Sprintf("%d", val.Value())
+			case metric.IHistogram:
+				// TODO(mira): Currently, histograms are assumed to store values of type
+				// time duration.
+				snapshot := val.CumulativeSnapshot()
+				makePretty := func(nanos float64) time.Duration {
+					return time.Duration(nanos).Round(time.Millisecond)
+				}
+				meanDur := makePretty(snapshot.Mean())
+				p50Dur := makePretty(snapshot.ValueAtQuantile(50))
+				p99Dur := makePretty(snapshot.ValueAtQuantile(99))
+				nodeMetrics[i][name] = fmt.Sprintf("μ=%v p50=%v p99=%v", meanDur, p50Dur, p99Dur)
+			default:
+				nodeMetrics[i][name] = fmt.Sprintf("unknown:%T", v)
+			}
+		}
+
+		clientRegistry := tc.Server(i).StorageLayer().MetricsRecorder().AppRegistry()
+		nodeRegistry := tc.Server(i).StorageLayer().MetricsRecorder().NodeRegistry()
+		storeID := tc.GetFirstStoreFromServer(t, i).StoreID()
+		storeRegistry := tc.Server(i).StorageLayer().MetricsRecorder().StoreRegistry(storeID)
+		clientRegistry.Select(metricsMap, func(name string, v interface{}) { processMetric(name, v) })
+		nodeRegistry.Select(metricsMap, func(name string, v interface{}) { processMetric(name, v) })
+		storeRegistry.Select(metricsMap, func(name string, v interface{}) { processMetric(name, v) })
+	}
+
+	header := fmt.Sprintf("%-35s", "Metric")
+	for i := 0; i < numNodes; i++ {
+		header += fmt.Sprintf(" | %-30s", fmt.Sprintf("Node %d", i+1))
+	}
+	t.Log(header)
+
+	separator := strings.Repeat("-", 35)
+	for i := 0; i < numNodes; i++ {
+		separator += "-+-" + strings.Repeat("-", 30)
+	}
+	t.Log(separator)
+
+	for _, metricName := range metricsOfInterest {
+		row := fmt.Sprintf("%-35s", metricName)
+		for i := 0; i < numNodes; i++ {
+			row += fmt.Sprintf(" | %-30s", nodeMetrics[i][metricName])
+		}
+		t.Log(row)
+	}
+}
+
 // TestRunReproductionSteps is a helper that allows quickly running a kvnemesis
 // history.
 func TestRunReproductionSteps(t *testing.T) {

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)
 		}
 	}
 
@@ -764,7 +764,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
@@ -965,7 +965,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
@@ -995,7 +995,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/rangefeed/budget.go

@@ -248,7 +248,7 @@ type SharedBudgetAllocation struct {
 func (a *SharedBudgetAllocation) Use(ctx context.Context) {
 	if a != nil {
 		if atomic.AddInt32(&a.refCount, 1) == 1 {
-			log.KvDistribution.Fatalf(ctx, "unexpected shared memory allocation usage increase after free")
+			log.KvExec.Fatalf(ctx, "unexpected shared memory allocation usage increase after free")
 		}
 	}
 }

pkg/kv/kvserver/rangefeed/buffered_registration.go

@@ -196,7 +196,7 @@ func (br *bufferedRegistration) outputLoop(ctx context.Context) error {
 	// If the registration has a catch-up scan, run it.
 	if err := br.maybeRunCatchUpScan(ctx); err != nil {
 		err = errors.Wrap(err, "catch-up scan failed")
-		log.KvDistribution.Errorf(ctx, "%v", err)
+		log.KvExec.Errorf(ctx, "%v", err)
 		return err
 	}
 
@@ -229,7 +229,7 @@ func (br *bufferedRegistration) outputLoop(ctx context.Context) error {
 
 		if overflowed {
 			if wasOverflowedOnFirstIteration && br.shouldLogOverflow(oneCheckpointWithTimestampSent) {
-				log.KvDistribution.Warningf(ctx, "rangefeed %s overflowed during catch up scan from %s (useful checkpoint sent: %v)",
+				log.KvExec.Warningf(ctx, "rangefeed %s overflowed during catch up scan from %s (useful checkpoint sent: %v)",
 					br.span, br.catchUpTimestamp, oneCheckpointWithTimestampSent)
 			}