Merge #148241 #148698 #148756

148241: kvserver: enable kv.rangefeed.buffered_sender.enabled by default r=stevendanna a=wenyihu6

This patch enables kv.rangefeed.buffered_sender.enabled by default. We've gained
confidence in the feature through scale testing and enabling it metamorphically
since v25.2. We also currently don't believe enabling this feature would cause
any performance regressions or negative impact.

Epic: none
Release note: none

148698: changefeedccl: add cluster setting for changefeed max retry backoff r=rharding6373 a=rharding6373

When changefeeds enter a high-level retry loop, e.g. as part of a rolling restart, there is an exponential backoff applied. The default max backoff was 10m, but due to considerations in #146448 a lower 1m max was considered for some rolling restart cases. This PR makes the max backoff configurable via the non-pubic cluster setting changefeed.max_retry_backoff, so that most users can keep the old setting of 10m, which is better for degenerate scenarios when changefeeds might try to retry frequently due to cluster instability. It also adds a separate cluster setting, changefeed.retry_backoff_reset, which is the amount of time between retries before the backoff timer resets. Both settings have a default of 10m.

Epic: none
Fixes: #148467

Release note: None

148756: roachtest: adding defensive code in ceph/reef test r=sravotto a=sravotto

We have seen sporadic failures in the ceph tests, due to failures in creating users in the ceph object gateway.

To address this we are adding code to check that the gateway is up by submitting a read only request, before attempting to add the user.

Epic: none

Fixes: #148731

Release note: None


Co-authored-by: wenyihu6 <[email protected]>
Co-authored-by: rharding6373 <[email protected]>
Co-authored-by: Silvano Ravotto <[email protected]>

