Merge #154076 #154486 #154563 #154639

154076: cmd/roachtest: fix cdc/multi-region-execution-locality-tpcc r=andyyang890 a=log-head

This roachtest hangs waiting for changefeed to complete. Changing changefeed to an initial scan.

Additionally, this test was flaky. This is fixed by removing the check of the exact span distribution and checking only that more than one aggregator was planned. Additionally, before not all lease-holders would always be set to the same region. Now, all leaseholders will be in the specified region.

Additionally, adding to nightly test suite.

Epic: CRDB-38755
Fixes: #153825

Release note: None

154486: cloud: have resumable reader manage span lifetime r=jeffswenson a=jeffswenson

This change adds a span to the cloud.ResumableReader that is tracked for the lifetime of the reader. This ensures that if the reader implementation needs to issue a new HTTP request, the span is still valid and can be used to annotate logs.

Fixes: #153347
Release note: none

154563: metric: permit counter resets in update assertion r=jasonlmfong a=dhartunian

When we call `Update` on counter metrics, we have an assertion that panics in tests if the counter is set to a lower value than the prior update. In practice, counters are permitted to reset and 3rd party metric systems expect this to happen occasionally. The assertion logic is adjusted to permit a setting of zero when the value is lower.

Resolves: #154364
Resolves: #154365
Resolves: #154366
Resolves: #154367
Resolves: #154368

Epic: None
Release note: None

154639: sql: skip a big test that ooms on the race detector r=bghal a=bghal

The test still passes without the race detector and the smaller tests
still pass with the race detector.

Epic: none
Fixes: #154140
Fixes: #154243

Release note: None


Co-authored-by: Matthew Lougheed <[email protected]>
Co-authored-by: Jeff Swenson <[email protected]>
Co-authored-by: David Hartunian <[email protected]>
Co-authored-by: Brendan Gerrity <[email protected]>

Author: 5 people

pkg/cloud/azure/azure_storage.go

@@ -366,6 +366,7 @@ func (s *azureStorage) ReadFile(
 			}
 		}
 	}
+	// BUG: we should follow the azure retry setting here.
 	reader := resp.NewRetryReader(ctx, &azblob.RetryReaderOptions{MaxRetries: 3})
 	return ioctx.ReadCloserAdapter(reader), fileSize, nil
 }

pkg/cloud/cloud_io.go

