kvcoord: setting to reject txns above lock span limit

This patch introduces kv.transaction.reject_over_max_intents_budget. If
set, this changes our behavior when a txn exceeds its locks+in-flight
write budget (kv.transaction.max_intents_bytes): instead of compacting
some of its lock spans with precision loss, the request causing the
budget to be exceeded will be rejected instead.

The idea is that we've seen transactions that exceed this budget be very
expensive to clean up - they have to scan a lot to find their intents,
and these cleanups take wide latches. So now one has the option to
reject these transactions, instead of risking this performance cliff.

Each request is checked against the budget by the pipeliner before being
sent out for evaluation. This check is not precise, since the exact
effects of the request on the memory budget are only known at response
time because of ResumeSpans, effects of QueryIntents, etc. So, the check
is best-effort. If a slips through and then the response overflows the
budget, we keep the locks non-condensed; if a further request in the txn
tries to lock more, it'll be rejected. A commit/rollback is
always allowed to pass through, since it doesn't lock anything by
itself.

Fixes cockroachdb#66742

Release note (general change): A new cluster setting
(kv.transaction.reject_over_max_intents_budget) affords control over the
behavior when a transaction exceeds its "locks-tracking memory budget"
(dictated by kv.transaction.max_intents_bytes). Instead of allowing such
transaction to continue with imprecise tracking of their locks, setting
this new option rejects the query that would push its transaction over
this budget with an error (error code 53400 - "configuration limit
exceeded). Transactions that don't track their locks precisely are
potentially destabilizing for the cluster since cleaning them up can
take considerable resources. Transactions that change many rows have the
potential to run into this memory budget issue.

docs/generated/settings/settings-for-tenants.txt

@@ -36,6 +36,7 @@ kv.rangefeed.enabled	boolean	false	if set, rangefeed registration is enabled
 kv.replication_reports.interval	duration	1m0s	the frequency for generating the replication_constraint_stats, replication_stats_report and replication_critical_localities reports (set to 0 to disable)
 kv.transaction.max_intents_bytes	integer	4194304	maximum number of bytes used to track locks in transactions
 kv.transaction.max_refresh_spans_bytes	integer	256000	maximum number of bytes used to track refresh spans in serializable transactions
+kv.transaction.reject_over_max_intents_budget.enabled	boolean	false	if set, transactions that exceed their lock tracking budget (kv.transaction.max_intents_bytes) are rejected instead of having their lock spans imprecisely compressed
 security.ocsp.mode	enumeration	off	"use OCSP to check whether TLS certificates are revoked. If the OCSP
 server is unreachable, in strict mode all certificates will be rejected
 and in lax mode all certificates will be accepted. [off = 0, lax = 1, strict = 2]"

docs/generated/settings/settings.html

@@ -40,6 +40,7 @@
 <tr><td><code>kv.snapshot_recovery.max_rate</code></td><td>byte size</td><td><code>8.0 MiB</code></td><td>the rate limit (bytes/sec) to use for recovery snapshots</td></tr>
 <tr><td><code>kv.transaction.max_intents_bytes</code></td><td>integer</td><td><code>4194304</code></td><td>maximum number of bytes used to track locks in transactions</td></tr>
 <tr><td><code>kv.transaction.max_refresh_spans_bytes</code></td><td>integer</td><td><code>256000</code></td><td>maximum number of bytes used to track refresh spans in serializable transactions</td></tr>
+<tr><td><code>kv.transaction.reject_over_max_intents_budget.enabled</code></td><td>boolean</td><td><code>false</code></td><td>if set, transactions that exceed their lock tracking budget (kv.transaction.max_intents_bytes) are rejected instead of having their lock spans imprecisely compressed</td></tr>
 <tr><td><code>security.ocsp.mode</code></td><td>enumeration</td><td><code>off</code></td><td>use OCSP to check whether TLS certificates are revoked. If the OCSP<br/>server is unreachable, in strict mode all certificates will be rejected<br/>and in lax mode all certificates will be accepted. [off = 0, lax = 1, strict = 2]</td></tr>
 <tr><td><code>security.ocsp.timeout</code></td><td>duration</td><td><code>3s</code></td><td>timeout before considering the OCSP server unreachable</td></tr>
 <tr><td><code>server.auth_log.sql_connections.enabled</code></td><td>boolean</td><td><code>false</code></td><td>if set, log SQL client connect and disconnect events (note: may hinder performance on loaded nodes)</td></tr>

pkg/kv/kvclient/kvcoord/BUILD.bazel

@@ -46,6 +46,8 @@ go_library(
         "//pkg/server/telemetry",
         "//pkg/settings",
         "//pkg/settings/cluster",
+        "//pkg/sql/pgwire/pgcode",
+        "//pkg/sql/pgwire/pgerror",
         "//pkg/storage/enginepb",
         "//pkg/util",
         "//pkg/util/contextutil",
@@ -64,8 +66,10 @@ go_library(
         "//pkg/util/tracing",
         "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_errors//errorspb",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_redact//:redact",
+        "@com_github_gogo_protobuf//proto",
         "@com_github_google_btree//:btree",
     ],
 )