Author: 4 people

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -411,7 +411,9 @@ func coreChangefeed(
 	p.ExtendedEvalContext().ChangefeedState = localState
 	knobs, _ := p.ExecCfg().DistSQLSrv.TestingKnobs.Changefeed.(*TestingKnobs)
 
-	for r := getRetry(ctx); ; {
+	maxBackoff := changefeedbase.MaxRetryBackoff.Get(&p.ExecCfg().Settings.SV)
+	backoffReset := changefeedbase.RetryBackoffReset.Get(&p.ExecCfg().Settings.SV)
+	for r := getRetry(ctx, maxBackoff, backoffReset); ; {
 		if !r.Next() {
 			// Retry loop exits when context is canceled.
 			log.Infof(ctx, "core changefeed retry loop exiting: %s", ctx.Err())
@@ -1441,7 +1443,9 @@ func (b *changefeedResumer) resumeWithRetries(
 		b.mu.perNodeAggregatorStats[componentID] = *meta
 	}
 
-	for r := getRetry(ctx); r.Next(); {
+	maxBackoff := changefeedbase.MaxRetryBackoff.Get(&execCfg.Settings.SV)
+	backoffReset := changefeedbase.RetryBackoffReset.Get(&execCfg.Settings.SV)
+	for r := getRetry(ctx, maxBackoff, backoffReset); r.Next(); {
 		flowErr := maybeUpgradePreProductionReadyExpression(ctx, jobID, details, jobExec)
 
 		if flowErr == nil {

pkg/ccl/changefeedccl/changefeedbase/settings.go

@@ -349,3 +349,23 @@ var Quantize = settings.RegisterDurationSettingWithExplicitUnit(
 	time.Duration(metamorphic.ConstantWithTestRange("changefeed.resolved_timestamp.granularity", 1, 0, 10))*time.Second,
 	settings.DurationWithMinimum(0),
 )
+
+// MaxRetryBackoff is the maximum time a changefeed will backoff when in
+// a top-level retry loop, for example during rolling restarts.
+var MaxRetryBackoff = settings.RegisterDurationSettingWithExplicitUnit(
+	settings.ApplicationLevel,
+	"changefeed.max_retry_backoff",
+	"the maximum time a changefeed will backoff when retrying after a restart and how long between retries before backoff resets",
+	10*time.Minute, /* defaultValue */
+	settings.DurationInRange(1*time.Second, 1*time.Hour),
+)
+
+// RetryBackoffReset is the time between changefeed retries before the
+// backoff timer resets.
+var RetryBackoffReset = settings.RegisterDurationSettingWithExplicitUnit(
+	settings.ApplicationLevel,
+	"changefeed.retry_backoff_reset",
+	"the time between changefeed retries before the backoff timer resets",
+	10*time.Minute, /* defaultValue */
+	settings.DurationInRange(1*time.Second, 1*time.Hour),
+)

pkg/ccl/changefeedccl/retry.go

@@ -18,11 +18,11 @@ var useFastRetry = envutil.EnvOrDefaultBool(
 	"COCKROACH_CHANGEFEED_TESTING_FAST_RETRY", false)
 
 // getRetry returns retry object for changefeed.
-func getRetry(ctx context.Context) Retry {
+func getRetry(ctx context.Context, maxBackoff, backoffReset time.Duration) Retry {
 	opts := retry.Options{
 		InitialBackoff: 1 * time.Second,
 		Multiplier:     2,
-		MaxBackoff:     1 * time.Minute,
+		MaxBackoff:     maxBackoff,
 	}
 
 	if useFastRetry {
@@ -33,7 +33,8 @@ func getRetry(ctx context.Context) Retry {
 		}
 	}
 
-	return Retry{Retry: retry.StartWithCtx(ctx, opts)}
+	return Retry{Retry: retry.StartWithCtx(ctx, opts),
+		resetRetryAfter: backoffReset}
 }
 
 func testingUseFastRetry() func() {
@@ -43,16 +44,15 @@ func testingUseFastRetry() func() {
 	}
 }
 
-// reset retry state after changefeed ran for that much time
-// without errors.
-const resetRetryAfter = 10 * time.Minute
-
 // Retry is a thin wrapper around retry.Retry which
 // resets retry state if changefeed been running for sufficiently
 // long time.
 type Retry struct {
 	retry.Retry
 	lastRetry time.Time
+	// reset retry state after changefeed ran for that much time
+	// without errors.
+	resetRetryAfter time.Duration
 }
 
 // Next returns whether the retry loop should continue, and blocks for the
@@ -63,7 +63,7 @@ func (r *Retry) Next() bool {
 	defer func() {
 		r.lastRetry = timeutil.Now()
 	}()
-	if timeutil.Since(r.lastRetry) > resetRetryAfter {
+	if timeutil.Since(r.lastRetry) > r.resetRetryAfter {
 		r.Reset()
 	}
 	return r.Retry.Next()

pkg/cmd/roachtest/tests/s3_microceph.go

@@ -10,11 +10,14 @@ import (
 	"fmt"
 	"net/url"
 	"path/filepath"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/cloud/amazon"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/cluster"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/test"
+	"github.com/cockroachdb/cockroach/pkg/roachprod/install"
+	"github.com/cockroachdb/cockroach/pkg/util/retry"
 )
 
 // cephDisksScript creates 3 4GB loop devices, e.g. virtual block devices that allows
@@ -124,7 +127,9 @@ func (m cephManager) install(ctx context.Context) {
 		rgwCmd = rgwCmd + ` --ssl-certificate="$(base64 -w0 certs/node.crt)" --ssl-private-key="$(base64 -w0 certs/node.key)"`
 	}
 	m.run(ctx, `starting object gateway`, rgwCmd)
-
+	// We have seen occasional failures in creating users, so we
+	// wait until a read only request succeeds before proceeding.
+	m.checkRGW(ctx)
 	m.run(ctx, `creating backup user`,
 		`sudo radosgw-admin user create --uid=backup --display-name=backup`)
 	m.run(ctx, `add keys to the user`,
@@ -166,3 +171,20 @@ func (m cephManager) run(ctx context.Context, msg string, cmd ...string) {
 	m.c.Run(ctx, option.WithNodes(m.cephNodes), cmd...)
 	m.t.Status(msg, " done")
 }
+
+// checkRGW verifies that the Ceph Object Gateway is up.
+func (m cephManager) checkRGW(ctx context.Context) {
+	m.t.Status("waiting for Ceph Object Gateway...")
+	if err := m.c.RunE(ctx,
+		option.WithNodes(m.cephNodes).
+			WithRetryOpts(retry.Options{
+				InitialBackoff: 2 * time.Second,
+				MaxBackoff:     30 * time.Second,
+				MaxRetries:     10,
+			}).
+			WithShouldRetryFn(func(*install.RunResultDetails) bool { return true }),
+		`sudo radosgw-admin user list`,
+	); err != nil {
+		m.t.Error("failed to connect to Ceph Object Gateway", err)
+	}
+}

pkg/kv/kvserver/replica_rangefeed.go

@@ -90,7 +90,7 @@ var RangefeedUseBufferedSender = settings.RegisterBoolSetting(
 	"kv.rangefeed.buffered_sender.enabled",
 	"use buffered sender for all range feeds instead of buffering events "+
 		"separately per client per range",
-	metamorphic.ConstantWithTestBool("kv.rangefeed.buffered_sender.enabled", false),
+	metamorphic.ConstantWithTestBool("kv.rangefeed.buffered_sender.enabled", true),
 )
 
 func init() {