replicationutils: move external conn loader to utility package

Prepping to use this in PCR.

Epic: none

Release note: none

Author: msbutler

pkg/crosscluster/logical/logical_replication_job.go

@@ -326,35 +326,13 @@ 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.Dev.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)
-				}
-			}
-		}
-	}
+	refreshConn := replicationutils.GetAlterConnectionChecker(
+		r.job.ID(),
+		uris[0].Serialize(),
+		geURIFromLoadedJobDetails,
+		execCfg,
+		confPoller,
+	)
 
 	defer func() {
 		if l := payload.MetricsLabel; l != "" {
@@ -1089,29 +1067,8 @@ func getRetryPolicy(knobs *sql.StreamingTestingKnobs) retry.Options {
 	}
 }
 
-func resolveDest(
-	ctx context.Context, execCfg *sql.ExecutorConfig, sourceURI string,
-) (string, error) {
-	configUri, err := streamclient.ParseConfigUri(sourceURI)
-	if err != nil {
-		return "", err
-	}
-
-	clusterUri, err := configUri.AsClusterUri(ctx, execCfg.InternalDB)
-	if err != nil {
-		return "", err
-	}
-
-	return clusterUri.Serialize(), nil
-}
-
-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 geURIFromLoadedJobDetails(details jobspb.Details) string {
+	return details.(jobspb.LogicalReplicationDetails).SourceClusterConnUri
 }
 
 func init() {

pkg/crosscluster/replicationutils/BUILD.bazel

@@ -3,12 +3,14 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "replicationutils",
     srcs = [
+        "connection_checker.go",
         "stats.go",
         "utils.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/crosscluster/replicationutils",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/crosscluster/streamclient",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/kv/kvpb",
@@ -33,6 +35,7 @@ go_library(
         "//pkg/testutils/fingerprintutils",
         "//pkg/util/ctxgroup",
         "//pkg/util/hlc",
+        "//pkg/util/log",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",

pkg/crosscluster/replicationutils/connection_checker.go

@@ -0,0 +1,74 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package replicationutils
+
+import (
+	"context"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/errors"
+)
+
+// GetAlterConnectionChecker returns a function that will poll for an altered
+// external connection. The initial URI passed to this function should be the
+// same one passed to subsequent client creation calls.
+func GetAlterConnectionChecker(
+	id jobspb.JobID,
+	initialURI string,
+	uriGetter URIGetter,
+	execCfg *sql.ExecutorConfig,
+	stopper chan struct{},
+) func(ctx context.Context) error {
+	return func(ctx context.Context) error {
+		pollingInterval := 2 * time.Minute
+		if knobs := 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 <-stopper:
+				return nil
+			case <-t.C:
+				reloadedJob, err := execCfg.JobRegistry.LoadJob(ctx, id)
+				if err != nil {
+					return err
+				}
+				newURI, err := resolveURI(ctx, execCfg, uriGetter(reloadedJob.Details()))
+				if err != nil {
+					log.Dev.Warningf(ctx, "failed to load uri: %v", err)
+				} else if newURI != initialURI {
+					return errors.Mark(errors.Newf("uri has been updated: old=%s, new=%s", errors.Redact(initialURI), errors.Redact(newURI)), sql.ErrPlanChanged)
+				}
+			}
+		}
+	}
+}
+
+type URIGetter func(details jobspb.Details) string
+
+func resolveURI(
+	ctx context.Context, execCfg *sql.ExecutorConfig, sourceURI string,
+) (string, error) {
+	configUri, err := streamclient.ParseConfigUri(sourceURI)
+	if err != nil {
+		return "", err
+	}
+
+	clusterUri, err := configUri.AsClusterUri(ctx, execCfg.InternalDB)
+	if err != nil {
+		return "", err
+	}
+
+	return clusterUri.Serialize(), nil
+}