From 74ca54bb9a564dfe0dd1b3fa7f8f670bf19b1f3f Mon Sep 17 00:00:00 2001 From: JH Date: Sun, 23 Aug 2026 21:58:57 -0700 Subject: [PATCH] Rewrite force replication's verify target for pre-1.22.2 servers Temporal servers older than v1.22.2 ignore TargetClusterName in the VerifyReplicationTasks activity and dial TargetClusterEndpoint verbatim. The caller fills that with the remote cluster's own address, which is generally not routable from the local cluster, so verification never completes and the migration stalls while replication itself succeeds. Add an opt-in translator that rewrites that one field to this proxy's replicationEndpoint, which is the same correction already applied 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 are unaffected whether or not the flag stays set. Scoped narrowly: inbound only, StartWorkflowExecution only, and only when the workflow type is force-replication. Enabled by a new forceReplicationEndpointOverride key, which also requires replicationEndpoint to be set. Co-Authored-By: Claude Opus 5 --- charts/s2s-proxy/files/default.yaml | 5 + config/cluster_conn_config.go | 31 +- config/config_test.go | 3 + .../config/sample-cluster-conn-config.yaml | 3 + interceptor/force_replication_translator.go | 115 +++++++ .../force_replication_translator_test.go | 305 ++++++++++++++++++ metrics/prometheus_defs.go | 5 +- proxy/cluster_connection.go | 22 +- 8 files changed, 469 insertions(+), 20 deletions(-) create mode 100644 interceptor/force_replication_translator.go create mode 100644 interceptor/force_replication_translator_test.go diff --git a/charts/s2s-proxy/files/default.yaml b/charts/s2s-proxy/files/default.yaml index 832b8c76..325136fe 100644 --- a/charts/s2s-proxy/files/default.yaml +++ b/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: diff --git a/config/cluster_conn_config.go b/config/cluster_conn_config.go index 1855a836..28621a4b 100644 --- a/config/cluster_conn_config.go +++ b/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"` diff --git a/config/config_test.go b/config/config_test.go index f42ce17d..764c00b6 100644 --- a/config/config_test.go +++ b/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()) } diff --git a/develop/config/sample-cluster-conn-config.yaml b/develop/config/sample-cluster-conn-config.yaml index 767d91c1..df9df372 100644 --- a/develop/config/sample-cluster-conn-config.yaml +++ b/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" diff --git a/interceptor/force_replication_translator.go b/interceptor/force_replication_translator.go new file mode 100644 index 00000000..93e7d8ec --- /dev/null +++ b/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(), ¶ms); 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 +} diff --git a/interceptor/force_replication_translator_test.go b/interceptor/force_replication_translator_test.go new file mode 100644 index 00000000..b86aa504 --- /dev/null +++ b/interceptor/force_replication_translator_test.go @@ -0,0 +1,305 @@ +package interceptor + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/api/common/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/common/api" + "go.temporal.io/server/common/log" + "google.golang.org/grpc" +) + +const ( + // proxyEndpoint is what the proxy writes into the payload. + proxyEndpoint = "my-s2s-proxy.svc.cluster.local:9233" + // cloudEndpoint is what the control plane sends and the local server cannot reach. + cloudEndpoint = "admin.example-cell.cluster.tmprl.cloud:7233" + + // forceReplicationParamsJSON is the shape the control plane sends: the SDK's default json/plain + // encoding of migration.ForceReplicationParams, whose fields carry no json tags. + forceReplicationParamsJSON = `{ + "Namespace": "my-namespace", + "Query": "", + "ConcurrentActivityCount": 4, + "OverallRps": 10, + "EnableVerification": true, + "TargetClusterEndpoint": "` + cloudEndpoint + `", + "TargetClusterName": "example-cell" + }` + // rewrittenParamsJSON is forceReplicationParamsJSON with only TargetClusterEndpoint replaced. + rewrittenParamsJSON = `{ + "Namespace": "my-namespace", + "Query": "", + "ConcurrentActivityCount": 4, + "OverallRps": 10, + "EnableVerification": true, + "TargetClusterEndpoint": "` + proxyEndpoint + `", + "TargetClusterName": "example-cell" + }` +) + +func payloadsWith(encoding string, data string) *common.Payloads { + return &common.Payloads{ + Payloads: []*common.Payload{ + { + Metadata: map[string][]byte{ + "encoding": []byte(encoding), + // A key we do not own, to prove Metadata survives the rewrite. + "someOtherKey": []byte("someOtherValue"), + }, + Data: []byte(data), + }, + }, + } +} + +func startWorkflowReq(workflowType string, input *common.Payloads) *workflowservice.StartWorkflowExecutionRequest { + return &workflowservice.StartWorkflowExecutionRequest{ + Namespace: "my-namespace", + WorkflowId: "force-replication-my-namespace", + WorkflowType: &common.WorkflowType{Name: workflowType}, + Input: input, + } +} + +func forceReplicationReq() *workflowservice.StartWorkflowExecutionRequest { + return startWorkflowReq(forceReplicationWorkflowType, payloadsWith(jsonPlainEncoding, forceReplicationParamsJSON)) +} + +// payloadData returns the bytes of the first input payload, or nil if there is none. +func payloadData(req any) []byte { + r, ok := req.(*workflowservice.StartWorkflowExecutionRequest) + if !ok { + return nil + } + payloads := r.GetInput().GetPayloads() + if len(payloads) == 0 { + return nil + } + return payloads[0].GetData() +} + +func TestForceReplicationEndpointTranslator_TranslateRequest(t *testing.T) { + cases := []struct { + name string + // req is a func so every case gets a fresh fixture. + req func() any + wantChanged bool + wantErr bool + // wantJSON, when set, is compared to the payload data after translation. When it is empty + // the payload must come out byte-for-byte unchanged. + wantJSON string + }{ + { + name: "rewrites force replication endpoint", + req: func() any { return forceReplicationReq() }, + wantChanged: true, + wantJSON: rewrittenParamsJSON, + }, + { + name: "ignores other workflow types", + req: func() any { + return startWorkflowReq("some-other-workflow", payloadsWith(jsonPlainEncoding, forceReplicationParamsJSON)) + }, + }, + { + name: "ignores other request types", + req: func() any { + return &workflowservice.DescribeWorkflowExecutionRequest{Namespace: "my-namespace"} + }, + }, + { + name: "ignores untyped nil", + req: func() any { return nil }, + wantErr: false, + }, + { + name: "ignores typed nil request", + req: func() any { + var r *workflowservice.StartWorkflowExecutionRequest + return r + }, + }, + { + name: "errors on nil input", + req: func() any { + return startWorkflowReq(forceReplicationWorkflowType, nil) + }, + wantErr: true, + }, + { + name: "errors on empty payloads", + req: func() any { + return startWorkflowReq(forceReplicationWorkflowType, &common.Payloads{}) + }, + wantErr: true, + }, + { + name: "errors on nil payload", + req: func() any { + return startWorkflowReq(forceReplicationWorkflowType, &common.Payloads{Payloads: []*common.Payload{nil}}) + }, + wantErr: true, + }, + { + name: "errors on non-json encoding", + req: func() any { + return startWorkflowReq(forceReplicationWorkflowType, payloadsWith("binary/encrypted", "not json at all")) + }, + wantErr: true, + }, + { + name: "errors on unparseable json", + req: func() any { + return startWorkflowReq(forceReplicationWorkflowType, payloadsWith(jsonPlainEncoding, "{not json")) + }, + wantErr: true, + }, + { + name: "errors when the endpoint field is absent", + req: func() any { + return startWorkflowReq(forceReplicationWorkflowType, payloadsWith(jsonPlainEncoding, `{"TargetClusterName":"example-cell"}`)) + }, + wantErr: true, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + tr := NewForceReplicationEndpointTranslator(log.NewTestLogger(), proxyEndpoint) + req := c.req() + before := append([]byte(nil), payloadData(req)...) + + changed, err := tr.TranslateRequest(req) + + if c.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + require.Equal(t, c.wantChanged, changed) + + if c.wantJSON != "" { + require.JSONEq(t, c.wantJSON, string(payloadData(req))) + } else { + // Nothing was rewritten, including on the error paths. + require.Equal(t, before, append([]byte(nil), payloadData(req)...)) + } + }) + } +} + +func TestForceReplicationEndpointTranslator_PreservesPayloadMetadata(t *testing.T) { + tr := NewForceReplicationEndpointTranslator(log.NewTestLogger(), proxyEndpoint) + req := forceReplicationReq() + + changed, err := tr.TranslateRequest(req) + require.NoError(t, err) + require.True(t, changed) + + require.Equal(t, map[string][]byte{ + "encoding": []byte(jsonPlainEncoding), + "someOtherKey": []byte("someOtherValue"), + }, req.GetInput().GetPayloads()[0].GetMetadata()) + // Fields other than the endpoint, TargetClusterName in particular, are left alone. + require.JSONEq(t, rewrittenParamsJSON, string(payloadData(req))) +} + +func TestForceReplicationEndpointTranslator_Idempotent(t *testing.T) { + tr := NewForceReplicationEndpointTranslator(log.NewTestLogger(), proxyEndpoint) + req := forceReplicationReq() + + changed, err := tr.TranslateRequest(req) + require.NoError(t, err) + require.True(t, changed) + once := append([]byte(nil), payloadData(req)...) + + // A second pass (retry, or a second proxy hop) must land on the same value. + changed, err = tr.TranslateRequest(req) + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, once, payloadData(req)) + require.JSONEq(t, rewrittenParamsJSON, string(payloadData(req))) +} + +func TestForceReplicationEndpointTranslator_MatchMethod(t *testing.T) { + // Pin the wire format so the cases below cannot pass vacuously. + require.Equal(t, "/temporal.api.workflowservice.v1.WorkflowService/StartWorkflowExecution", startWorkflowExecutionMethod) + + cases := []struct { + method string + wantMatch bool + }{ + {method: api.WorkflowServicePrefix + "StartWorkflowExecution", wantMatch: true}, + {method: api.WorkflowServicePrefix + "DescribeWorkflowExecution"}, + {method: api.WorkflowServicePrefix + "SignalWithStartWorkflowExecution"}, + // WorkflowServicePrefix already ends in "/", so a second one is not the real method name. + {method: api.WorkflowServicePrefix + "/StartWorkflowExecution"}, + {method: api.AdminServicePrefix + "StartWorkflowExecution"}, + {method: "StartWorkflowExecution"}, + {method: ""}, + } + + tr := NewForceReplicationEndpointTranslator(log.NewTestLogger(), proxyEndpoint) + for _, c := range cases { + t.Run(c.method, func(t *testing.T) { + require.Equal(t, c.wantMatch, tr.MatchMethod(c.method)) + }) + } +} + +func TestForceReplicationEndpointTranslator_TranslateResponse(t *testing.T) { + tr := NewForceReplicationEndpointTranslator(log.NewTestLogger(), proxyEndpoint) + + for _, resp := range []any{ + nil, + &workflowservice.StartWorkflowExecutionResponse{RunId: "run-id"}, + forceReplicationReq(), + } { + changed, err := tr.TranslateResponse(resp) + require.NoError(t, err) + require.False(t, changed) + } +} + +// TestForceReplicationEndpointTranslator_ViaInterceptor proves the translator is actually reached +// through the unary interceptor for the matched method, and only for that method. +func TestForceReplicationEndpointTranslator_ViaInterceptor(t *testing.T) { + cases := []struct { + name string + fullMethod string + wantJSON string + }{ + { + name: "matched method is rewritten", + fullMethod: api.WorkflowServicePrefix + "StartWorkflowExecution", + wantJSON: rewrittenParamsJSON, + }, + { + name: "unmatched method is untouched", + fullMethod: api.WorkflowServicePrefix + "SignalWithStartWorkflowExecution", + wantJSON: forceReplicationParamsJSON, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + tr := NewForceReplicationEndpointTranslator(log.NewTestLogger(), proxyEndpoint) + i := NewTranslationInterceptor(log.NewTestLogger(), []Translator{tr}) + + var seenByHandler []byte + handler := func(_ context.Context, req any) (any, error) { + seenByHandler = append([]byte(nil), payloadData(req)...) + return &workflowservice.StartWorkflowExecutionResponse{}, nil + } + + _, err := i.Intercept(context.Background(), forceReplicationReq(), + &grpc.UnaryServerInfo{FullMethod: c.fullMethod}, handler) + require.NoError(t, err) + require.JSONEq(t, c.wantJSON, string(seenByHandler)) + }) + } +} diff --git a/metrics/prometheus_defs.go b/metrics/prometheus_defs.go index fbb26e04..20532514 100644 --- a/metrics/prometheus_defs.go +++ b/metrics/prometheus_defs.go @@ -78,7 +78,10 @@ var ( UTF8RepairTranslationKind = "utf8repair" NamespaceTranslationKind = "namespace" SearchAttrTranslationKind = "search-attribute" - HistoryBlobMessageType = "HistoryEventBlob" + // ForceReplicationEndpointTranslationKind labels the rewrite of the force replication + // workflow's TargetClusterEndpoint argument. + ForceReplicationEndpointTranslationKind = "force-replication-endpoint" + HistoryBlobMessageType = "HistoryEventBlob" ) // GetGRPCClientMetrics helps the GRPC client metrics objects feel more like the server one diff --git a/proxy/cluster_connection.go b/proxy/cluster_connection.go index f93a2da6..4585f5d7 100644 --- a/proxy/cluster_connection.go +++ b/proxy/cluster_connection.go @@ -100,12 +100,14 @@ type ( // managedClient is updated by the multi-mux-manager that also owns the server. Needs some more cleanup. managedClient closableClientConn // nsTranslations and saTranslations are used to translate namespace and search attribute names. - nsTranslations collect.StaticBiMap[string, string] - saTranslations config.SearchAttributeTranslation - overrides AdminServiceOverrides - aclPolicy *config.ACLPolicy - shardCountConfig config.ShardCountConfig - loggers logging.LoggerProvider + nsTranslations collect.StaticBiMap[string, string] + saTranslations config.SearchAttributeTranslation + overrides AdminServiceOverrides + // Inbound only. See interceptor.NewForceReplicationEndpointTranslator. + forceReplicationEndpointOverride bool + aclPolicy *config.ACLPolicy + shardCountConfig config.ShardCountConfig + loggers logging.LoggerProvider shardManager ShardManager lcmParameters LCMParameters @@ -193,6 +195,7 @@ func NewClusterConnection(lifetime context.Context, connConfig config.ClusterCon ReplicationEndpoint: connConfig.ReplicationEndpoint, CustomSearchAttributeAliases: connConfig.CustomSearchAttributeAliases, }, + forceReplicationEndpointOverride: connConfig.ForceReplicationEndpointOverride, // TODO: There is no test checking that ACLPolicy isn't accidentally dropped aclPolicy: connConfig.ACLPolicy, shardCountConfig: connConfig.ShardCountConfig, @@ -395,6 +398,13 @@ func makeServerOptions(c serverConfiguration, tlsConfig encryption.TLSConfig) ([ c.saTranslations.FlattenMaps(), c.saTranslations.Inverse().FlattenMaps())) } + if c.forceReplicationEndpointOverride && c.overrides.ReplicationEndpoint != "" { + c.loggers.Get(LogClusterConnection).Info("force replication endpoint override enabled", + tag.Address(c.overrides.ReplicationEndpoint)) + translators = append(translators, interceptor.NewForceReplicationEndpointTranslator( + c.loggers.Get(LogInterceptor), c.overrides.ReplicationEndpoint)) + } + if len(translators) > 0 { c.loggers.Get("init").Info("Translators enabled", tag.NewAnyTag("translators", translators)) tr := interceptor.NewTranslationInterceptor(c.loggers.Get(LogInterceptor), translators)