go.temporal.io/server/common/rpc/rpc.go
413 LOC · 198 covered · 215 uncovered · 54 ranges · 82 concepts · 18 introducers · 19 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package rpc
2
3
import (
4
"cmp"
5
"context"
6
"crypto/tls"
7
"fmt"
8
"math"
9
"math/rand"
10
"net"
11
"net/http"
12
"net/url"
13
"sync"
14
"time"
15
16
"go.temporal.io/api/serviceerror"
17
"go.temporal.io/server/common"
18
"go.temporal.io/server/common/config"
19
"go.temporal.io/server/common/convert"
20
"go.temporal.io/server/common/log"
21
"go.temporal.io/server/common/log/tag"
22
"go.temporal.io/server/common/membership"
23
"go.temporal.io/server/common/metrics"
24
"go.temporal.io/server/common/primitives"
25
"go.temporal.io/server/common/rpc/auth"
26
"go.temporal.io/server/common/rpc/encryption"
27
"go.temporal.io/server/temporal/environment"
28
"google.golang.org/grpc"
29
"google.golang.org/grpc/codes"
30
"google.golang.org/grpc/credentials"
31
"google.golang.org/grpc/keepalive"
32
"google.golang.org/grpc/status"
33
)
34
35
var _ common.RPCFactory = (*RPCFactory)(nil)
36
37
// RPCFactory is an implementation of common.RPCFactory interface
38
type RPCFactory struct {
39
config *config.Config
40
serviceName primitives.ServiceName
41
logger log.Logger
42
metricsHandler metrics.Handler
43
44
frontendURL string
45
frontendHTTPURL string
46
frontendHTTPPort int
47
frontendTLSConfig *tls.Config
48
49
grpcListener func() net.Listener
50
tlsFactory encryption.TLSConfigProvider
51
commonDialOptions []grpc.DialOption
52
perServiceDialOptions map[primitives.ServiceName][]grpc.DialOption
53
tokenProvider auth.TokenProvider
54
authHeaderName string
55
requireRemoteClusterAuth bool
56
monitor membership.Monitor
57
// A OnceValues wrapper for createLocalFrontendHTTPClient.
58
localFrontendClient func() (*common.FrontendHTTPClient, error)
59
60
// TODO: Remove these flags once the keepalive settings are rolled out
61
EnableInternodeServerKeepalive bool
62
EnableInternodeClientKeepalive bool
63
}
64
65
// NewFactory builds a new RPCFactory
66
// conforming to the underlying configuration
67
func NewFactory(
68
cfg *config.Config,
69
sName primitives.ServiceName,
70
logger log.Logger,
71
metricsHandler metrics.Handler,
72
tlsProvider encryption.TLSConfigProvider,
73
frontendURL string,
74
frontendHTTPURL string,
75
frontendHTTPPort int,
76
frontendTLSConfig *tls.Config,
77
commonDialOptions []grpc.DialOption,
78
perServiceDialOptions map[primitives.ServiceName][]grpc.DialOption,
79
monitor membership.Monitor,
80
tokenProvider auth.TokenProvider,
82
>
authHeaderName := "authorization"
83
>
requireRemoteClusterAuth := false
84
>
if cfg != nil {
86
>
requireRemoteClusterAuth = cfg.Global.Authorization.RemoteClusterAuth.Require
87
>
}
89
>
config: cfg,
90
>
serviceName: sName,
91
>
logger: logger,
92
>
metricsHandler: metricsHandler,
93
>
frontendURL: frontendURL,
94
>
frontendHTTPURL: frontendHTTPURL,
95
>
frontendHTTPPort: frontendHTTPPort,
96
>
frontendTLSConfig: frontendTLSConfig,
97
>
tlsFactory: tlsProvider,
98
>
commonDialOptions: commonDialOptions,
99
>
perServiceDialOptions: perServiceDialOptions,
100
>
tokenProvider: tokenProvider,
101
>
authHeaderName: authHeaderName,
102
>
requireRemoteClusterAuth: requireRemoteClusterAuth,
103
>
monitor: monitor,
104
>
}
105
>
f.grpcListener = sync.OnceValue(f.createGRPCListener)
106
>
f.localFrontendClient = sync.OnceValues(f.createLocalFrontendHTTPClient)
107
>
return f
108
}
109
111
>
var opts []grpc.ServerOption
112
>
113
>
if d.tlsFactory != nil {
115
>
if err != nil {
116
return nil, err
117
}
119
>
return opts, nil
120
>
}
121
opts = append(opts, grpc.Creds(credentials.NewTLS(serverConfig)))
122
}
123
125
}
126
127
func (d *RPCFactory) GetFrontendClientTlsConfig() (*tls.Config, error) {
128
if d.tlsFactory != nil {
129
return d.tlsFactory.GetFrontendClientConfig()
130
}
131
132
return nil, nil
133
}
134
135
func (d *RPCFactory) GetRemoteClusterClientConfig(hostname string) (*tls.Config, error) {
136
if d.tlsFactory != nil {
137
return d.tlsFactory.GetRemoteClusterClientConfig(hostname)
138
}
139
140
return nil, nil
141
}
142
144
>
var opts []grpc.ServerOption
145
>
146
>
if d.EnableInternodeServerKeepalive {
147
rpcConfig := d.config.Services[string(d.serviceName)].RPC
148
kep := rpcConfig.KeepAliveServerConfig.GetKeepAliveEnforcementPolicy()
149
kp := rpcConfig.KeepAliveServerConfig.GetKeepAliveServerParameters()
150
opts = append(opts, grpc.KeepaliveEnforcementPolicy(kep), grpc.KeepaliveParams(kp))
151
}
154
>
if err != nil {
155
return nil, err
156
}
158
>
return opts, nil
159
>
}
160
opts = append(opts, grpc.Creds(credentials.NewTLS(serverConfig)))
161
}
162
164
}
165
166
func (d *RPCFactory) GetInternodeClientTlsConfig() (*tls.Config, error) {
167
if d.tlsFactory != nil {
168
return d.tlsFactory.GetInternodeClientConfig()
169
}
170
171
return nil, nil
172
}
173
174
// GetGRPCListener returns cached dispatcher for gRPC inbound or creates one
176
>
return d.grpcListener()
177
>
}
178
180
>
rpcConfig := d.config.Services[string(d.serviceName)].RPC
181
>
hostAddress := net.JoinHostPort(getListenIP(&rpcConfig, d.logger).String(), convert.IntToString(rpcConfig.GRPCPort))
182
>
183
>
grpcListener, err := net.Listen("tcp", hostAddress)
184
>
if err != nil || grpcListener == nil || grpcListener.Addr() == nil {
185
d.logger.Fatal("Failed to start gRPC listener", tag.Error(err), tag.Service(d.serviceName), tag.Address(hostAddress))
186
}
187
188
>
d.logger.Info("Created gRPC listener", tag.Service(d.serviceName), tag.Address(hostAddress))
fx.go ×44
189
>
return grpcListener
190
}
191
193
>
if cfg.BindOnLocalHost && len(cfg.BindOnIP) > 0 {
194
logger.Fatal("ListenIP failed, bindOnLocalHost and bindOnIP are mutually exclusive")
195
return nil
196
}
197
200
>
}
201
203
>
ip := net.ParseIP(cfg.BindOnIP)
204
>
if ip != nil {
205
>
return ip
206
>
}
207
logger.Fatal("ListenIP failed, unable to parse bindOnIP value", tag.Address(cfg.BindOnIP))
208
return nil
209
}
210
ip, err := config.ListenIP()
211
if err != nil {
212
logger.Fatal("ListenIP failed", tag.Error(err))
213
return nil
214
}
215
return ip
216
}
217
218
// CreateRemoteFrontendGRPCConnection creates a gRPC connection for cross-cluster calls.
219
>
func (d *RPCFactory) CreateRemoteFrontendGRPCConnection(rpcAddress string) *grpc.ClientConn {
rpc.go ×4
220
>
var tlsClientConfig *tls.Config
221
>
var err error
222
>
if d.tlsFactory != nil {
223
>
hostname, _, err2 := net.SplitHostPort(rpcAddress)
224
>
if err2 != nil {
225
d.logger.Fatal("Invalid rpcAddress for remote cluster", tag.Error(err2))
226
}
228
>
if err != nil {
229
d.logger.Fatal("Failed to create tls config for gRPC connection", tag.Error(err))
230
return nil
231
}
232
}
233
235
>
additionalDialOptions := append([]grpc.DialOption{}, d.perServiceDialOptions[primitives.FrontendService]...)
236
>
237
>
// requireRemoteClusterAuth is defense-in-depth: temporal/fx.go boot validation
238
>
// rejects (require=true, tokenProvider=nil), but RPCFactory is also constructed in
239
>
// tests where the boot path doesn't run.
240
>
if d.tokenProvider != nil || d.requireRemoteClusterAuth {
242
>
var token string
243
>
if d.tokenProvider != nil {
244
>
t, err := d.tokenProvider.GetToken(ctx, rpcAddress)
245
>
if err != nil {
246
return "", err
247
}
249
}
251
>
return "", status.Error(codes.Unauthenticated, "no auth token available for outbound remote-cluster RPC")
rpc.go ×1
252
>
}
254
}
256
>
additionalDialOptions = append(additionalDialOptions, grpc.WithPerRPCCredentials(creds))
257
}
258
259
>
return d.dial(rpcAddress, tlsClientConfig, append(additionalDialOptions, keepAliveOption)...)
rpc.go ×4
260
}
261
262
// CreateLocalFrontendGRPCConnection creates connection for internal frontend calls
264
>
additionalDialOptions := append([]grpc.DialOption{}, d.perServiceDialOptions[primitives.InternalFrontendService]...)
265
>
266
>
return d.dial(d.frontendURL, d.frontendTLSConfig, additionalDialOptions...)
267
>
}
268
269
// createInternodeGRPCConnection creates connection for gRPC calls
270
>
func (d *RPCFactory) createInternodeGRPCConnection(hostName string, serviceName primitives.ServiceName) *grpc.ClientConn {
service_grpc.pb.go ×20
271
>
var tlsClientConfig *tls.Config
272
>
var err error
273
>
if d.tlsFactory != nil {
275
>
if err != nil {
276
d.logger.Fatal("Failed to create tls config for gRPC connection", tag.Error(err))
277
return nil
278
}
279
}
280
>
additionalDialOptions := append([]grpc.DialOption{}, d.perServiceDialOptions[serviceName]...)
service_grpc.pb.go ×20
281
>
return d.dial(hostName, tlsClientConfig, append(additionalDialOptions, d.getClientKeepAliveConfig(serviceName))...)
282
}
283
284
>
func (d *RPCFactory) CreateHistoryGRPCConnection(rpcAddress string) *grpc.ClientConn {
handler.go ×25
285
>
return d.createInternodeGRPCConnection(rpcAddress, primitives.HistoryService)
286
>
}
287
288
>
func (d *RPCFactory) CreateMatchingGRPCConnection(rpcAddress string) *grpc.ClientConn {
service_grpc.pb.go ×20
289
>
return d.createInternodeGRPCConnection(rpcAddress, primitives.MatchingService)
290
>
}
291
292
>
func (d *RPCFactory) dial(hostName string, tlsClientConfig *tls.Config, dialOptions ...grpc.DialOption) *grpc.ClientConn {
grpc.go ×4
293
>
dialOptions = append(d.commonDialOptions, dialOptions...)
294
>
connection, err := Dial(hostName, tlsClientConfig, d.logger, d.metricsHandler, dialOptions...)
295
>
if err != nil {
296
d.logger.Fatal("Failed to create gRPC connection", tag.Error(err))
297
return nil
298
}
299
301
}
302
303
>
func (d *RPCFactory) getClientKeepAliveConfig(serviceName primitives.ServiceName) grpc.DialOption {
grpc.go ×3
304
>
// default keepalive settings for clients
305
>
params := keepalive.ClientParameters{
306
>
Time: time.Duration(math.MaxInt64),
307
>
Timeout: 20 * time.Second,
308
>
PermitWithoutStream: false,
309
>
}
310
>
if d.EnableInternodeClientKeepalive {
311
serviceConfig := d.config.Services[string(serviceName)]
312
params = serviceConfig.RPC.ClientConnectionConfig.GetKeepAliveClientParameters()
313
}
315
}
316
317
func (d *RPCFactory) GetTLSConfigProvider() encryption.TLSConfigProvider {
318
return d.tlsFactory
319
}
320
321
// CreateLocalFrontendHTTPClient gets or creates a cached frontend client.
322
>
func (d *RPCFactory) CreateLocalFrontendHTTPClient() (*common.FrontendHTTPClient, error) {
rpc.go ×6
323
>
return d.localFrontendClient()
324
>
}
325
326
// createLocalFrontendHTTPClient creates an HTTP client for communicating with the frontend.
327
// It uses either the provided frontendURL or membership to resolve the frontend address.
328
>
func (d *RPCFactory) createLocalFrontendHTTPClient() (*common.FrontendHTTPClient, error) {
rpc.go ×6
329
>
// dialer and transport field values copied from http.DefaultTransport.
330
>
dialer := &net.Dialer{
331
>
Timeout: 30 * time.Second,
332
>
KeepAlive: 30 * time.Second,
333
>
}
334
>
transport := &http.Transport{
335
>
Proxy: http.ProxyFromEnvironment,
336
>
DialContext: dialer.DialContext,
337
>
ForceAttemptHTTP2: true,
338
>
MaxIdleConns: 100,
339
>
IdleConnTimeout: 90 * time.Second,
340
>
TLSHandshakeTimeout: 10 * time.Second,
341
>
ExpectContinueTimeout: 1 * time.Second,
342
>
}
343
>
client := http.Client{}
344
>
345
>
// Default to http unless TLS is configured.
346
>
scheme := "http"
347
>
if d.frontendTLSConfig != nil {
349
>
scheme = "https"
350
>
}
351
353
>
if r := serviceResolverFromGRPCURL(d.frontendHTTPURL); r != nil {
355
>
resolver: r,
356
>
underlying: transport,
357
>
httpPort: d.frontendHTTPPort,
358
>
}
359
>
address = "internal" // This will be replaced by the roundTripper
362
>
client.Transport = transport
363
>
address = d.frontendHTTPURL
364
>
}
365
367
>
Client: client,
368
>
Address: address,
369
>
Scheme: scheme,
370
>
}, nil
371
}
372
373
type roundTripper struct {
374
resolver membership.ServiceResolver
375
underlying http.RoundTripper
376
httpPort int
377
}
378
380
>
// Pick a frontend host at random.
381
>
members := rt.resolver.AvailableMembers()
382
>
if len(members) == 0 {
383
return nil, serviceerror.NewUnavailable("no frontend host to route request to")
384
}
386
>
member := members[idx]
387
>
388
>
// Replace port with the HTTP port.
389
>
host, _, err := net.SplitHostPort(member.Identity())
390
>
if err != nil {
391
return nil, fmt.Errorf("failed to extract port from frontend member: %w", err)
392
}
394
>
395
>
// Replace request's host.
396
>
req.URL.Host = address
397
>
req.Host = address
398
>
return rt.underlying.RoundTrip(req)
399
}
400
401
// serviceResolverFromGRPCURL returns a ServiceResolver if ustr corresponds to a
402
// membership url, otherwise nil.
404
>
u, err := url.Parse(ustr)
405
>
if err != nil {
407
>
}
409
>
if err != nil {
410
return nil
411
}
413
}