Merge #154025 #154485 #154535 #154540

154025: util: add yamlutil r=RaduBerinde a=RaduBerinde

Add an `UnmarshalStrict()` wrapper to match the same function in
yaml.v2 or earlier. Also add a `Marshal()` that indents to two spaces.

Epic: none
Release note: None

154485: roachprod/opentelemetry: add new CDC 25.4 metrics to DRT clusters r=log-head,asg0451 a=andyyang890

This patch adds the following metrics to DRT clusters:
* changefeed.progress_skew.span
* changefeed.progress_skew.table
* changefeed.sink_backpressure_nanos
* changefeed.stage.frontier_persistence.latency
* changefeed.stage.pts.manage.latency
* changefeed.stage.pts.manage_error.latency
* changefeed.stage.pts.create.latency

Epic: None

Release note: None

154535: lease: remove compatibility with pre-24.2 lease table r=rafiss a=rafiss

Epic: None
Release note: None

154540: backupsink: fix ooming in TestFileSSTSinkWrite r=jeffswenson a=jeffswenson

Previously, the test would allocate many 20 MiB+ buffers which sometimes
oom'd the test runner. Now, the buffers allocated by the test are
~10KiB. It seems that the main reason the large buffers were required by
the test is the flushes measure the compressed sst size. The test was
generating long strings of zero values which compressed to almost
nothing, so using correctly sized buffers would not trigger the required
flushes.

Release note: none
Informs: #154360
Informs: #153662
Informs: #153162

Co-authored-by: Radu Berinde <[email protected]>
Co-authored-by: Andy Yang <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>
Co-authored-by: Jeff Swenson <[email protected]>

Author: 5 people

pkg/BUILD.bazel

