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)