Rewrite force replication's verify target for pre-1.22.2 servers

charts/s2s-proxy/files/default.yaml

@@ -35,6 +35,11 @@ clusterConnections:
           caServerName: "" # required unless skipCAVerification: true; must match the remote cert's SAN/CN
     # This should be the routable address
     replicationEndpoint: "my-s2s-proxy.svc.cluster.local:9233"
+    # If your local Temporal server is older than v1.22.2, uncomment this. Those versions dial the
+    # force replication workflow's TargetClusterEndpoint argument directly instead of resolving the
+    # target cluster by name, and the address the cloud sends is not routable from your network.
+    # With this on, the proxy rewrites that argument to replicationEndpoint above.
+    # forceReplicationEndpointOverride: true
     # This is what the remote cluster is allowed to access on your local cluster
     aclPolicy:
       allowedMethods:

config/cluster_conn_config.go

@@ -8,19 +8,24 @@ import (
 // Looking for examples? Check ./develop/sample-cluster-conn-config.yaml
 type (
 	ClusterConnConfig struct {
-		Name                         string              `yaml:"name"`
-		Local                        ClusterDefinition   `yaml:"local"`
-		Remote                       ClusterDefinition   `yaml:"remote"`
-		ReplicationEndpoint          string              `yaml:"replicationEndpoint"`
-		FVITranslation               IntMapping          `yaml:"failoverVersionIncrementTranslation"`
-		ACLPolicy                    *ACLPolicy          `yaml:"aclPolicy"`
-		NamespaceTranslation         StringTranslator    `yaml:"namespaceTranslation"`
-		SearchAttributeTranslation   SATranslationConfig `yaml:"searchAttributeTranslation"`
-		CustomSearchAttributeAliases CustomSAAliasConfig `yaml:"customSearchAttributeAliases"`
-		RemoteClusterHealthCheck     HealthCheckConfig   `yaml:"remoteClusterHealthCheck"`
-		LocalClusterHealthCheck      HealthCheckConfig   `yaml:"localClusterHealthCheck"`
-		ShardCountConfig             ShardCountConfig    `yaml:"shardCount"`
-		MemberlistConfig             *MemberlistConfig   `yaml:"memberlist"`
+		Name                string            `yaml:"name"`
+		Local               ClusterDefinition `yaml:"local"`
+		Remote              ClusterDefinition `yaml:"remote"`
+		ReplicationEndpoint string            `yaml:"replicationEndpoint"`
+		// ForceReplicationEndpointOverride rewrites the force replication workflow's
+		// TargetClusterEndpoint argument to ReplicationEndpoint on inbound StartWorkflowExecution
+		// requests. Needed for local Temporal servers older than v1.22.2, which dial that address
+		// verbatim instead of resolving TargetClusterName. Requires replicationEndpoint to be set.
+		ForceReplicationEndpointOverride bool                `yaml:"forceReplicationEndpointOverride"`
+		FVITranslation                   IntMapping          `yaml:"failoverVersionIncrementTranslation"`
+		ACLPolicy                        *ACLPolicy          `yaml:"aclPolicy"`
+		NamespaceTranslation             StringTranslator    `yaml:"namespaceTranslation"`
+		SearchAttributeTranslation       SATranslationConfig `yaml:"searchAttributeTranslation"`
+		CustomSearchAttributeAliases     CustomSAAliasConfig `yaml:"customSearchAttributeAliases"`
+		RemoteClusterHealthCheck         HealthCheckConfig   `yaml:"remoteClusterHealthCheck"`
+		LocalClusterHealthCheck          HealthCheckConfig   `yaml:"localClusterHealthCheck"`
+		ShardCountConfig                 ShardCountConfig    `yaml:"shardCount"`
+		MemberlistConfig                 *MemberlistConfig   `yaml:"memberlist"`
 	}
 	StringTranslator struct {
 		Mappings    []StringMapping `yaml:"mappings"`

config/config_test.go

@@ -97,6 +97,7 @@ func TestBasic(t *testing.T) {
 
 	cc := proxyConfig.ClusterConnections[0]
 	require.Equal(t, "127.0.0.1:9002", cc.ReplicationEndpoint)
+	require.True(t, cc.ForceReplicationEndpointOverride)
 	require.Equal(t, IntMapping{Local: 100, Remote: 1000000}, cc.FVITranslation)
 	require.NotNil(t, cc.ACLPolicy)
 	require.Contains(t, cc.ACLPolicy.AllowedMethods.AdminService, "AddOrUpdateRemoteCluster")
@@ -277,6 +278,8 @@ func TestDefaultChart(t *testing.T) {
 	require.Equal(t, ConnectionType("mux-client"), cc.Remote.ConnectionType)
 	require.Equal(t, "remote_proxy_service:8233", cc.Remote.MuxAddressInfo.ConnectionString)
 	require.Equal(t, "my-s2s-proxy.svc.cluster.local:9233", cc.ReplicationEndpoint)
+	// Opt-in: present but commented out in the chart default.
+	require.False(t, cc.ForceReplicationEndpointOverride)
 	require.False(t, cc.Remote.MuxAddressInfo.TLSConfig.IsEnabled())
 }
 

develop/config/sample-cluster-conn-config.yaml

@@ -49,6 +49,9 @@ clusterConnections:
       local: 100
       remote: 1000000
     replicationEndpoint: "127.0.0.1:9002"
+    # Rewrite force replication's verify target for local servers older than v1.22.2.
+    # Requires replicationEndpoint to be set.
+    forceReplicationEndpointOverride: true
     namespaceTranslation:
       mappings:
         - local: "localName"

interceptor/force_replication_translator.go

@@ -0,0 +1,115 @@
+package interceptor
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/common/api"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
+
+	"github.com/temporalio/s2s-proxy/metrics"
+)
+
+const (
+	// Registered by the server's service/worker/migration package.
+	forceReplicationWorkflowType = "force-replication"
+
+	// ForceReplicationParams carries no json tags, so the key is the Go field name verbatim.
+	targetClusterEndpointKey = "TargetClusterEndpoint"
+
+	payloadEncodingMetadataKey = "encoding"
+	jsonPlainEncoding          = "json/plain"
+
+	// WorkflowServicePrefix already ends with "/".
+	startWorkflowExecutionMethod = api.WorkflowServicePrefix + "StartWorkflowExecution"
+)
+
+type (
+	// frEndpointTranslator rewrites ForceReplicationParams.TargetClusterEndpoint in the
+	// StartWorkflowExecution request that kicks off the force replication workflow.
+	//
+	// Temporal servers older than v1.22.2 dial that address verbatim from the
+	// VerifyReplicationTasks activity, and the address the caller sends is the remote cluster's own
+	// address, which is generally not routable from the local cluster. Rewriting it to this proxy's
+	// replicationEndpoint is the same correction the proxy already applies to FrontendAddress in
+	// AddOrUpdateRemoteCluster; that one is a typed proto field, this one rides inside a
+	// workflow-args payload. TargetClusterName is left alone, so servers from v1.22.2 on (which
+	// prefer the name and resolve the address from their own cluster registry) are unaffected.
+	//
+	// Unlike the other translators this one does not use the reflection visitor: it type-asserts the
+	// request instead, which is also what keeps it inert on the stream path where MatchMethod is
+	// never consulted and every translator sees every message.
+	frEndpointTranslator struct {
+		logger              log.Logger
+		replicationEndpoint string
+	}
+)
+
+func NewForceReplicationEndpointTranslator(logger log.Logger, replicationEndpoint string) Translator {
+	return &frEndpointTranslator{
+		logger:              logger,
+		replicationEndpoint: replicationEndpoint,
+	}
+}
+
+func (t *frEndpointTranslator) Kind() string {
+	return metrics.ForceReplicationEndpointTranslationKind
+}
+
+func (t *frEndpointTranslator) MatchMethod(m string) bool {
+	return m == startWorkflowExecutionMethod
+}
+
+func (t *frEndpointTranslator) TranslateRequest(req any) (bool, error) {
+	// Type assert first: on the stream path MatchMethod is not consulted, so this is what keeps the
+	// translator inert there.
+	r, ok := req.(*workflowservice.StartWorkflowExecutionRequest)
+	if !ok || r.GetWorkflowType().GetName() != forceReplicationWorkflowType {
+		return false, nil
+	}
+
+	// Past this point the request is definitely a force replication start. Any failure to rewrite is
+	// reported as an error so it shows up on the translation_error metric under this Kind: a silent
+	// no-op here means a migration that hangs for a week with no signal.
+	payloads := r.GetInput().GetPayloads()
+	if len(payloads) == 0 {
+		return false, fmt.Errorf("%s request has no input payloads", forceReplicationWorkflowType)
+	}
+
+	payload := payloads[0]
+	if encoding := string(payload.GetMetadata()[payloadEncodingMetadataKey]); encoding != jsonPlainEncoding {
+		return false, fmt.Errorf("%s params have unsupported payload encoding %q, want %q",
+			forceReplicationWorkflowType, encoding, jsonPlainEncoding)
+	}
+
+	var params map[string]json.RawMessage
+	if err := json.Unmarshal(payload.GetData(), &params); err != nil {
+		return false, fmt.Errorf("failed to decode %s params: %w", forceReplicationWorkflowType, err)
+	}
+	if _, found := params[targetClusterEndpointKey]; !found {
+		return false, fmt.Errorf("%s params have no %s field", forceReplicationWorkflowType, targetClusterEndpointKey)
+	}
+
+	newEndpoint, err := json.Marshal(t.replicationEndpoint)
+	if err != nil {
+		return false, fmt.Errorf("failed to encode %s: %w", targetClusterEndpointKey, err)
+	}
+	params[targetClusterEndpointKey] = newEndpoint
+
+	data, err := json.Marshal(params)
+	if err != nil {
+		return false, fmt.Errorf("failed to re-encode %s params: %w", forceReplicationWorkflowType, err)
+	}
+
+	// Mutate in place so the payload's Metadata is preserved by construction.
+	payload.Data = data
+	t.logger.Info("Overwrote force replication target cluster endpoint",
+		tag.Address(t.replicationEndpoint))
+	return true, nil
+}
+
+func (t *frEndpointTranslator) TranslateResponse(any) (bool, error) {
+	return false, nil
+}