@@ -840,6 +840,7 @@ ALL_TESTS = [
     "//pkg/util/uuid:uuid_test",
     "//pkg/util/vector:vector_test",
     "//pkg/util/vfsutil:vfsutil_test",
+    "//pkg/util/yamlutil:yamlutil_test",
     "//pkg/util:util_test",
     "//pkg/workload/bank:bank_test",
     "//pkg/workload/cli:cli_test",
@@ -2851,6 +2852,8 @@ GO_TARGETS = [
     "//pkg/util/vector:vector_test",
     "//pkg/util/vfsutil:vfsutil",
     "//pkg/util/vfsutil:vfsutil_test",
+    "//pkg/util/yamlutil:yamlutil",
+    "//pkg/util/yamlutil:yamlutil_test",
     "//pkg/util:util",
     "//pkg/util:util_test",
     "//pkg/workload/bank:bank",

pkg/backup/backupsink/file_sst_sink_test.go

@@ -8,6 +8,7 @@ package backupsink
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"reflect"
 	"strconv"
 	"strings"
@@ -122,6 +123,15 @@ func TestFileSSTSinkExtendOneFile(t *testing.T) {
 	require.Equal(t, 1, len(progDetails.Files))
 }
 
+func randomValue(n int64) []byte {
+	// Create random data so that it does not compress well.
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = byte(rand.Int())
+	}
+	return b
+}
+
 // TestFileSSTSinkWrite tests the contents of flushed files and the internal
 // unflushed files of the FileSSTSink under different write scenarios. Each test
 // writes a sequence of exportedSpans into a FileSSTSink. The test then verifies
@@ -133,6 +143,12 @@ func TestFileSSTSinkWrite(t *testing.T) {
 	defer log.Scope(t).Close(t)
 
 	ctx := context.Background()
+	testTargetFileSize := int64(10 << 10)
+
+	// Override the fileSpanByteLimit so we can test going over the limit without
+	// needing large buffers that may oom the test node.
+	defer func(oldLimit int64) { fileSpanByteLimit = oldLimit }(fileSpanByteLimit)
+	fileSpanByteLimit = testTargetFileSize / 2
 
 	type testCase struct {
 		name              string
@@ -145,8 +161,7 @@ func TestFileSSTSinkWrite(t *testing.T) {
 		//
 		// TODO (msbutler): we currently don't test expected error handling. If this
 		// is non-empty, we just skip the test.
-		errorExplanation  string
-		noSSTSizeOverride bool
+		errorExplanation string
 	}
 
 	for _, tt := range []testCase{{name: "out-of-order-key-boundary",
@@ -278,7 +293,7 @@ func TestFileSSTSinkWrite(t *testing.T) {
 		{
 			name: "size-flush",
 			exportSpans: []ExportedSpan{
-				newExportedSpanBuilder("a", "c").withKVs([]kvAndTS{{key: "a", timestamp: 10, value: make([]byte, 20<<20)}, {key: "b", timestamp: 10}}).build(),
+				newExportedSpanBuilder("a", "c").withKVs([]kvAndTS{{key: "a", timestamp: 10, value: randomValue(testTargetFileSize)}, {key: "b", timestamp: 10}}).build(),
 				newExportedSpanBuilder("d", "f").withKVs([]kvAndTS{{key: "d", timestamp: 10}, {key: "e", timestamp: 10}}).build(),
 			},
 			flushedSpans: []roachpb.Spans{
@@ -292,7 +307,7 @@ func TestFileSSTSinkWrite(t *testing.T) {
 			// No flush can occur between two versions of the same key. Further, we must combine flushes which split a row.
 			name: "no-size-flush-if-mid-mvcc",
 			exportSpans: []ExportedSpan{
-				newRawExportedSpanBuilder(s2k0("a"), s2k0("c"), s2k0("c")).withKVs([]kvAndTS{{key: "a", timestamp: 10, value: make([]byte, 20<<20)}, {key: "c", timestamp: 10}}).build(),
+				newRawExportedSpanBuilder(s2k0("a"), s2k0("c"), s2k0("c")).withKVs([]kvAndTS{{key: "a", timestamp: 10, value: randomValue(testTargetFileSize)}, {key: "c", timestamp: 10}}).build(),
 				newRawExportedSpanBuilder(s2k0("c"), s2k0("f"), s2k0("f")).withKVs([]kvAndTS{{key: "c", timestamp: 8}, {key: "f", timestamp: 10}}).build(),
 			},
 			flushedSpans: []roachpb.Spans{},
@@ -305,9 +320,9 @@ func TestFileSSTSinkWrite(t *testing.T) {
 			name: "no-size-flush-mid-col-family",
 			exportSpans: []ExportedSpan{
 				newRawExportedSpanBuilder(s2kWithColFamily("c", 0), s2kWithColFamily("c", 1), s2kWithColFamily("c", 1)).withKVs([]kvAndTS{
-					{key: "c", timestamp: 10, value: make([]byte, 20<<20)}}).build(),
+					{key: "c", timestamp: 10, value: randomValue(testTargetFileSize)}}).build(),
 				newRawExportedSpanBuilder(s2kWithColFamily("c", 1), s2kWithColFamily("c", 2), s2kWithColFamily("c", 2)).withKVs([]kvAndTS{
-					{key: "c", timestamp: 10, value: make([]byte, 20<<20)}}).buildWithEncoding(func(stingedKey string) roachpb.Key { return s2kWithColFamily(stingedKey, 1) }),
+					{key: "c", timestamp: 10, value: randomValue(testTargetFileSize)}}).buildWithEncoding(func(stingedKey string) roachpb.Key { return s2kWithColFamily(stingedKey, 1) }),
 			},
 			flushedSpans: []roachpb.Spans{},
 			unflushedSpans: []roachpb.Spans{
@@ -318,7 +333,7 @@ func TestFileSSTSinkWrite(t *testing.T) {
 			// It's safe to flush at the range boundary.
 			name: "size-flush-at-range-boundary",
 			exportSpans: []ExportedSpan{
-				newRawExportedSpanBuilder(s2k("a"), s2k("d"), s2k("d")).withKVs([]kvAndTS{{key: "a", timestamp: 10, value: make([]byte, 20<<20)}, {key: "c", timestamp: 10}}).build(),
+				newRawExportedSpanBuilder(s2k("a"), s2k("d"), s2k("d")).withKVs([]kvAndTS{{key: "a", timestamp: 10, value: randomValue(testTargetFileSize)}, {key: "c", timestamp: 10}}).build(),
 			},
 			flushedSpans: []roachpb.Spans{
 				{{Key: s2k("a"), EndKey: s2k("d")}},
@@ -332,7 +347,7 @@ func TestFileSSTSinkWrite(t *testing.T) {
 			// row between two column families.
 			name: "trim-resume-key",
 			exportSpans: []ExportedSpan{
-				newRawExportedSpanBuilder(s2k0("a"), s2k0("c"), s2k("c")).withKVs([]kvAndTS{{key: "a", timestamp: 10, value: make([]byte, 20<<20)}}).build(),
+				newRawExportedSpanBuilder(s2k0("a"), s2k0("c"), s2k("c")).withKVs([]kvAndTS{{key: "a", timestamp: 10, value: randomValue(testTargetFileSize)}}).build(),
 			},
 			flushedSpans: []roachpb.Spans{
 				{{Key: s2k0("a"), EndKey: s2k("c")}},
@@ -344,24 +359,23 @@ func TestFileSSTSinkWrite(t *testing.T) {
 			// even if the next span's start key matches the file's end key.
 			name: "file-size-cut",
 			exportSpans: []ExportedSpan{
-				newExportedSpanBuilder("a", "c").withKVs([]kvAndTS{{key: "a", timestamp: 10, value: make([]byte, 64<<20)}, {key: "b", timestamp: 10}}).build(),
+				newExportedSpanBuilder("a", "c").withKVs([]kvAndTS{{key: "a", timestamp: 10, value: randomValue(fileSpanByteLimit)}, {key: "b", timestamp: 10}}).build(),
 				newExportedSpanBuilder("c", "f").withKVs([]kvAndTS{{key: "c", timestamp: 10}, {key: "e", timestamp: 10}}).build(),
 			},
 			flushedSpans: []roachpb.Spans{},
 			unflushedSpans: []roachpb.Spans{
 				{{Key: s2k0("a"), EndKey: s2k0("c")}, {Key: s2k0("c"), EndKey: s2k0("f")}},
 			},
-			noSSTSizeOverride: true,
 		},
 		{
 			// No file cut can occur between the two column families of the same row,
 			// even if the file is sufficiently large to get cut.
 			name: "no-file-cut-mid-col-family",
 			exportSpans: []ExportedSpan{
 				newRawExportedSpanBuilder(s2kWithColFamily("c", 0), s2kWithColFamily("c", 1), s2kWithColFamily("c", 1)).withKVs([]kvAndTS{
-					{key: "c", timestamp: 10, value: make([]byte, 65<<20)}}).build(),
+					{key: "c", timestamp: 10, value: randomValue(testTargetFileSize)}}).build(),
 				newRawExportedSpanBuilder(s2kWithColFamily("c", 1), s2kWithColFamily("c", 2), s2kWithColFamily("c", 2)).withKVs([]kvAndTS{
-					{key: "c", timestamp: 10, value: make([]byte, 20<<20)}}).buildWithEncoding(func(stingedKey string) roachpb.Key { return s2kWithColFamily(stingedKey, 1) }),
+					{key: "c", timestamp: 10, value: randomValue(testTargetFileSize / 2)}}).buildWithEncoding(func(stingedKey string) roachpb.Key { return s2kWithColFamily(stingedKey, 1) }),
 			},
 			flushedSpans: []roachpb.Spans{},
 			unflushedSpans: []roachpb.Spans{
@@ -377,9 +391,7 @@ func TestFileSSTSinkWrite(t *testing.T) {
 					return
 				}
 				st := cluster.MakeTestingClusterSettings()
-				if !tt.noSSTSizeOverride {
-					targetFileSize.Override(ctx, &st.SV, 10<<10)
-				}
+				targetFileSize.Override(ctx, &st.SV, testTargetFileSize)
 
 				sink, store := fileSSTSinkTestSetup(t, st, elide)
 				defer func() {
@@ -534,7 +546,7 @@ func TestFileSSTSinkStats(t *testing.T) {
 			sinkStats{hlc.Timestamp{WallTime: 10}, 3, 3, 0, 0, 0, 1}},
 		{
 			// Write an exported span that comes after all spans so far. This span has enough data for a size flush.
-			newExportedSpanBuilder("g", "h").withKVs([]kvAndTS{{key: "g", timestamp: 10, value: make([]byte, 20<<20)}}).build(),
+			newExportedSpanBuilder("g", "h").withKVs([]kvAndTS{{key: "g", timestamp: 10, value: randomValue(10 << 10)}}).build(),
 			sinkStats{hlc.Timestamp{WallTime: 0}, 0, 4, 1, 0, 1, 1}},
 		{
 			// Write the first exported span after the flush.

pkg/roachprod/install/BUILD.bazel

@@ -85,6 +85,7 @@ go_test(
         "//pkg/util/randutil",
         "//pkg/util/retry",
         "//pkg/util/syncutil",
+        "//pkg/util/yamlutil",
         "@com_github_cockroachdb_datadriven//:datadriven",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",

pkg/roachprod/install/cluster_settings_test.go

@@ -6,9 +6,9 @@
 package install
 
 import (
-	"bytes"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/util/yamlutil"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
 )
@@ -22,10 +22,7 @@ func TestClusterSettingOptionListCodec(t *testing.T) {
 	require.NoError(t, err)
 
 	var decOpts ClusterSettingOptionList
-	dec := yaml.NewDecoder(bytes.NewReader(data))
-	dec.KnownFields(true)
-	err = dec.Decode(&decOpts)
-	require.NoError(t, err)
+	require.NoError(t, yamlutil.UnmarshalStrict(data, &decOpts))
 
 	require.Equal(t, opts, decOpts)
 	require.Equal(t, MakeClusterSettings(opts...), MakeClusterSettings(decOpts...))

pkg/roachprod/opentelemetry/cockroachdb_metrics.go

@@ -405,6 +405,8 @@ var cockroachdbMetrics = map[string]string{
 	"changefeed_parallel_io_result_queue_nanos_bucket":                    "changefeed.parallel_io_result_queue_nanos.bucket",
 	"changefeed_parallel_io_result_queue_nanos_count":                     "changefeed.parallel_io_result_queue_nanos.count",
 	"changefeed_parallel_io_result_queue_nanos_sum":                       "changefeed.parallel_io_result_queue_nanos.sum",
+	"changefeed_progress_skew_span":                                       "changefeed.progress_skew.span",
+	"changefeed_progress_skew_table":                                      "changefeed.progress_skew.table",
 	"changefeed_queue_time_nanos":                                         "changefeed.queue_time",
 	"changefeed_running":                                                  "changefeed.running",
 	"changefeed_schema_registry_registrations":                            "changefeed.schema_registry.registrations",
@@ -415,6 +417,10 @@ var cockroachdbMetrics = map[string]string{
 	"changefeed_sink_batch_hist_nanos_bucket":                             "changefeed.sink_batch_hist_nanos.bucket",
 	"changefeed_sink_batch_hist_nanos_count":                              "changefeed.sink_batch_hist_nanos.count",
 	"changefeed_sink_batch_hist_nanos_sum":                                "changefeed.sink_batch_hist_nanos.sum",
+	"changefeed_sink_backpressure_nanos":                                  "changefeed.sink_backpressure_nanos",
+	"changefeed_sink_backpressure_nanos_bucket":                           "changefeed.sink_backpressure_nanos.bucket",
+	"changefeed_sink_backpressure_nanos_count":                            "changefeed.sink_backpressure_nanos.count",
+	"changefeed_sink_backpressure_nanos_sum":                              "changefeed.sink_backpressure_nanos.sum",
 	"changefeed_sink_io_inflight":                                         "changefeed.sink_io_inflight",
 	"changefeed_size_based_flushes":                                       "changefeed.size_based_flushes",
 	"changefeed_stage_checkpoint_job_progress_latency":                    "changefeed.stage.checkpoint_job_progress.latency",
@@ -433,6 +439,10 @@ var cockroachdbMetrics = map[string]string{
 	"changefeed_stage_encode_latency_bucket":                              "changefeed.stage.encode.latency.bucket",
 	"changefeed_stage_encode_latency_count":                               "changefeed.stage.encode.latency.count",
 	"changefeed_stage_encode_latency_sum":                                 "changefeed.stage.encode.latency.sum",
+	"changefeed_stage_frontier_persistence_latency":                       "changefeed.stage.frontier_persistence.latency",
+	"changefeed_stage_frontier_persistence_latency_bucket":                "changefeed.stage.frontier_persistence.latency.bucket",
+	"changefeed_stage_frontier_persistence_latency_count":                 "changefeed.stage.frontier_persistence.latency.count",
+	"changefeed_stage_frontier_persistence_latency_sum":                   "changefeed.stage.frontier_persistence.latency.sum",
 	"changefeed_stage_kv_feed_buffer_latency":                             "changefeed.stage.kv_feed_buffer.latency",
 	"changefeed_stage_kv_feed_buffer_latency_bucket":                      "changefeed.stage.kv_feed_buffer.latency.bucket",
 	"changefeed_stage_kv_feed_buffer_latency_count":                       "changefeed.stage.kv_feed_buffer.latency.count",
@@ -441,6 +451,18 @@ var cockroachdbMetrics = map[string]string{
 	"changefeed_stage_kv_feed_wait_for_table_event_latency_bucket":        "changefeed.stage.kv_feed_wait_for_table_event.latency.bucket",
 	"changefeed_stage_kv_feed_wait_for_table_event_latency_count":         "changefeed.stage.kv_feed_wait_for_table_event.latency.count",
 	"changefeed_stage_kv_feed_wait_for_table_event_latency_sum":           "changefeed.stage.kv_feed_wait_for_table_event.latency.sum",
+	"changefeed_stage_pts_create_latency":                                 "changefeed.stage.pts.create.latency",
+	"changefeed_stage_pts_create_latency_bucket":                          "changefeed.stage.pts.create.latency.bucket",
+	"changefeed_stage_pts_create_latency_count":                           "changefeed.stage.pts.create.latency.count",
+	"changefeed_stage_pts_create_latency_sum":                             "changefeed.stage.pts.create.latency.sum",
+	"changefeed_stage_pts_manage_error_latency":                           "changefeed.stage.pts.manage_error.latency",
+	"changefeed_stage_pts_manage_error_latency_bucket":                    "changefeed.stage.pts.manage_error.latency.bucket",
+	"changefeed_stage_pts_manage_error_latency_count":                     "changefeed.stage.pts.manage_error.latency.count",
+	"changefeed_stage_pts_manage_error_latency_sum":                       "changefeed.stage.pts.manage_error.latency.sum",
+	"changefeed_stage_pts_manage_latency":                                 "changefeed.stage.pts.manage.latency",
+	"changefeed_stage_pts_manage_latency_bucket":                          "changefeed.stage.pts.manage.latency.bucket",
+	"changefeed_stage_pts_manage_latency_count":                           "changefeed.stage.pts.manage.latency.count",
+	"changefeed_stage_pts_manage_latency_sum":                             "changefeed.stage.pts.manage.latency.sum",
 	"changefeed_stage_rangefeed_buffer_checkpoint_latency":                "changefeed.stage.rangefeed_buffer_checkpoint.latency",
 	"changefeed_stage_rangefeed_buffer_checkpoint_latency_bucket":         "changefeed.stage.rangefeed_buffer_checkpoint.latency.bucket",
 	"changefeed_stage_rangefeed_buffer_checkpoint_latency_count":          "changefeed.stage.rangefeed_buffer_checkpoint.latency.count",

pkg/roachprod/roachprodutil/codec/BUILD.bazel

@@ -16,6 +16,7 @@ go_test(
     ],
     embed = [":codec"],
     deps = [
+        "//pkg/util/yamlutil",
         "@com_github_stretchr_testify//require",
         "@in_gopkg_yaml_v3//:yaml_v3",
     ],

pkg/roachprod/roachprodutil/codec/types_test.go

@@ -6,10 +6,10 @@
 package codec
 
 import (
-	"bytes"
 	"reflect"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/util/yamlutil"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
 )
@@ -60,10 +60,7 @@ func TestDynamicTypes(t *testing.T) {
 	require.NoError(t, err)
 
 	var decoded ListWrapper[Animal]
-	dec := yaml.NewDecoder(bytes.NewReader(data))
-	dec.KnownFields(true)
-	err = dec.Decode(&decoded)
-	require.NoError(t, err)
+	require.NoError(t, yamlutil.UnmarshalStrict(data, &decoded))
 
 	// Verify equality of the decoded list and the original list.
 	require.Equal(t, animals, decoded.Get())

pkg/sql/catalog/lease/count.go

@@ -94,7 +94,6 @@ func countLeasesWithDetail(
 ) (countDetail, error) {
 	var whereClause []string
 	forceMultiRegionQuery := false
-	useBytesOnRetry := false
 	for _, t := range versions {
 		versionClause := ""
 		if !forAnyVersion {
@@ -120,19 +119,13 @@ func countLeasesWithDetail(
 		// entire table.
 		if (cachedDatabaseRegions != nil && cachedDatabaseRegions.IsMultiRegion()) ||
 			forceMultiRegionQuery {
-			// If we are injecting a raw leases descriptors, that will not have the enum
-			// type set, so convert the region to byte equivalent physical representation.
-			detail, err = countLeasesByRegion(ctx, txn, prober, regionMap, cachedDatabaseRegions,
-				useBytesOnRetry, at, whereClause)
+			detail, err = countLeasesByRegion(ctx, txn, prober, regionMap, at, whereClause)
 		} else {
 			detail, err = countLeasesNonMultiRegion(ctx, txn, at, whereClause)
 		}
 		// If any transient region column errors occur then we should retry the count query.
 		if isTransientRegionColumnError(err) {
 			forceMultiRegionQuery = true
-			// If the query was already multi-region aware, then the system database is MR,
-			// but our lease descriptor has not been upgraded yet.
-			useBytesOnRetry = cachedDatabaseRegions != nil && cachedDatabaseRegions.IsMultiRegion()
 			return txn.KV().GenerateForcedRetryableErr(ctx, "forcing retry once with MR columns")
 		}
 
@@ -187,15 +180,10 @@ func countLeasesByRegion(
 	txn isql.Txn,
 	prober regionliveness.Prober,
 	regionMap regionliveness.LiveRegions,
-	cachedDBRegions regionliveness.CachedDatabaseRegions,
-	convertRegionsToBytes bool,
 	at hlc.Timestamp,
 	whereClauses []string,
 ) (countDetail, error) {
 	regionClause := "crdb_region=$2::system.crdb_internal_region"
-	if convertRegionsToBytes {
-		regionClause = "crdb_region=$2"
-	}
 	stmt := fmt.Sprintf(
 		`SELECT %[1]s FROM system.public.lease AS OF SYSTEM TIME '%[2]s' WHERE %[3]s `,
 		getCountLeaseColumns(),
@@ -204,28 +192,13 @@ func countLeasesByRegion(
 	)
 	var detail countDetail
 	if err := regionMap.ForEach(func(region string) error {
-		regionEnumValue := region
-		// The leases table descriptor injected does not have the type of the column
-		// set to the region enum type. So, instead convert the logical value to
-		// the physical one for comparison.
-		// TODO(fqazi): In 24.2 when this table format is default we can stop using
-		// synthetic descriptors and use the first code path.
-		if convertRegionsToBytes {
-			regionTypeDesc := cachedDBRegions.GetRegionEnumTypeDesc().AsRegionEnumTypeDescriptor()
-			for i := 0; i < regionTypeDesc.NumEnumMembers(); i++ {
-				if regionTypeDesc.GetMemberLogicalRepresentation(i) == region {
-					regionEnumValue = string(regionTypeDesc.GetMemberPhysicalRepresentation(i))
-					break
-				}
-			}
-		}
 		var values tree.Datums
 		queryRegionRows := func(countCtx context.Context) error {
 			var err error
 			values, err = txn.QueryRowEx(
 				countCtx, "count-leases", txn.KV(),
 				sessiondata.NodeUserSessionDataOverride,
-				stmt, at.GoTime(), regionEnumValue,
+				stmt, at.GoTime(), region,
 			)
 			return err
 		}