@@ -242,6 +242,7 @@ type ReaderOpenerAt func(ctx context.Context, pos int64) (io.ReadCloser, int64,
 type ResumingReader struct {
 	Opener       ReaderOpenerAt   // Get additional content
 	Reader       io.ReadCloser    // Currently opened reader
+	ReaderSpan   *tracing.Span    // Span for the current reader, if Reader is non-nil
 	Filename     string           // Used for logging
 	Pos          int64            // How much data was received so far
 	Size         int64            // Total size of the file
@@ -284,6 +285,10 @@ func NewResumingReader(
 
 // Open opens the reader at its current offset.
 func (r *ResumingReader) Open(ctx context.Context) error {
+	if r.Reader != nil {
+		return errors.AssertionFailedf("reader already open")
+	}
+
 	if r.Size > 0 && r.Pos >= r.Size {
 		// Don't try to open a file if the size has been set and the position is
 		// at size. This generally results in an invalid range error for the
@@ -294,10 +299,18 @@ func (r *ResumingReader) Open(ctx context.Context) error {
 
 	return DelayedRetry(ctx, "Open", r.ErrFn, func() error {
 		var readErr error
+
+		ctx, span := tracing.ForkSpan(ctx, "resuming-reader")
 		r.Reader, r.Size, readErr = r.Opener(ctx, r.Pos)
 		if readErr != nil {
+			span.Finish()
 			return errors.Wrapf(readErr, "open %s", r.Filename)
 		}
+
+		// We hold onto the span for the lifetime of the reader because the reader
+		// may issue new HTTP requests after Open returns.
+		r.ReaderSpan = span
+
 		return nil
 	})
 }
@@ -340,10 +353,9 @@ func (r *ResumingReader) Read(ctx context.Context, p []byte) (int, error) {
 			}
 			log.Dev.Errorf(ctx, "Retry IO error: %s", lastErr)
 			lastErr = nil
-			if r.Reader != nil {
-				r.Reader.Close()
-			}
-			r.Reader = nil
+			// Ignore the error from Close(). We are already handling a read error
+			// so we know the handle is in a bad state.
+			_ = r.Close(ctx)
 		}
 	}
 
@@ -356,10 +368,14 @@ func (r *ResumingReader) Read(ctx context.Context, p []byte) (int, error) {
 
 // Close implements io.Closer.
 func (r *ResumingReader) Close(ctx context.Context) error {
-	if r.Reader != nil {
-		return r.Reader.Close()
+	if r.Reader == nil {
+		return nil
 	}
-	return nil
+
+	err := r.Reader.Close()
+	r.ReaderSpan.Finish()
+	r.Reader = nil
+	return err
 }
 
 // CheckHTTPContentRangeHeader parses Content-Range header and ensures that

pkg/cloud/cloudtestutils/BUILD.bazel

@@ -23,6 +23,8 @@ go_library(
         "//pkg/util/syncutil",
         "//pkg/util/sysutil",
         "//pkg/util/timeutil",
+        "//pkg/util/tracing",
+        "//pkg/util/tracing/tracingpb",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
         "@org_golang_x_sync//errgroup",

pkg/cloud/cloudtestutils/cloud_nemesis.go

@@ -19,6 +19,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/util/ioctx"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
+	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
@@ -39,9 +41,15 @@ func RunCloudNemesisTest(t *testing.T, storage cloud.ExternalStorage) {
 		listConcurrency:  1,
 	}
 
-	// We create a context here because we don't want to support a caller supplied
-	// cancelation signal.
-	ctx := context.Background()
+	// Create a root tracing span for the test to ensure ops create spans
+	// which will help flush out span lifetime bugs.
+	tracer := tracing.NewTracerWithOpt(
+		context.Background(),
+		tracing.WithUseAfterFinishOpt(true, true),
+		tracing.WithSpanReusePercent(0))
+	ctx, rootSpan := tracer.StartSpanCtx(context.Background(), "cloud-nemesis-test", tracing.WithRecording(tracingpb.RecordingStructured))
+	defer rootSpan.Finish()
+
 	if err := nemesis.run(ctx, 2*time.Minute); err != nil {
 		t.Fatalf("%+v", err)
 	}
@@ -85,9 +93,13 @@ func (c *cloudNemesis) run(ctx context.Context, duration time.Duration) error {
 	g, ctx := errgroup.WithContext(ctx)
 
 	g.Go(func() error {
-		time.Sleep(duration)
+		wait := time.NewTimer(duration)
+		select {
+		case <-ctx.Done():
+		case <-wait.C:
+		}
 		close(done)
-		return nil
+		return ctx.Err()
 	})
 
 	for i := 0; i < c.writeConcurrency; i++ {

pkg/cmd/roachtest/tests/cdc.go

@@ -20,6 +20,7 @@ import (
 	"encoding/pem"
 	"fmt"
 	"io"
+	"maps"
 	"math/big"
 	"math/rand"
 	"net"
@@ -31,6 +32,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"runtime"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -1950,26 +1952,18 @@ func getDiagramProcessors(ctx context.Context, db *gosql.DB) ([]any, error) {
 }
 
 type ChangefeedDistribution struct {
-	NodeToSpansWatched map[int]int
 	ZoneToSpansWatched map[string]int
 	TotalSpansWatched  int
 	TotalAggregators   int
-	TotalLeaseHolders  int
-	TotalRanges        int
-	NodeToZone         map[int]string
 }
 
 func getChangefeedDistribution(
 	processors []any, nodeToZone map[int]string, t test.Test,
 ) ChangefeedDistribution {
 	changefeedDistribution := ChangefeedDistribution{
-		NodeToSpansWatched: make(map[int]int),
 		ZoneToSpansWatched: make(map[string]int),
 		TotalSpansWatched:  0,
 		TotalAggregators:   0,
-		TotalLeaseHolders:  0,
-		TotalRanges:        0,
-		NodeToZone:         nodeToZone,
 	}
 	for _, p := range processors {
 		procMap, ok := p.(map[string]any)
@@ -1992,10 +1986,8 @@ func getChangefeedDistribution(
 					if len(matches) > 1 {
 						numWatches, err := strconv.Atoi(matches[1])
 						require.NoError(t, err)
-						changefeedDistribution.NodeToSpansWatched[int(nodeIdx)] += numWatches
 						changefeedDistribution.TotalSpansWatched += numWatches
-						changefeedDistribution.ZoneToSpansWatched[changefeedDistribution.NodeToZone[int(nodeIdx)]] += numWatches
-
+						changefeedDistribution.ZoneToSpansWatched[nodeToZone[int(nodeIdx)]] += numWatches
 					}
 				}
 			}
@@ -2004,42 +1996,36 @@ func getChangefeedDistribution(
 	return changefeedDistribution
 }
 
-func veryifyLeaseHolderDistribution(
-	db *gosql.DB, t test.Test, nodeToZone map[int]string,
-) map[string]int {
-	var rows *gosql.Rows
-	// Get lease holders for all ranges in tpcc database.
-	leaseHolderQuery := `SELECT r.start_pretty, r.replicas, r.replica_localities, r.lease_holder 
-	FROM crdb_internal.ranges r 
-	JOIN crdb_internal.tables t ON r.start_pretty like concat('/Table/', t.table_id::STRING,'%')
-	WHERE t.database_name = 'tpcc'`
-	rows, err := db.Query(leaseHolderQuery)
-	zoneToLeaseHolderCount := make(map[string]int)
-	require.NoError(t, err)
-	defer rows.Close()
-	for rows.Next() {
-		var startKeyPretty string
-		var replicas []uint8
-		var replicaLocalities []uint8
-		var leaseHolder int
-		require.NoError(t, rows.Scan(&startKeyPretty, &replicas, &replicaLocalities, &leaseHolder))
-		for indx := range replicas {
-			require.NotEqual(t, replicas[indx], 0)
-			replicas[indx]--
+func verifyLeaseHolderLocality(db *gosql.DB, t test.Test, primaryRegion string) {
+	leaseHolderQuery := `SELECT NOT EXISTS (
+	SELECT 1
+	FROM [SHOW CLUSTER RANGES WITH TABLES, DETAILS]
+	WHERE database_name = 'tpcc'
+		AND (lease_holder_locality IS DISTINCT FROM $1::STRING OR lease_holder_locality IS NULL)
+)`
+	t.L().Printf("Waiting for all lease holders to be in region %s", primaryRegion)
+	start := timeutil.Now()
+	ok := false
+	for {
+		if timeutil.Since(start) > 5*time.Minute {
+			t.Fatalf("Timeout waiting for lease holders to be in region %s; waited for %s", primaryRegion, timeutil.Since(start))
 		}
-		leaseHolder--
-		zoneToLeaseHolderCount[nodeToZone[leaseHolder]]++
+		require.NoError(t, db.QueryRow(leaseHolderQuery, primaryRegion).Scan(&ok))
+		if ok {
+			break
+		}
+		time.Sleep(time.Second)
 	}
-	return zoneToLeaseHolderCount
 }
 
 func registerCDC(r registry.Registry) {
 	r.Add(registry.TestSpec{
 		// This test
-		// 1. Creates a cluster with 3 nodes each in us-east and us-west
-		// 2. Runs a tpcc workload, then sets tpcc database to primary region us-west
-		// 3. Creates a changefeed with execution locality set to us-east
-		// 4. Gets the changefeed diagram and creates mappings
+		// 1. Creates a cluster with 3 nodes each in us-east and us-west;
+		// 2. Runs a tpcc workload, then congigures tpcc database to have lease holders in region us-west;
+		// 3. Creates a changefeed with execution locality set to us-east;
+		// 4. Gets the changefeed diagram and creates mappings;
+		// 5. Verifies that spans are assigned to multiple change aggregators in region us-east.
 
 		// This test is used to verify that ranges are evenly distributed across
 		// change aggregators in the execution_locality region while targeting tables
@@ -2051,7 +2037,7 @@ func registerCDC(r registry.Registry) {
 		Owner:            registry.OwnerCDC,
 		Cluster:          r.MakeClusterSpec(7, spec.Geo(), spec.GatherCores(), spec.GCEZones("us-east1-b,us-west1-b")),
 		CompatibleClouds: registry.OnlyGCE,
-		Suites:           registry.Suites(),
+		Suites:           registry.Suites(registry.Nightly),
 		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
 			nodeToZone := map[int]string{
 				0: "us-east1-b",
@@ -2064,17 +2050,24 @@ func registerCDC(r registry.Registry) {
 			ct := newCDCTester(ctx, t, c)
 			defer ct.Close()
 
-			ct.runTPCCWorkload(tpccArgs{warehouses: 100})
+			ct.runTPCCWorkload(tpccArgs{warehouses: 20})
 
 			var err error
-			_, err = ct.DB().Exec("ALTER DATABASE tpcc SET PRIMARY REGION 'us-west1'")
+			_, err = ct.DB().Exec(`ALTER DATABASE tpcc 
+CONFIGURE ZONE USING 
+	constraints = '{+region=us-west1: 1, +region=us-east1: 1}', 
+	lease_preferences = '[[+region=us-west1]]', num_replicas = 3`)
 			require.NoError(t, err)
 
+			// Verify lease holders are in us-west1-b.
+			verifyLeaseHolderLocality(ct.DB(), t, "cloud=gce,region=us-west1,zone=us-west1-b")
+
 			feed := ct.newChangefeed(feedArgs{
 				sinkType: cloudStorageSink,
 				targets:  allTpccTargets,
 				opts: map[string]string{
 					"execution_locality": "'region=us-east1'",
+					"initial_scan":       "'only'",
 				},
 			})
 			ct.waitForWorkload()
@@ -2083,18 +2076,12 @@ func registerCDC(r registry.Registry) {
 			processors, err := getDiagramProcessors(ctx, ct.DB())
 			require.NoError(t, err)
 
+			// Verify changefeed aggregators are distributed across nodes in region us-east.
 			changefeedDistribution := getChangefeedDistribution(processors, nodeToZone, t)
 			require.Greater(t, changefeedDistribution.TotalAggregators, 1)
-			for nodeIdx, spansWatched := range changefeedDistribution.NodeToSpansWatched {
-				require.LessOrEqual(t, spansWatched, changefeedDistribution.TotalSpansWatched/2, "nodeIdx %d watched %d spans, total spans watched %d", nodeIdx, spansWatched, changefeedDistribution.TotalSpansWatched)
-			}
-			require.Equal(t, 1, len(changefeedDistribution.ZoneToSpansWatched))
+			require.ElementsMatch(t, []string{"us-east1-b"}, slices.Collect(maps.Keys(changefeedDistribution.ZoneToSpansWatched)))
 			require.Equal(t, changefeedDistribution.ZoneToSpansWatched["us-east1-b"], changefeedDistribution.TotalSpansWatched)
-			zoneToLeaseHolderCount := veryifyLeaseHolderDistribution(ct.DB(), t, nodeToZone)
-			// Majority of lease holders should be in us-west1-b. Some may not, but most should.
-			if zoneToLeaseHolderCount["us-east1-b"] != 0 {
-				require.Greater(t, zoneToLeaseHolderCount["us-west1-b"]/zoneToLeaseHolderCount["us-east1-b"], 10)
-			}
+			require.Greater(t, changefeedDistribution.TotalSpansWatched, 0)
 		},
 	})
 	r.Add(registry.TestSpec{

pkg/sql/sem/tree/pretty_test.go

@@ -227,6 +227,7 @@ func TestPrettyVerify(t *testing.T) {
 func TestPrettyBigStatement(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
+	skip.UnderRace(t, "excessive memory usage")
 
 	// Create a SELECT statement with a 1 million item IN expression. Without
 	// mitigation, this can cause stack overflows - see #91197.

pkg/util/metric/metric.go

@@ -825,7 +825,7 @@ func (c *Counter) Inc(v int64) {
 // maintained elsewhere.
 func (c *Counter) Update(val int64) {
 	if buildutil.CrdbTestBuild {
-		if prev := c.count.Load(); val < prev {
+		if prev := c.count.Load(); val < prev && val != 0 {
 			panic(fmt.Sprintf("Counters should not decrease, prev: %d, new: %d.", prev, val))
 		}
 	}
@@ -1446,7 +1446,7 @@ func (cv *CounterVec) Update(labels map[string]string, v int64) {
 	}
 
 	currentValue := cv.Count(labels)
-	if currentValue > v {
+	if currentValue > v && v != 0 {
 		panic(fmt.Sprintf("Counters should not decrease, prev: %d, new: %d.", currentValue, v))
 	}