@@ -148,6 +152,8 @@ go_test(
         "//pkg/security/securitytest",
         "//pkg/server",
         "//pkg/settings/cluster",
+        "//pkg/sql/pgwire/pgcode",
+        "//pkg/sql/pgwire/pgerror",
         "//pkg/storage",
         "//pkg/storage/enginepb",
         "//pkg/testutils",

pkg/kv/kvclient/kvcoord/condensable_span_set.go

@@ -51,7 +51,7 @@ func (s *condensableSpanSet) insert(spans ...roachpb.Span) {
 // increase the overall bounds of the span set, but will eliminate duplicated
 // spans and combine overlapping spans.
 //
-// The method has the side effect of sorting the stable write set.
+// The method has the side effect of sorting the set.
 func (s *condensableSpanSet) mergeAndSort() {
 	oldLen := len(s.s)
 	s.s, _ = roachpb.MergeSpans(&s.s)
@@ -175,6 +175,61 @@ func (s *condensableSpanSet) clear() {
 	*s = condensableSpanSet{}
 }
 
+// estimateSize returns the size that the spanSet would grow to if spans were
+// added to the set. As a side-effect, the receiver might get its spans merged.
+//
+// The result doesn't take into consideration the effect of condensing the
+// spans, but might take into consideration the effects of merging the spans
+// (which is not a lossy operation): mergeThresholdBytes instructs the
+// simulation to perform merging and de-duping if the size grows over this
+// threshold.
+func (s *condensableSpanSet) estimateSize(spans []roachpb.Span, mergeThresholdBytes int64) int64 {
+	var bytes int64
+	for _, sp := range spans {
+		bytes += spanSize(sp)
+	}
+	{
+		estimate := s.bytes + bytes
+		if estimate <= mergeThresholdBytes {
+			return estimate
+		}
+	}
+
+	// Merge and de-dupe in the hope of saving some space.
+
+	// First, merge the existing spans in-place. Doing it in-place instead of
+	// operating on a copy avoids the risk of quadratic behavior over a series of
+	// estimateSize() calls, where each call has to repeatedly merge a copy in
+	// order to discover that the merger saves a lot of space.
+	s.mergeAndSort()
+
+	// See if merging s was enough.
+	estimate := s.bytes + bytes
+	if estimate <= mergeThresholdBytes {
+		return estimate
+	}
+
+	// Try harder - merge (a copy of) the existing spans with the new spans.
+	spans = append(spans, s.s...)
+	lenBeforeMerge := len(spans)
+	spans, _ = roachpb.MergeSpans(&spans)
+	if len(spans) == lenBeforeMerge {
+		// Nothing changed -i.e. we failed to merge any spans.
+		return estimate
+	}
+	// Recompute the size.
+	bytes = 0
+	for _, sp := range spans {
+		bytes += spanSize(sp)
+	}
+	return bytes
+}
+
+// bytesSize returns the size of the tracked spans.
+func (s *condensableSpanSet) bytesSize() int64 {
+	return s.bytes
+}
+
 func spanSize(sp roachpb.Span) int64 {
 	return int64(len(sp.Key) + len(sp.EndKey))
 }

pkg/kv/kvclient/kvcoord/condensable_span_set_test.go

@@ -31,3 +31,50 @@ func TestCondensableSpanSetMergeContiguousSpans(t *testing.T) {
 	s.mergeAndSort()
 	require.Equal(t, int64(2), s.bytes)
 }
+
+func TestCondensableSpanSetEstimateSize(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ab := roachpb.Span{Key: roachpb.Key("a"), EndKey: roachpb.Key("b")}
+	bc := roachpb.Span{Key: roachpb.Key("b"), EndKey: roachpb.Key("c")}
+
+	tests := []struct {
+		name           string
+		set            []roachpb.Span
+		newSpans       []roachpb.Span
+		mergeThreshold int64
+		expEstimate    int64
+	}{
+		{
+			name:           "new spans fit without merging",
+			set:            []roachpb.Span{ab, bc},
+			newSpans:       []roachpb.Span{ab},
+			mergeThreshold: 100,
+			expEstimate:    6,
+		},
+		{
+			// The set gets merged, the new spans don't.
+			name:           "set needs merging",
+			set:            []roachpb.Span{ab, bc},
+			newSpans:       []roachpb.Span{ab},
+			mergeThreshold: 5,
+			expEstimate:    4,
+		},
+		{
+			// The set gets merged, and then it gets merged again with the newSpans.
+			name:           "new spans fit without merging",
+			set:            []roachpb.Span{ab, bc},
+			newSpans:       []roachpb.Span{ab, bc},
+			mergeThreshold: 5,
+			expEstimate:    2,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s := condensableSpanSet{}
+			s.insert(tc.set...)
+			require.Equal(t, tc.expEstimate, s.estimateSize(tc.newSpans, tc.mergeThreshold))
+		})
+	}
+}