go.temporal.io/server/service/fx.go
180 LOC · 78 covered · 102 uncovered · 14 ranges · 64 concepts · 4 introducers · 12 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 service
2
3
import (
4
"sync/atomic"
5
6
"go.temporal.io/server/common"
7
"go.temporal.io/server/common/dynamicconfig"
8
"go.temporal.io/server/common/log"
9
"go.temporal.io/server/common/log/tag"
10
"go.temporal.io/server/common/membership"
11
"go.temporal.io/server/common/metrics"
12
persistenceClient "go.temporal.io/server/common/persistence/client"
13
"go.temporal.io/server/common/primitives"
14
"go.temporal.io/server/common/quotas/calculator"
15
"go.temporal.io/server/common/rpc"
16
"go.temporal.io/server/common/rpc/interceptor"
17
"go.temporal.io/server/common/telemetry"
18
"go.uber.org/fx"
19
"google.golang.org/grpc"
20
)
21
22
type (
23
PersistenceLazyLoadedServiceResolver struct {
24
*atomic.Value // value type is membership.ServiceResolver
25
}
26
27
PersistenceRateLimitingParams struct {
28
fx.Out
29
30
PersistenceMaxQps persistenceClient.PersistenceMaxQps
31
PersistenceNamespaceMaxQps persistenceClient.PersistenceNamespaceMaxQps
32
PersistencePerShardNamespaceMaxQPS persistenceClient.PersistencePerShardNamespaceMaxQPS
33
OperatorRPSRatio persistenceClient.OperatorRPSRatio
34
PersistenceBurstRatio persistenceClient.PersistenceBurstRatio
35
DynamicRateLimitingParams persistenceClient.DynamicRateLimitingParams
36
}
37
38
GrpcServerOptionsParams struct {
39
fx.In
40
41
Logger log.Logger
42
RPCFactory common.RPCFactory
43
ServiceErrorInterceptor *interceptor.ServiceErrorInterceptor
44
RetryableInterceptor *interceptor.RetryableInterceptor
45
TelemetryInterceptor *interceptor.TelemetryInterceptor
46
NamespaceRateLimitInterceptor interceptor.NamespaceRateLimitInterceptor `optional:"true"`
47
RateLimitInterceptor *interceptor.RateLimitInterceptor
48
TracingStatsHandler telemetry.ServerStatsHandler
49
MetricsStatsHandler metrics.ServerStatsHandler
50
ContextMetadataInterceptor *interceptor.ContextMetadataInterceptor `optional:"true"`
51
AdditionalInterceptors []grpc.UnaryServerInterceptor `optional:"true"`
52
AdditionalStreamInterceptors []grpc.StreamServerInterceptor `optional:"true"`
53
}
54
)
55
56
var PersistenceLazyLoadedServiceResolverModule = fx.Options(
58
>
return PersistenceLazyLoadedServiceResolver{
59
>
Value: &atomic.Value{},
60
>
}
61
>
}),
62
fx.Invoke(initPersistenceLazyLoadedServiceResolver),
63
)
64
65
func initPersistenceLazyLoadedServiceResolver(
66
serviceName primitives.ServiceName,
67
logger log.SnTaggedLogger,
68
serviceResolver membership.ServiceResolver,
69
lazyLoadedServiceResolver PersistenceLazyLoadedServiceResolver,
71
>
lazyLoadedServiceResolver.Store(serviceResolver)
72
>
logger.Info("Initialized service resolver for persistence rate limiting", tag.Service(serviceName))
73
>
}
74
75
func (p PersistenceLazyLoadedServiceResolver) AvailableMemberCount() int {
76
if value := p.Load(); value != nil {
77
return value.(membership.ServiceResolver).AvailableMemberCount()
78
}
79
return 0
80
}
81
82
func NewPersistenceRateLimitingParams(
83
maxQps dynamicconfig.IntPropertyFn,
84
globalMaxQps dynamicconfig.IntPropertyFn,
85
namespaceMaxQps dynamicconfig.IntPropertyFnWithNamespaceFilter,
86
globalNamespaceMaxQps dynamicconfig.IntPropertyFnWithNamespaceFilter,
87
perShardNamespaceMaxQps dynamicconfig.IntPropertyFnWithNamespaceFilter,
88
operatorRPSRatio dynamicconfig.FloatPropertyFn,
89
burstRatio dynamicconfig.FloatPropertyFn,
90
dynamicRateLimitingParams dynamicconfig.TypedPropertyFn[dynamicconfig.DynamicRateLimitingParams],
91
lazyLoadedServiceResolver PersistenceLazyLoadedServiceResolver,
92
logger log.Logger,
94
>
hostCalculator := calculator.NewLoggedCalculator(
95
>
calculator.ClusterAwareQuotaCalculator{
96
>
MemberCounter: lazyLoadedServiceResolver,
97
>
PerInstanceQuota: maxQps,
98
>
GlobalQuota: globalMaxQps,
99
>
},
100
>
log.With(logger, tag.ComponentPersistence, tag.ScopeHost),
101
>
)
102
>
namespaceCalculator := calculator.NewLoggedNamespaceCalculator(
103
>
calculator.ClusterAwareNamespaceQuotaCalculator{
104
>
MemberCounter: lazyLoadedServiceResolver,
105
>
PerInstanceQuota: namespaceMaxQps,
106
>
GlobalQuota: globalNamespaceMaxQps,
107
>
},
108
>
log.With(logger, tag.ComponentPersistence, tag.ScopeNamespace),
109
>
)
110
>
return PersistenceRateLimitingParams{
111
>
PersistenceMaxQps: func() int {
112
>
return int(hostCalculator.GetQuota())
113
>
},
115
>
return int(namespaceCalculator.GetQuota(namespace))
116
>
},
117
PersistencePerShardNamespaceMaxQPS: persistenceClient.PersistencePerShardNamespaceMaxQPS(perShardNamespaceMaxQps),
118
OperatorRPSRatio: persistenceClient.OperatorRPSRatio(operatorRPSRatio),
119
PersistenceBurstRatio: persistenceClient.PersistenceBurstRatio(burstRatio),
120
DynamicRateLimitingParams: persistenceClient.DynamicRateLimitingParams(dynamicRateLimitingParams),
121
}
122
}
123
124
func GrpcServerOptionsProvider(
125
params GrpcServerOptionsParams,
127
>
128
>
grpcServerOptions, err := params.RPCFactory.GetInternodeGRPCServerOptions()
129
>
if err != nil {
130
params.Logger.Fatal("creating gRPC server options failed", tag.Error(err))
131
}
132
134
>
if params.TracingStatsHandler != nil {
136
>
}
138
>
multiStats = append(multiStats, params.MetricsStatsHandler)
139
>
}
140
>
if len(multiStats) > 0 {
141
>
grpcServerOptions = append(grpcServerOptions, grpc.StatsHandler(multiStats))
142
>
}
143
145
>
params.TelemetryInterceptor.StreamIntercept,
146
>
interceptor.CustomErrorStreamInterceptor,
147
>
}
148
>
if len(params.AdditionalStreamInterceptors) > 0 {
149
>
streamInterceptors = append(streamInterceptors, params.AdditionalStreamInterceptors...)
onebox.go ×75
150
>
}
151
153
>
grpcServerOptions,
154
>
grpc.ChainUnaryInterceptor(getUnaryInterceptors(params)...),
155
>
grpc.ChainStreamInterceptor(streamInterceptors...),
156
>
)
157
}
158
159
>
func getUnaryInterceptors(params GrpcServerOptionsParams) []grpc.UnaryServerInterceptor {
fx.go ×44
160
>
interceptors := []grpc.UnaryServerInterceptor{
161
>
params.ServiceErrorInterceptor.Intercept,
162
>
metrics.NewServerMetricsContextInjectorInterceptor(),
163
>
metrics.NewServerMetricsTrailerPropagatorInterceptor(params.Logger),
164
>
params.TelemetryInterceptor.UnaryIntercept,
165
>
}
166
>
167
>
interceptors = append(interceptors, params.AdditionalInterceptors...)
168
>
169
>
if params.NamespaceRateLimitInterceptor != nil {
170
>
interceptors = append(interceptors, params.NamespaceRateLimitInterceptor.Intercept)
171
>
}
172
174
>
175
>
if params.ContextMetadataInterceptor != nil {
176
>
interceptors = append(interceptors, params.ContextMetadataInterceptor.Intercept)
177
>
}
178
180
}