crosscluster/logical: auto-update config on external conn changes

Release note (sql change): logical cluster now uses
external connection and automatically updates its
configuration when that connection changes.

Epic: none

Author: edwardguo-crl

pkg/crosscluster/logical/BUILD.bazel

@@ -34,6 +34,7 @@ go_library(
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/utilccl",
         "//pkg/cloud",
+        "//pkg/cloud/externalconn",
         "//pkg/clusterversion",
         "//pkg/crosscluster",
         "//pkg/crosscluster/physical",

pkg/crosscluster/logical/logical_replication_job.go

@@ -8,9 +8,11 @@ package logical
 import (
 	"context"
 	"fmt"
+	"net/url"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/physical"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationutils"
@@ -279,7 +281,9 @@ func (r *logicalReplicationResumer) ingest(
 		stopReplanner()
 	}()
 
+	confPoller := make(chan struct{})
 	execPlan := func(ctx context.Context) error {
+		defer close(confPoller)
 		rh := rowHandler{
 			replicatedTimeAtStart: replicatedTimeAtStart,
 			frontier:              frontier,
@@ -322,6 +326,36 @@ func (r *logicalReplicationResumer) ingest(
 		return err
 	}
 
+	refreshConn := func(ctx context.Context) error {
+		ingestionJob := r.job
+		details := ingestionJob.Details().(jobspb.LogicalReplicationDetails)
+		resolvedDest, err := resolveDest(ctx, jobExecCtx.ExecCfg(), details.SourceClusterConnUri)
+		if err != nil {
+			return err
+		}
+		pollingInterval := 2 * time.Minute
+		if knobs := jobExecCtx.ExecCfg().StreamingTestingKnobs; knobs != nil && knobs.ExternalConnectionPollingInterval != nil {
+			pollingInterval = *knobs.ExternalConnectionPollingInterval
+		}
+		t := time.NewTicker(pollingInterval)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-confPoller:
+				return nil
+			case <-t.C:
+				newDest, err := reloadDest(ctx, ingestionJob.ID(), jobExecCtx.ExecCfg())
+				if err != nil {
+					log.Warningf(ctx, "failed to check for updated configuration: %v", err)
+				} else if newDest != resolvedDest {
+					return errors.Mark(errors.Newf("replan due to detail change: old=%s, new=%s", resolvedDest, newDest), sql.ErrPlanChanged)
+				}
+			}
+		}
+	}
+
 	defer func() {
 		if l := payload.MetricsLabel; l != "" {
 			metrics.LabeledScanningRanges.Update(map[string]string{"label": l}, 0)
@@ -331,7 +365,7 @@ func (r *logicalReplicationResumer) ingest(
 		metrics.CatchupRanges.Update(0)
 	}()
 
-	err = ctxgroup.GoAndWait(ctx, execPlan, replanner, startHeartbeat)
+	err = ctxgroup.GoAndWait(ctx, execPlan, replanner, startHeartbeat, refreshConn)
 	if errors.Is(err, sql.ErrPlanChanged) {
 		metrics.ReplanCount.Inc(1)
 	}
@@ -890,6 +924,7 @@ func (r *logicalReplicationResumer) ingestWithRetries(
 	ro := getRetryPolicy(execCtx.ExecCfg().StreamingTestingKnobs)
 	var err error
 	var lastReplicatedTime hlc.Timestamp
+
 	for retrier := retry.Start(ro); retrier.Next(); {
 		err = r.ingest(ctx, execCtx)
 		if err == nil {
@@ -1054,6 +1089,35 @@ func getRetryPolicy(knobs *sql.StreamingTestingKnobs) retry.Options {
 	}
 }
 
+func resolveDest(
+	ctx context.Context, execCfg *sql.ExecutorConfig, sourceURI string,
+) (string, error) {
+	u, err := url.Parse(sourceURI)
+	if err != nil {
+		return "", err
+	}
+
+	resolved := ""
+	err = execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		conn, err := externalconn.LoadExternalConnection(ctx, u.Host, txn)
+		if err != nil {
+			return err
+		}
+		resolved = conn.UnredactedConnectionStatement()
+		return nil
+	})
+	return resolved, err
+}
+
+func reloadDest(ctx context.Context, id jobspb.JobID, execCfg *sql.ExecutorConfig) (string, error) {
+	reloadedJob, err := execCfg.JobRegistry.LoadJob(ctx, id)
+	if err != nil {
+		return "", err
+	}
+	newDetails := reloadedJob.Details().(jobspb.LogicalReplicationDetails)
+	return resolveDest(ctx, execCfg, newDetails.SourceClusterConnUri)
+}
+
 func init() {
 	m := MakeMetrics(base.DefaultHistogramWindowInterval())
 	jobs.RegisterConstructor(

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -2443,6 +2443,102 @@ func TestLogicalReplicationGatewayRoute(t *testing.T) {
 	require.Empty(t, progress.Details.(*jobspb.Progress_LogicalReplication).LogicalReplication.PartitionConnUris)
 }
 
+// TestAlterExternalConnection tests that logical replication streams can
+// dynamically switch between different source nodes when the external
+// connection URI is updated. It verifies that data continues to replicate
+// correctly after the connection change.
+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
+
+	clusterArgs := base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			DefaultTestTenant: base.TestControlsTenantsExplicitly,
+			Knobs: base.TestingKnobs{
+				JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
+				Streaming: &sql.StreamingTestingKnobs{
+					ExternalConnectionPollingInterval: &pollingInterval,
+				},
+			},
+		},
+	}
+
+	activeLogicalSessionSQL := "SELECT count(*) > 0 FROM crdb_internal.node_sessions WHERE application_name like '$ internal repstream job id=%' AND status='ACTIVE'"
+	countLogicalSessionSQL := "SELECT count(*) FROM crdb_internal.node_sessions WHERE application_name like '$ internal repstream job id=%' AND status='ACTIVE'"
+	server, node0, runners, dbNames := setupServerWithNumDBs(t, ctx, clusterArgs, 3, 2)
+	defer server.Stopper().Stop(ctx)
+
+	dbA := runners[0]
+	dbB := runners[1]
+
+	dbANode0URL, cleanup := node0.PGUrl(t, serverutils.DBName(dbNames[0]))
+	defer cleanup()
+	dbANode0 := sqlutils.MakeSQLRunner(node0.SQLConn(t, serverutils.DBName(dbNames[0])))
+	node1 := server.Server(1).ApplicationLayer()
+	dbANode1URL, cleanup := node1.PGUrl(t, serverutils.DBName(dbNames[0]))
+	defer cleanup()
+	dbANode1 := sqlutils.MakeSQLRunner(node1.SQLConn(t, serverutils.DBName(dbNames[0])))
+
+	q0 := dbANode0URL.Query()
+	q0.Set(streamclient.RoutingModeKey, string(streamclient.RoutingModeGateway))
+	dbANode0URL.RawQuery = q0.Encode()
+
+	q1 := dbANode1URL.Query()
+	q1.Set(streamclient.RoutingModeKey, string(streamclient.RoutingModeGateway))
+	dbANode1URL.RawQuery = q1.Encode()
+
+	// We want to make sure operations for cluster B is on seperate node from cluster A.
+	node2 := server.Server(2).ApplicationLayer()
+	dbBNode2 := sqlutils.MakeSQLRunner(node2.SQLConn(t, serverutils.DBName(dbNames[1])))
+
+	require.NotEqual(t, dbANode0URL.String(), dbANode1URL.String())
+
+	externalConnName := "test_conn"
+	dbBNode2.Exec(t, fmt.Sprintf("CREATE EXTERNAL CONNECTION '%s' AS '%s'", externalConnName, dbANode0URL.String()))
+
+	var jobID jobspb.JobID
+	dbBNode2.QueryRow(t, fmt.Sprintf(
+		"CREATE LOGICAL REPLICATION STREAM FROM TABLE tab ON 'external://%s' INTO TABLE tab",
+		externalConnName)).Scan(&jobID)
+
+	dbANode0.Exec(t, "INSERT INTO tab VALUES (1, 'via_node_0')")
+
+	now := node0.Clock().Now()
+	WaitUntilReplicatedTime(t, now, dbB, jobID)
+
+	dbANode0.CheckQueryResults(t,
+		activeLogicalSessionSQL,
+		[][]string{{"true"}})
+	dbANode1.CheckQueryResults(t,
+		countLogicalSessionSQL,
+		[][]string{{"0"}})
+
+	dbBNode2.CheckQueryResults(t, "SELECT * FROM tab WHERE pk = 1", [][]string{
+		{"1", "via_node_0"},
+	})
+
+	dbBNode2.Exec(t, fmt.Sprintf("ALTER EXTERNAL CONNECTION '%s' AS '%s'", externalConnName, dbANode1URL.String()))
+	dbANode1.CheckQueryResultsRetry(t,
+		activeLogicalSessionSQL,
+		[][]string{{"true"}})
+	dbANode0.CheckQueryResultsRetry(t,
+		countLogicalSessionSQL,
+		[][]string{{"0"}})
+
+	dbA.Exec(t, "INSERT INTO tab VALUES (2, 'via_node_1')")
+	now = node0.Clock().Now()
+	WaitUntilReplicatedTime(t, now, dbB, jobID)
+
+	dbBNode2.CheckQueryResults(t, "SELECT * FROM tab WHERE pk = 2", [][]string{
+		{"2", "via_node_1"},
+	})
+}
+
 func TestMismatchColIDs(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	skip.UnderDeadlock(t)

pkg/sql/exec_util.go

@@ -2048,6 +2048,10 @@ 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
 
 	AfterRetryIteration func(err error)