crosscluster/physical: add alter connection watcher

PCR now automatically updates the uri used if it's apart of an external
connection that is altered.

Informs #153471

Release note: none

Author: msbutler

pkg/crosscluster/physical/replication_stream_e2e_test.go

@@ -1800,3 +1800,62 @@ func splitPrimaryKeyIndexSpan(
 	require.NoError(t, db.AdminSplit(ctx, pkStartKey, hlc.MaxTimestamp))
 	require.NoError(t, db.AdminSplit(ctx, pkEndKey, hlc.MaxTimestamp))
 }
+
+func TestAlterExternalConnection(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	skip.UnderDeadlock(t)
+	skip.UnderRace(t)
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	pollingInterval := 100 * time.Millisecond
+
+	var alreadyReplanned atomic.Int32
+
+	args := replicationtestutils.DefaultTenantStreamingClustersArgs
+	args.TestingKnobs = &sql.StreamingTestingKnobs{
+		ExternalConnectionPollingInterval: &pollingInterval,
+		AfterRetryIteration: func(err error) {
+			alreadyReplanned.Add(1)
+		},
+	}
+
+	c, cleanup := replicationtestutils.CreateTenantStreamingClusters(ctx, t, args)
+	defer cleanup()
+
+	externalConnection := "replication-source-addr"
+	ogConnection := c.SrcURL.String()
+	c.DestSysSQL.Exec(c.T, fmt.Sprintf(`CREATE EXTERNAL CONNECTION "%s" AS "%s"`,
+		externalConnection, ogConnection))
+	c.DestSysSQL.Exec(c.T, c.BuildCreateTenantQuery(externalConnection))
+	streamProducerJobID, ingestionJobID := replicationtestutils.GetStreamJobIds(c.T, ctx, c.DestSysSQL, c.Args.DestTenantName)
+
+	jobutils.WaitForJobToRun(c.T, c.SrcSysSQL, jobspb.JobID(streamProducerJobID))
+	jobutils.WaitForJobToRun(c.T, c.DestSysSQL, jobspb.JobID(ingestionJobID))
+	c.WaitUntilStartTimeReached(jobspb.JobID(ingestionJobID))
+
+	// Alter the external connection to break the stream, to test that pcr watches for new connection changes
+	c.SrcSysSQL.Exec(t, fmt.Sprintf("CREATE USER %s", username.TestUser))
+	srcAppURL, cleanupSinkCert := pgurlutils.PGUrl(t, c.SrcSysServer.AdvSQLAddr(), t.Name(), url.User(username.TestUser))
+	defer cleanupSinkCert()
+
+	beforeReplanCount := alreadyReplanned.Load()
+	c.DestSysSQL.Exec(c.T, fmt.Sprintf(`ALTER EXTERNAL CONNECTION "%s" AS "%s"`,
+		externalConnection, &srcAppURL))
+
+	testutils.SucceedsSoon(t, func() error {
+		if alreadyReplanned.Load() <= beforeReplanCount+2 {
+			return errors.New("not yet replanned twice")
+		}
+		return nil
+	})
+
+	// Alter the external connection to fix the stream, and ensure replication resumes
+	c.DestSysSQL.Exec(c.T, fmt.Sprintf(`ALTER EXTERNAL CONNECTION "%s" AS "%s"`,
+		externalConnection, ogConnection))
+	c.DestSysSQL.Exec(c.T, fmt.Sprintf("RESUME JOB %d", ingestionJobID))
+	jobutils.WaitForJobToRun(c.T, c.DestSysSQL, jobspb.JobID(ingestionJobID))
+	// ensure the stream advances
+	srcTime := c.SrcCluster.Server(0).Clock().Now()
+	c.WaitUntilReplicatedTime(srcTime, jobspb.JobID(ingestionJobID))
+}

pkg/crosscluster/physical/stream_ingestion_dist.go

@@ -96,12 +96,16 @@ func startDistIngestion(
 		updateStatus(ctx, ingestionJob, jobspb.InitializingReplication, msg)
 	}
 
-	client, err := connectToActiveClient(ctx, ingestionJob, execCtx.ExecCfg().InternalDB,
-		streamclient.WithStreamID(streamID))
+	clusterUris, err := getClusterUris(ctx, ingestionJob, execCtx.ExecCfg().InternalDB)
+	if err != nil {
+		return err
+	}
+	client, err := streamclient.GetFirstActiveClient(ctx, clusterUris, execCtx.ExecCfg().InternalDB, streamclient.WithStreamID(streamID))
 	if err != nil {
 		return err
 	}
 	defer closeAndLog(ctx, client)
+
 	if err := waitUntilProducerActive(ctx, client, streamID, heartbeatTimestamp, ingestionJob.ID()); err != nil {
 		return err
 	}
@@ -197,11 +201,23 @@ func startDistIngestion(
 		}
 		return ingestor.ingestSpanConfigs(ctx, details.SourceTenantName)
 	}
+
+	refreshConnStopper := make(chan struct{})
+
+	refreshConn := replicationutils.GetAlterConnectionChecker(
+		ingestionJob.ID(),
+		clusterUris[0].Serialize(),
+		geURIFromIngestionJobDetails,
+		execCtx.ExecCfg(),
+		refreshConnStopper,
+	)
+
 	execInitialPlan := func(ctx context.Context) error {
 		defer func() {
 			stopReplanner()
 			close(tracingAggCh)
 			close(spanConfigIngestStopper)
+			close(refreshConnStopper)
 		}()
 		ctx = logtags.AddTag(ctx, "stream-ingest-distsql", nil)
 
@@ -273,13 +289,17 @@ func startDistIngestion(
 		return err
 	}
 
-	err = ctxgroup.GoAndWait(ctx, execInitialPlan, replanner, tracingAggLoop, streamSpanConfigs)
+	err = ctxgroup.GoAndWait(ctx, execInitialPlan, replanner, tracingAggLoop, streamSpanConfigs, refreshConn)
 	if errors.Is(err, sql.ErrPlanChanged) {
 		execCtx.ExecCfg().JobRegistry.MetricsStruct().StreamIngest.(*Metrics).ReplanCount.Inc(1)
 	}
 	return err
 }
 
+func geURIFromIngestionJobDetails(details jobspb.Details) string {
+	return details.(jobspb.StreamIngestionDetails).SourceClusterConnUri
+}
+
 func sortSpans(partitions []streamclient.PartitionInfo) roachpb.Spans {
 	spansToSort := make(roachpb.Spans, 0)
 	for i := range partitions {

pkg/sql/exec_util.go

@@ -2160,8 +2160,6 @@ type StreamingTestingKnobs struct {
 	// for whether the job record is updated on a progress update.
 	CutoverProgressShouldUpdate func() bool
 
-	//ExternalConnectionPollingInterval override the LDR alter
-	// connection polling frequency.
 	ExternalConnectionPollingInterval *time.Duration
 
 	DistSQLRetryPolicy *retry.Options