go.temporal.io/server/common/rpc/grpc.go
137 LOC · 58 covered · 79 uncovered · 9 ranges · 76 concepts · 4 introducers · 16 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.
package rpc
import (
"context"
"crypto/tls"
"net"
"time"
"go.temporal.io/server/common/headers"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/metrics"
"go.temporal.io/server/common/rpc/interceptor"
serviceerrors "go.temporal.io/server/common/serviceerror"
"google.golang.org/grpc"
"google.golang.org/grpc/backoff"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/status"
)
const (
// DefaultServiceConfig is a default gRPC connection service config which enables DNS round robin between IPs.
// To use DNS resolver, a "dns:///" prefix should be applied to the hostPort.
// https://github.com/grpc/grpc/blob/master/doc/naming.md
DefaultServiceConfig = `{"loadBalancingConfig": [{"round_robin":{}}]}`
// MaxBackoffDelay is a maximum interval between reconnect attempts.
MaxBackoffDelay = 10 * time.Second
// MaxHTTPAPIRequestBytes is the maximum number of bytes an HTTP API request
// can have. This is currently set to the max gRPC request size.
MaxHTTPAPIRequestBytes = 4 * 1024 * 1024
// MaxNexusAPIRequestBodyBytes is the maximum number of bytes a Nexus HTTP API request can have. Because the body is
// read into a Payload object, this is currently set to the max Payload size. Content headers are transformed to
// Payload metadata and contribute to the Payload size as well. A separate limit is enforced on top of this.
MaxNexusAPIRequestBodyBytes = 2 * 1024 * 1024
// minConnectTimeout is the minimum amount of time we are willing to give a connection to complete.
minConnectTimeout = 20 * time.Second
// maxInternodeRecvPayloadSize indicates the internode max receive payload size.
maxInternodeRecvPayloadSize = 128 * 1024 * 1024 // 128 Mb
)
// Dial creates a client connection to the given target with default options.
// The hostName syntax is defined in
// https://github.com/grpc/grpc/blob/master/doc/naming.md.
// dns resolver is used by default
func Dial(
hostName string,
tlsConfig *tls.Config,
logger log.Logger,
metricsHandler metrics.Handler,
opts ...grpc.DialOption,
var grpcSecureOpt grpc.DialOption
if tlsConfig == nil {
}
// gRPC maintains connection pool inside grpc.ClientConn.
// This connection pool has auto reconnect feature.
// If connection goes down, gRPC will try to reconnect using exponential backoff strategy:
// https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md.
// Default MaxDelay is 120 seconds which is too high.
Backoff: backoff.DefaultConfig,
MinConnectTimeout: minConnectTimeout,
}
cp.Backoff.MaxDelay = MaxBackoffDelay
dtrace := newDialTracer(hostName, metricsHandler, logger)
contextDialer := func(ctx context.Context, s string) (net.Conn, error) {
// We are on Go 1.23+ and can use KeepAliveConfig directly instead of the old KeepAlive/Control hacks.
dialer := &net.Dialer{
KeepAliveConfig: net.KeepAliveConfig{
Enable: true,
},
}
var ndt *networkDialTrace
ctx, ndt = dtrace.beginNetworkDial(ctx)
conn, dialErr := dialer.DialContext(ctx, "tcp", s)
dtrace.endNetworkDial(ndt, dialErr)
return conn, dialErr
}
grpcSecureOpt,
grpc.WithContextDialer(contextDialer),
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxInternodeRecvPayloadSize)),
grpc.WithChainUnaryInterceptor(
headersInterceptor,
metrics.NewClientMetricsTrailerPropagatorInterceptor(logger),
errorInterceptor,
),
grpc.WithChainStreamInterceptor(
interceptor.StreamErrorInterceptor,
),
grpc.WithDefaultServiceConfig(DefaultServiceConfig),
grpc.WithDisableServiceConfig(),
grpc.WithConnectParams(cp),
}
dialOptions = append(dialOptions, opts...)
return grpc.NewClient(hostName, dialOptions...)
}
func errorInterceptor(
ctx context.Context,
method string,
req, reply any,
cc *grpc.ClientConn,
invoker grpc.UnaryInvoker,
opts ...grpc.CallOption,
err := invoker(ctx, method, req, reply, cc, opts...)
err = serviceerrors.FromStatus(status.Convert(err))
return err
}
func headersInterceptor(
ctx context.Context,
method string,
req, reply any,
cc *grpc.ClientConn,
invoker grpc.UnaryInvoker,
opts ...grpc.CallOption,
ctx = headers.Propagate(ctx)
return invoker(ctx, method, req, reply, cc, opts...)
}