Merge cockroachdb#103142

103142: c2c: introduce automatic distsql replanning r=stevendanna a=msbutler

This patch introduces the ability for a replication job to periodically refresh its physical plan. Here's how replanning works:

- The replication consumer's dist sql planner proposes a new physical plan every 10 minutes, as determined by the new `replication_stream.replan_flow.frequency` setting.

-  it then computes how different the proposed plan is to the current plan. This patch computes difference as the number of _participating_ node changes (addition or removal) in the source and destination clusters as a fraction of the total number of participating nodes in both clusters in the previous plan.

- execute the new plan if this difference metric is greater than 0.1, the default of the new `repliaction_stream.replan_flow.threshold` setting.

In future work, we should consider replanning on data distribution changes, as outlined in cockroachdb#99164.

Fixes cockroachdb#102900

Release note: None

Co-authored-by: Michael Butler <[email protected]>

Author: craig[bot]

pkg/ccl/streamingccl/replicationtestutils/testutils.go

@@ -222,10 +222,7 @@ func waitForTenantPodsActive(
 	}, 10*time.Second)
 }
 
-func CreateTenantStreamingClusters(
-	ctx context.Context, t *testing.T, args TenantStreamingClustersArgs,
-) (*TenantStreamingClusters, func()) {
-
+func CreateServerArgs(args TenantStreamingClustersArgs) base.TestServerArgs {
 	if args.TestingKnobs != nil && args.TestingKnobs.DistSQLRetryPolicy == nil {
 		args.TestingKnobs.DistSQLRetryPolicy = &retry.Options{
 			InitialBackoff: time.Microsecond,
@@ -234,7 +231,7 @@ func CreateTenantStreamingClusters(
 			MaxRetries:     TestingMaxDistSQLRetries,
 		}
 	}
-	serverArgs := base.TestServerArgs{
+	return base.TestServerArgs{
 		// Test fails because it tries to set a cluster setting only accessible
 		// to system tenants. Tracked with #76378.
 		DefaultTestTenant: base.TestTenantDisabled,
@@ -252,6 +249,12 @@ func CreateTenantStreamingClusters(
 			},
 		},
 	}
+}
+
+func CreateTenantStreamingClusters(
+	ctx context.Context, t *testing.T, args TenantStreamingClustersArgs,
+) (*TenantStreamingClusters, func()) {
+	serverArgs := CreateServerArgs(args)
 
 	startTestCluster := func(
 		ctx context.Context,

pkg/ccl/streamingccl/streamingest/BUILD.bazel

@@ -88,6 +88,7 @@ go_test(
         "rangekey_batcher_test.go",
         "replication_random_client_test.go",
         "replication_stream_e2e_test.go",
+        "stream_ingestion_dist_test.go",
         "stream_ingestion_job_test.go",
         "stream_ingestion_processor_test.go",
     ],
@@ -129,6 +130,7 @@ go_test(
         "//pkg/sql/execinfra",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/isql",
+        "//pkg/sql/physicalplan",
         "//pkg/sql/sem/eval",
         "//pkg/sql/sem/tree",
         "//pkg/storage",

pkg/ccl/streamingccl/streamingest/metrics.go

@@ -128,6 +128,12 @@ var (
 		Measurement: "Ranges",
 		Unit:        metric.Unit_COUNT,
 	}
+	metaDistSQLReplanCount = metric.Metadata{
+		Name:        "replication.distsql_replan_count",
+		Help:        "Total number of dist sql replanning events",
+		Measurement: "Events",
+		Unit:        metric.Unit_COUNT,
+	}
 )
 
 // Metrics are for production monitoring of stream ingestion jobs.
@@ -138,6 +144,7 @@ type Metrics struct {
 	Flushes                     *metric.Counter
 	JobProgressUpdates          *metric.Counter
 	ResolvedEvents              *metric.Counter
+	ReplanCount                 *metric.Counter
 	FlushHistNanos              metric.IHistogram
 	CommitLatency               metric.IHistogram
 	AdmitLatency                metric.IHistogram
@@ -162,6 +169,7 @@ func MakeMetrics(histogramWindow time.Duration) metric.Struct {
 		Flushes:              metric.NewCounter(metaReplicationFlushes),
 		ResolvedEvents:       metric.NewCounter(metaReplicationResolvedEventsIngested),
 		JobProgressUpdates:   metric.NewCounter(metaJobProgressUpdates),
+		ReplanCount:          metric.NewCounter(metaDistSQLReplanCount),
 		FlushHistNanos: metric.NewHistogram(metric.HistogramOptions{
 			Metadata: metaReplicationFlushHistNanos,
 			Duration: histogramWindow,

pkg/ccl/streamingccl/streamingest/replication_stream_e2e_test.go

@@ -708,6 +708,87 @@ func TestTenantStreamingMultipleNodes(t *testing.T) {
 	require.Greater(t, len(clientAddresses), 1)
 }
 
+// TestStreamingAutoReplan asserts that if a new node can participate in the
+// replication job, it will trigger distSQL replanning.
+func TestStreamingAutoReplan(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	skip.UnderStressRace(t, "c2c multi node unit tests flake under stress race. see #106194")
+
+	ctx := context.Background()
+	args := replicationtestutils.DefaultTenantStreamingClustersArgs
+	args.SrcNumNodes = 1
+	args.DestNumNodes = 1
+
+	retryErrorChan := make(chan error)
+	turnOffReplanning := make(chan struct{})
+
+	// Track the number of unique addresses that we're connected to.
+	clientAddresses := make(map[string]struct{})
+	var addressesMu syncutil.Mutex
+	args.TestingKnobs = &sql.StreamingTestingKnobs{
+		BeforeClientSubscribe: func(addr string, token string, clientStartTime hlc.Timestamp) {
+			addressesMu.Lock()
+			defer addressesMu.Unlock()
+			clientAddresses[addr] = struct{}{}
+		},
+		AfterRetryIteration: func(err error) {
+			if err != nil {
+				retryErrorChan <- err
+				<-turnOffReplanning
+			}
+		},
+	}
+
+	c, cleanup := replicationtestutils.CreateTenantStreamingClusters(ctx, t, args)
+	defer cleanup()
+	serverutils.SetClusterSetting(t, c.DestCluster, "stream_replication.replan_flow_threshold", 0.1)
+	serverutils.SetClusterSetting(t, c.DestCluster, "stream_replication.replan_flow_frequency", time.Millisecond*500)
+
+	// Begin the job on a single source node.
+	producerJobID, ingestionJobID := c.StartStreamReplication(ctx)
+	jobutils.WaitForJobToRun(c.T, c.SrcSysSQL, jobspb.JobID(producerJobID))
+	jobutils.WaitForJobToRun(c.T, c.DestSysSQL, jobspb.JobID(ingestionJobID))
+
+	c.WaitUntilStartTimeReached(jobspb.JobID(ingestionJobID))
+	require.Equal(t, len(clientAddresses), 1)
+
+	// Add 2 source nodes to enable full replication.
+	c.SrcCluster.AddAndStartServer(c.T, replicationtestutils.CreateServerArgs(c.Args))
+	c.SrcCluster.AddAndStartServer(c.T, replicationtestutils.CreateServerArgs(c.Args))
+	require.NoError(t, c.SrcCluster.WaitForFullReplication())
+
+	replicationtestutils.CreateScatteredTable(t, c, 3)
+	require.NoError(t, c.SrcCluster.WaitForFullReplication())
+
+	// The ingestion job should eventually retry because it detects new nodes to add to the plan.
+	require.Error(t, <-retryErrorChan, sql.ErrPlanChanged)
+
+	// Prevent continuous replanning to reduce test runtime. dsp.PartitionSpans()
+	// on the src cluster may return a different set of src nodes that can
+	// participate in the replication job (especially under stress), so if we
+	// repeatedly replan the job, we will repeatedly restart the job, preventing
+	// job progress.
+	serverutils.SetClusterSetting(t, c.DestCluster, "stream_replication.replan_flow_threshold", 0)
+	serverutils.SetClusterSetting(t, c.DestCluster, "stream_replication.replan_flow_frequency", time.Minute*10)
+	close(turnOffReplanning)
+
+	cutoverTime := c.DestSysServer.Clock().Now()
+	c.WaitUntilReplicatedTime(cutoverTime, jobspb.JobID(ingestionJobID))
+
+	// After the node additions, multiple nodes should've been connected to. When
+	// this test is run under stress, however, dsp.PartitionSpans() on the src
+	// cluster does not always return multiple src nodes that can participate in
+	// the replication job, therefore, under stress, do not require that multiple
+	// nodes participate from the src cluster. This potentially occurs because cpu
+	// contention renders a test server "unhealthy". In general, running two
+	// multinode 2 clusters makes everything messy.
+	if !skip.Stress() {
+		require.Greater(t, len(clientAddresses), 1)
+	}
+}
+
 // TestTenantReplicationProtectedTimestampManagement tests the active protected
 // timestamps management on the destination tenant's keyspan.
 func TestTenantReplicationProtectedTimestampManagement(t *testing.T) {