Merge #150854 #150933 #150964 #150975

150854: roachtest: don't specify virtual cluster in debug zips r=golgeek,herkolategan a=DarrylWong

Previously, we were always passing the system tenant when taking debug zips in case a roachtest had overriden the default virtual cluster. However, the debug zip command already attempts to fetch all virtual clusters by passing the --ccluster flag for each tenant. This behavior was suppresed due to our hard coding of the system tenant, causing all debug zips to be of the system tenant.

This changes it to instead pass nothing so the url has no existing --ccluster flag.

Fixes: #143841
Release note: none

150933: server: change sort order of local hot ranges to descending r=angles-n-daemons a=angles-n-daemons

Currently the hot ranges logging system expects that ranges are sorted in descending order, so that it can inspect the values from greatest to least
([link](https://github.com/cockroachdb/cockroach/blob/master/pkg/server/application_api/storage_inspection_test.go#L377)). This is not the case, as the system currently returns those values in ascending order.

To fix this, we just reverse the comparison used for sorting at the local hot ranges collector.

Epic: none
Fixes: none
Release note: none

150964: changefeedccl: support rangefeed bulk delivery r=andyyang890,dt a=asg0451

Add support for the new rangefeed bulk delivery
feature in changefeeds. This should improve
catchup scan performance.

Fixes: #150862

Release note: None


150975: roachtest: fix test failed message r=herkolategan a=DarrylWong

Release note: none
Epic: none
Fixes: none

Co-authored-by: DarrylWong <[email protected]>
Co-authored-by: Brian Dillmann <[email protected]>
Co-authored-by: Miles Frankel <[email protected]>

Author: 4 people

pkg/ccl/changefeedccl/kvfeed/physical_kv_feed.go

@@ -93,7 +93,11 @@ func (p rangefeedFactory) Run(ctx context.Context, sink kvevent.Writer, cfg rang
 	}
 	g := ctxgroup.WithContext(ctx)
 	g.GoCtx(feed.addEventsToBuffer)
-	var rfOpts []kvcoord.RangeFeedOption
+
+	// Bulk delivery is an optimization that decreases rangefeed overhead during
+	// catchup scans. It results in the emission of BulkEvents instead of
+	// individual events where possible.
+	rfOpts := []kvcoord.RangeFeedOption{kvcoord.WithBulkDelivery()}
 	if cfg.WithDiff {
 		rfOpts = append(rfOpts, kvcoord.WithDiff())
 	}
@@ -132,6 +136,78 @@ func quantizeTS(ts hlc.Timestamp, granularity time.Duration) hlc.Timestamp {
 	}
 }
 
+func (p *rangefeed) handleRangefeedEvent(ctx context.Context, e *kvpb.RangeFeedEvent) error {
+	switch t := e.GetValue().(type) {
+	case *kvpb.RangeFeedValue:
+		if p.cfg.Knobs.OnRangeFeedValue != nil {
+			if err := p.cfg.Knobs.OnRangeFeedValue(); err != nil {
+				return err
+			}
+		}
+		stop := p.st.RangefeedBufferValue.Start()
+		if err := p.memBuf.Add(
+			ctx, kvevent.MakeKVEvent(e),
+		); err != nil {
+			return err
+		}
+		stop()
+	case *kvpb.RangeFeedCheckpoint:
+		ev := e.ShallowCopy()
+		ev.Checkpoint.ResolvedTS = quantizeTS(ev.Checkpoint.ResolvedTS, p.cfg.WithFrontierQuantize)
+		if resolvedTs := ev.Checkpoint.ResolvedTS; !resolvedTs.IsEmpty() && resolvedTs.Less(p.cfg.Frontier) {
+			// RangeFeed happily forwards any closed timestamps it receives as
+			// soon as there are no outstanding intents under them.
+			// Changefeeds don't care about these at all, so throw them out.
+			return nil
+		}
+		if p.knobs.ShouldSkipCheckpoint != nil && p.knobs.ShouldSkipCheckpoint(t) {
+			return nil
+		}
+		stop := p.st.RangefeedBufferCheckpoint.Start()
+		if err := p.memBuf.Add(
+			ctx, kvevent.MakeResolvedEvent(ev, jobspb.ResolvedSpan_NONE),
+		); err != nil {
+			return err
+		}
+		stop()
+	case *kvpb.RangeFeedSSTable:
+		// For now, we just error on SST ingestion, since we currently don't
+		// expect SST ingestion into spans with active changefeeds.
+		return errors.Errorf("unexpected SST ingestion: %v", t)
+	case *kvpb.RangeFeedBulkEvents:
+		// TODO(#138346): We can handle these more gracefully once we
+		// migrate to the new rangefeed client. Until then, this is
+		// still an improvement over not using these.
+		for _, e := range t.Events {
+			if err := p.handleRangefeedEvent(ctx, e); err != nil {
+				return err
+			}
+		}
+	case *kvpb.RangeFeedDeleteRange:
+		// For now, we just ignore on MVCC range tombstones. These are currently
+		// only expected to be used by schema GC and IMPORT INTO, and such spans
+		// should not have active changefeeds across them, at least at the times
+		// of interest. A case where one will show up in a changefeed is when
+		// the primary index changes while we're watching it and then the old
+		// primary index is dropped. In this case, we'll get a schema event to
+		// restart into the new primary index, but the DeleteRange may come
+		// through before the schema event.
+		//
+		// TODO(erikgrinaker): Write an end-to-end test which verifies that an
+		// IMPORT INTO which gets rolled back using MVCC range tombstones will
+		// not be visible to a changefeed, neither when it was started before
+		// the import or when resuming from a timestamp before the import. The
+		// table decriptor should be marked as offline during the import, and
+		// catchup scans should detect that this happened and prevent reading
+		// anything in that timespan. See:
+		// https://github.com/cockroachdb/cockroach/issues/70433
+		return nil
+	default:
+		return errors.Errorf("unexpected RangeFeedEvent variant %v", t)
+	}
+	return nil
+}
+
 // addEventsToBuffer consumes rangefeed events from `p.eventCh`, transforms
 // them to kvevent.Event's, and pushes them into `p.memBuf`.
 func (p *rangefeed) addEventsToBuffer(ctx context.Context) error {
@@ -141,69 +217,8 @@ func (p *rangefeed) addEventsToBuffer(ctx context.Context) error {
 	for {
 		select {
 		case e := <-p.eventCh:
-			switch t := e.GetValue().(type) {
-			case *kvpb.RangeFeedValue:
-				if p.cfg.Knobs.OnRangeFeedValue != nil {
-					if err := p.cfg.Knobs.OnRangeFeedValue(); err != nil {
-						return err
-					}
-				}
-				stop := p.st.RangefeedBufferValue.Start()
-				if err := p.memBuf.Add(
-					ctx, kvevent.MakeKVEvent(e.RangeFeedEvent),
-				); err != nil {
-					return err
-				}
-				stop()
-			case *kvpb.RangeFeedCheckpoint:
-				ev := e.ShallowCopy()
-				ev.Checkpoint.ResolvedTS = quantizeTS(ev.Checkpoint.ResolvedTS, p.cfg.WithFrontierQuantize)
-				if resolvedTs := ev.Checkpoint.ResolvedTS; !resolvedTs.IsEmpty() && resolvedTs.Less(p.cfg.Frontier) {
-					// RangeFeed happily forwards any closed timestamps it receives as
-					// soon as there are no outstanding intents under them.
-					// Changefeeds don't care about these at all, so throw them out.
-					continue
-				}
-				if p.knobs.ShouldSkipCheckpoint != nil && p.knobs.ShouldSkipCheckpoint(t) {
-					continue
-				}
-				stop := p.st.RangefeedBufferCheckpoint.Start()
-				if err := p.memBuf.Add(
-					ctx, kvevent.MakeResolvedEvent(ev, jobspb.ResolvedSpan_NONE),
-				); err != nil {
-					return err
-				}
-				stop()
-			case *kvpb.RangeFeedSSTable:
-				// For now, we just error on SST ingestion, since we currently don't
-				// expect SST ingestion into spans with active changefeeds.
-				return errors.Errorf("unexpected SST ingestion: %v", t)
-			case *kvpb.RangeFeedBulkEvents:
-				// Should be disabled so this is unreachable.
-				return errors.Errorf("unexpected bulk delivery: %v", t)
-
-			case *kvpb.RangeFeedDeleteRange:
-				// For now, we just ignore on MVCC range tombstones. These are currently
-				// only expected to be used by schema GC and IMPORT INTO, and such spans
-				// should not have active changefeeds across them, at least at the times
-				// of interest. A case where one will show up in a changefeed is when
-				// the primary index changes while we're watching it and then the old
-				// primary index is dropped. In this case, we'll get a schema event to
-				// restart into the new primary index, but the DeleteRange may come
-				// through before the schema event.
-				//
-				// TODO(erikgrinaker): Write an end-to-end test which verifies that an
-				// IMPORT INTO which gets rolled back using MVCC range tombstones will
-				// not be visible to a changefeed, neither when it was started before
-				// the import or when resuming from a timestamp before the import. The
-				// table decriptor should be marked as offline during the import, and
-				// catchup scans should detect that this happened and prevent reading
-				// anything in that timespan. See:
-				// https://github.com/cockroachdb/cockroach/issues/70433
-				continue
-
-			default:
-				return errors.Errorf("unexpected RangeFeedEvent variant %v", t)
+			if err := p.handleRangefeedEvent(ctx, e.RangeFeedEvent); err != nil {
+				return err
 			}
 		case <-ctx.Done():
 			return ctx.Err()

pkg/cmd/roachtest/cluster.go

@@ -1389,12 +1389,14 @@ func (c *clusterImpl) FetchDebugZip(
 		for _, node := range nodes {
 			pgURLOpts := roachprod.PGURLOptions{
 				// `cockroach debug zip` does not support non root authentication.
-				Auth: install.AuthRootCert,
-				// request the system tenant specifically in case the test
-				// changed the default virtual cluster.
-				VirtualClusterName: install.SystemInterfaceName,
+				Auth:   install.AuthRootCert,
+				Secure: c.IsSecure(),
 			}
-			nodePgUrl, err := c.InternalPGUrl(ctx, l, c.Node(node), pgURLOpts)
+			// Use roachprod.PgURL directly as we want to bypass the default virtual cluster
+			// logic. The debug zip command already handles fetching all virtual clusters by passing
+			// the --ccluster for each tenant. Attempting to pass a --ccluster here will override
+			// that behavior and cause all debug zips to be of the same tenant.
+			urls, err := roachprod.PgURL(ctx, l, c.MakeNodes(c.Node(node)), install.CockroachNodeCertsDir, pgURLOpts)
 			if err != nil {
 				l.Printf("cluster.FetchDebugZip failed to retrieve PGUrl on node %d: %v", node, err)
 				continue
@@ -1409,7 +1411,7 @@ func (c *clusterImpl) FetchDebugZip(
 			cmd := roachtestutil.NewCommand("%s debug zip", test.DefaultCockroachPath).
 				Option("include-range-info").
 				Flag("exclude-files", fmt.Sprintf("'%s'", excludeFiles)).
-				Flag("url", fmt.Sprintf("'%s'", nodePgUrl[0])).
+				Flag("url", urls[0]).
 				MaybeFlag(c.IsSecure(), "certs-dir", install.CockroachNodeCertsDir).
 				Arg(zipName).
 				String()

pkg/cmd/roachtest/test_runner.go

@@ -471,7 +471,7 @@ func (r *testRunner) Run(
 		err = errors.Join(err, errSomeClusterProvisioningFailed)
 	}
 	if len(r.status.fail) > 0 {
-		shout(ctx, l, lopt.stdout, "%d tests failed", r.status.fail)
+		shout(ctx, l, lopt.stdout, "%d tests failed", len(r.status.fail))
 		err = errors.Join(err, errTestsFailed)
 	}
 	if err != nil {

pkg/roachprod/roachprod.go

@@ -1173,6 +1173,12 @@ func PgURL(
 	if len(urls) != len(nodes) {
 		return nil, errors.Errorf("have nodes %v, but urls %v from ips %v", nodes, urls, ips)
 	}
+	// We should never return an empty list of URLs as roachprod clusters always have at least
+	// one node. However, many callers of this function directly index into the slice returned,
+	// so check just in case.
+	if len(urls) == 0 {
+		return nil, errors.Newf("have nodes %v, but no urls were found", nodes)
+	}
 	return urls, nil
 }
 

pkg/server/application_api/storage_inspection_test.go

@@ -340,8 +340,6 @@ func TestHotRanges2Response(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.WithIssue(t, 146917)
-
 	srv := rangetestutils.StartServer(t)
 	defer srv.Stopper().Stop(context.Background())
 	ts := srv.ApplicationLayer()

pkg/server/status.go

@@ -3069,7 +3069,7 @@ func (s *systemStatusServer) localHotRanges(
 
 	// sort the slices by cpu
 	slices.SortFunc(resp.Ranges, func(a, b *serverpb.HotRangesResponseV2_HotRange) int {
-		return cmp.Compare(a.CPUTimePerSecond, b.CPUTimePerSecond)
+		return cmp.Compare(b.CPUTimePerSecond, a.CPUTimePerSecond)
 	})
 
 	// truncate the response if localLimit is set