11package nexusoperation
22
33import (
4+ "context"
5+ "fmt"
6+ "net/http"
7+
8+ "go.temporal.io/api/serviceerror"
9+ persistencespb "go.temporal.io/server/api/persistence/v1"
410 "go.temporal.io/server/chasm"
11+ "go.temporal.io/server/common"
12+ "go.temporal.io/server/common/cluster"
13+ "go.temporal.io/server/common/collection"
14+ "go.temporal.io/server/common/dynamicconfig"
15+ "go.temporal.io/server/common/log"
16+ "go.temporal.io/server/common/metrics"
17+ commonnexus "go.temporal.io/server/common/nexus"
18+ "go.temporal.io/server/common/nexus/nexusrpc"
19+ "go.temporal.io/server/common/persistence"
20+ "go.temporal.io/server/common/resource"
21+ "go.temporal.io/server/common/rpc"
522 "go.uber.org/fx"
623)
724
25+ const nexusCallbackSourceHeader = "Nexus-Callback-Source"
26+
827var Module = fx .Module (
9- "chasm.lib.nexusoperations " ,
28+ "chasm.lib.nexusoperation " ,
1029 fx .Provide (configProvider ),
11- fx .Provide (NewCancellationBackoffTaskHandler ),
12- fx .Provide (NewCancellationTaskHandler ),
13- fx .Provide (NewOperationBackoffTaskHandler ),
14- fx .Provide (NewOperationInvocationTaskHandler ),
15- fx .Provide (NewOperationScheduleToCloseTimeoutTaskHandler ),
16- fx .Provide (NewOperationScheduleToStartTimeoutTaskHandler ),
17- fx .Provide (NewOperationStartToCloseTimeoutTaskHandler ),
30+ fx .Provide (commonnexus .NewCallbackTokenGenerator ),
31+ fx .Provide (endpointRegistryProvider ),
32+ fx .Invoke (endpointRegistryLifetimeHooks ),
33+ fx .Provide (defaultNexusTransportProvider ),
34+ fx .Provide (clientProviderFactory ),
35+ fx .Provide (newCancellationBackoffTaskHandler ),
36+ fx .Provide (newCancellationTaskHandler ),
37+ fx .Provide (newOperationBackoffTaskHandler ),
38+ fx .Provide (newOperationInvocationTaskHandler ),
39+ fx .Provide (newOperationScheduleToCloseTimeoutTaskHandler ),
40+ fx .Provide (newOperationScheduleToStartTimeoutTaskHandler ),
41+ fx .Provide (newOperationStartToCloseTimeoutTaskHandler ),
1842 fx .Provide (newLibrary ),
1943 fx .Invoke (register ),
2044)
@@ -25,3 +49,116 @@ func register(
2549) error {
2650 return registry .Register (library )
2751}
52+
53+ func endpointRegistryProvider (
54+ matchingClient resource.MatchingClient ,
55+ endpointManager persistence.NexusEndpointManager ,
56+ dc * dynamicconfig.Collection ,
57+ logger log.Logger ,
58+ metricsHandler metrics.Handler ,
59+ ) commonnexus.EndpointRegistry {
60+ registryConfig := commonnexus .NewEndpointRegistryConfig (dc )
61+ return commonnexus .NewEndpointRegistry (
62+ registryConfig ,
63+ matchingClient ,
64+ endpointManager ,
65+ logger ,
66+ metricsHandler ,
67+ )
68+ }
69+
70+ func endpointRegistryLifetimeHooks (lc fx.Lifecycle , registry commonnexus.EndpointRegistry ) {
71+ lc .Append (fx .StartStopHook (registry .StartLifecycle , registry .StopLifecycle ))
72+ }
73+
74+ // NexusTransportProvider allows customization of the HTTP transport used for Nexus requests.
75+ type NexusTransportProvider func (namespaceID , serviceName string ) http.RoundTripper
76+
77+ func defaultNexusTransportProvider () NexusTransportProvider {
78+ return func (namespaceID , serviceName string ) http.RoundTripper {
79+ return http .DefaultTransport
80+ }
81+ }
82+
83+ // responseSizeLimiter wraps an http.RoundTripper to limit response body size.
84+ type responseSizeLimiter struct {
85+ rt http.RoundTripper
86+ }
87+
88+ func (r responseSizeLimiter ) RoundTrip (request * http.Request ) (* http.Response , error ) {
89+ response , err := r .rt .RoundTrip (request )
90+ if err != nil {
91+ return nil , err
92+ }
93+ response .Body = http .MaxBytesReader (nil , response .Body , rpc .MaxNexusAPIRequestBodyBytes )
94+ return response , nil
95+ }
96+
97+ type clientProviderCacheKey struct {
98+ namespaceID , endpointID string
99+ url string
100+ }
101+
102+ func clientProviderFactory (
103+ httpTransportProvider NexusTransportProvider ,
104+ clusterMetadata cluster.Metadata ,
105+ rpcFactory common.RPCFactory ,
106+ config * Config ,
107+ ) (ClientProvider , error ) {
108+ cl , err := rpcFactory .CreateLocalFrontendHTTPClient ()
109+ if err != nil {
110+ return nil , fmt .Errorf ("cannot create local frontend HTTP client: %w" , err )
111+ }
112+ var clusterID string
113+
114+ if clusterInfo , ok := clusterMetadata .GetAllClusterInfo ()[clusterMetadata .GetCurrentClusterName ()]; ok {
115+ clusterID = clusterInfo .ClusterID
116+ }
117+ m := collection .NewFallibleOnceMap (func (key clientProviderCacheKey ) (* http.Client , error ) {
118+ transport := httpTransportProvider (key .namespaceID , key .endpointID )
119+ return & http.Client {
120+ Transport : responseSizeLimiter {transport },
121+ }, nil
122+ })
123+
124+ return func (ctx context.Context , namespaceID string , entry * persistencespb.NexusEndpointEntry , service string ) (* nexusrpc.HTTPClient , error ) {
125+ var url string
126+ var httpClient * http.Client
127+ httpCaller := httpClient .Do
128+ switch variant := entry .Endpoint .Spec .Target .Variant .(type ) {
129+ case * persistencespb.NexusEndpointTarget_External_ :
130+ url = variant .External .GetUrl ()
131+ var err error
132+ httpClient , err = m .Get (clientProviderCacheKey {namespaceID , entry .Id , url })
133+ if err != nil {
134+ return nil , err
135+ }
136+ if clusterID != "" {
137+ httpCaller = func (r * http.Request ) (* http.Response , error ) {
138+ resp , callErr := httpClient .Do (r )
139+ commonnexus .SetFailureSourceOnContext (ctx , resp )
140+ return resp , callErr
141+ }
142+ }
143+ case * persistencespb.NexusEndpointTarget_Worker_ :
144+ url = cl .BaseURL () + "/" + commonnexus .RouteDispatchNexusTaskByEndpoint .Path (entry .Id )
145+ httpClient = & cl .Client
146+ if clusterID != "" {
147+ httpCaller = func (r * http.Request ) (* http.Response , error ) {
148+ r .Header .Set (nexusCallbackSourceHeader , clusterID )
149+ resp , callErr := httpClient .Do (r )
150+ commonnexus .SetFailureSourceOnContext (ctx , resp )
151+ return resp , callErr
152+ }
153+ }
154+ default :
155+ return nil , serviceerror .NewInternal ("got unexpected endpoint target" )
156+ }
157+ return nexusrpc .NewHTTPClient (nexusrpc.HTTPClientOptions {
158+ BaseURL : url ,
159+ Service : service ,
160+ HTTPCaller : httpCaller ,
161+ Serializer : commonnexus .PayloadSerializer ,
162+ })
163+ }, nil
164+ }
0 commit comments