blob: 4f8836d48f6f1546af8f0e32066d38df1b02b3a3 [file] [log] [blame]
Holger Hildebrandtfa074992020-03-27 15:42:06 +00001/*
2 *
3 * Copyright 2017 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18
19package grpc
20
21import (
22 "encoding/json"
23 "fmt"
24 "strconv"
25 "strings"
26 "time"
27
28 "google.golang.org/grpc/balancer"
29 "google.golang.org/grpc/codes"
30 "google.golang.org/grpc/grpclog"
31 "google.golang.org/grpc/internal"
32 "google.golang.org/grpc/serviceconfig"
33)
34
35const maxInt = int(^uint(0) >> 1)
36
37// MethodConfig defines the configuration recommended by the service providers for a
38// particular method.
39//
40// Deprecated: Users should not use this struct. Service config should be received
41// through name resolver, as specified here
42// https://github.com/grpc/grpc/blob/master/doc/service_config.md
43type MethodConfig struct {
44 // WaitForReady indicates whether RPCs sent to this method should wait until
45 // the connection is ready by default (!failfast). The value specified via the
46 // gRPC client API will override the value set here.
47 WaitForReady *bool
48 // Timeout is the default timeout for RPCs sent to this method. The actual
49 // deadline used will be the minimum of the value specified here and the value
50 // set by the application via the gRPC client API. If either one is not set,
51 // then the other will be used. If neither is set, then the RPC has no deadline.
52 Timeout *time.Duration
53 // MaxReqSize is the maximum allowed payload size for an individual request in a
54 // stream (client->server) in bytes. The size which is measured is the serialized
55 // payload after per-message compression (but before stream compression) in bytes.
56 // The actual value used is the minimum of the value specified here and the value set
57 // by the application via the gRPC client API. If either one is not set, then the other
58 // will be used. If neither is set, then the built-in default is used.
59 MaxReqSize *int
60 // MaxRespSize is the maximum allowed payload size for an individual response in a
61 // stream (server->client) in bytes.
62 MaxRespSize *int
63 // RetryPolicy configures retry options for the method.
64 retryPolicy *retryPolicy
65}
66
67type lbConfig struct {
68 name string
69 cfg serviceconfig.LoadBalancingConfig
70}
71
72// ServiceConfig is provided by the service provider and contains parameters for how
73// clients that connect to the service should behave.
74//
75// Deprecated: Users should not use this struct. Service config should be received
76// through name resolver, as specified here
77// https://github.com/grpc/grpc/blob/master/doc/service_config.md
78type ServiceConfig struct {
79 serviceconfig.Config
80
81 // LB is the load balancer the service providers recommends. The balancer
82 // specified via grpc.WithBalancer will override this. This is deprecated;
83 // lbConfigs is preferred. If lbConfig and LB are both present, lbConfig
84 // will be used.
85 LB *string
86
87 // lbConfig is the service config's load balancing configuration. If
88 // lbConfig and LB are both present, lbConfig will be used.
89 lbConfig *lbConfig
90
91 // Methods contains a map for the methods in this service. If there is an
92 // exact match for a method (i.e. /service/method) in the map, use the
93 // corresponding MethodConfig. If there's no exact match, look for the
94 // default config for the service (/service/) and use the corresponding
95 // MethodConfig if it exists. Otherwise, the method has no MethodConfig to
96 // use.
97 Methods map[string]MethodConfig
98
99 // If a retryThrottlingPolicy is provided, gRPC will automatically throttle
100 // retry attempts and hedged RPCs when the client’s ratio of failures to
101 // successes exceeds a threshold.
102 //
103 // For each server name, the gRPC client will maintain a token_count which is
104 // initially set to maxTokens, and can take values between 0 and maxTokens.
105 //
106 // Every outgoing RPC (regardless of service or method invoked) will change
107 // token_count as follows:
108 //
109 // - Every failed RPC will decrement the token_count by 1.
110 // - Every successful RPC will increment the token_count by tokenRatio.
111 //
112 // If token_count is less than or equal to maxTokens / 2, then RPCs will not
113 // be retried and hedged RPCs will not be sent.
114 retryThrottling *retryThrottlingPolicy
115 // healthCheckConfig must be set as one of the requirement to enable LB channel
116 // health check.
117 healthCheckConfig *healthCheckConfig
118 // rawJSONString stores service config json string that get parsed into
119 // this service config struct.
120 rawJSONString string
121}
122
123// healthCheckConfig defines the go-native version of the LB channel health check config.
124type healthCheckConfig struct {
125 // serviceName is the service name to use in the health-checking request.
126 ServiceName string
127}
128
129// retryPolicy defines the go-native version of the retry policy defined by the
130// service config here:
131// https://github.com/grpc/proposal/blob/master/A6-client-retries.md#integration-with-service-config
132type retryPolicy struct {
133 // MaxAttempts is the maximum number of attempts, including the original RPC.
134 //
135 // This field is required and must be two or greater.
136 maxAttempts int
137
138 // Exponential backoff parameters. The initial retry attempt will occur at
139 // random(0, initialBackoffMS). In general, the nth attempt will occur at
140 // random(0,
141 // min(initialBackoffMS*backoffMultiplier**(n-1), maxBackoffMS)).
142 //
143 // These fields are required and must be greater than zero.
144 initialBackoff time.Duration
145 maxBackoff time.Duration
146 backoffMultiplier float64
147
148 // The set of status codes which may be retried.
149 //
150 // Status codes are specified as strings, e.g., "UNAVAILABLE".
151 //
152 // This field is required and must be non-empty.
153 // Note: a set is used to store this for easy lookup.
154 retryableStatusCodes map[codes.Code]bool
155}
156
157type jsonRetryPolicy struct {
158 MaxAttempts int
159 InitialBackoff string
160 MaxBackoff string
161 BackoffMultiplier float64
162 RetryableStatusCodes []codes.Code
163}
164
165// retryThrottlingPolicy defines the go-native version of the retry throttling
166// policy defined by the service config here:
167// https://github.com/grpc/proposal/blob/master/A6-client-retries.md#integration-with-service-config
168type retryThrottlingPolicy struct {
169 // The number of tokens starts at maxTokens. The token_count will always be
170 // between 0 and maxTokens.
171 //
172 // This field is required and must be greater than zero.
173 MaxTokens float64
174 // The amount of tokens to add on each successful RPC. Typically this will
175 // be some number between 0 and 1, e.g., 0.1.
176 //
177 // This field is required and must be greater than zero. Up to 3 decimal
178 // places are supported.
179 TokenRatio float64
180}
181
182func parseDuration(s *string) (*time.Duration, error) {
183 if s == nil {
184 return nil, nil
185 }
186 if !strings.HasSuffix(*s, "s") {
187 return nil, fmt.Errorf("malformed duration %q", *s)
188 }
189 ss := strings.SplitN((*s)[:len(*s)-1], ".", 3)
190 if len(ss) > 2 {
191 return nil, fmt.Errorf("malformed duration %q", *s)
192 }
193 // hasDigits is set if either the whole or fractional part of the number is
194 // present, since both are optional but one is required.
195 hasDigits := false
196 var d time.Duration
197 if len(ss[0]) > 0 {
198 i, err := strconv.ParseInt(ss[0], 10, 32)
199 if err != nil {
200 return nil, fmt.Errorf("malformed duration %q: %v", *s, err)
201 }
202 d = time.Duration(i) * time.Second
203 hasDigits = true
204 }
205 if len(ss) == 2 && len(ss[1]) > 0 {
206 if len(ss[1]) > 9 {
207 return nil, fmt.Errorf("malformed duration %q", *s)
208 }
209 f, err := strconv.ParseInt(ss[1], 10, 64)
210 if err != nil {
211 return nil, fmt.Errorf("malformed duration %q: %v", *s, err)
212 }
213 for i := 9; i > len(ss[1]); i-- {
214 f *= 10
215 }
216 d += time.Duration(f)
217 hasDigits = true
218 }
219 if !hasDigits {
220 return nil, fmt.Errorf("malformed duration %q", *s)
221 }
222
223 return &d, nil
224}
225
226type jsonName struct {
227 Service *string
228 Method *string
229}
230
231func (j jsonName) generatePath() (string, bool) {
232 if j.Service == nil {
233 return "", false
234 }
235 res := "/" + *j.Service + "/"
236 if j.Method != nil {
237 res += *j.Method
238 }
239 return res, true
240}
241
242// TODO(lyuxuan): delete this struct after cleaning up old service config implementation.
243type jsonMC struct {
244 Name *[]jsonName
245 WaitForReady *bool
246 Timeout *string
247 MaxRequestMessageBytes *int64
248 MaxResponseMessageBytes *int64
249 RetryPolicy *jsonRetryPolicy
250}
251
252type loadBalancingConfig map[string]json.RawMessage
253
254// TODO(lyuxuan): delete this struct after cleaning up old service config implementation.
255type jsonSC struct {
256 LoadBalancingPolicy *string
257 LoadBalancingConfig *[]loadBalancingConfig
258 MethodConfig *[]jsonMC
259 RetryThrottling *retryThrottlingPolicy
260 HealthCheckConfig *healthCheckConfig
261}
262
263func init() {
264 internal.ParseServiceConfigForTesting = parseServiceConfig
265}
266func parseServiceConfig(js string) *serviceconfig.ParseResult {
267 if len(js) == 0 {
268 return &serviceconfig.ParseResult{Err: fmt.Errorf("no JSON service config provided")}
269 }
270 var rsc jsonSC
271 err := json.Unmarshal([]byte(js), &rsc)
272 if err != nil {
273 grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
274 return &serviceconfig.ParseResult{Err: err}
275 }
276 sc := ServiceConfig{
277 LB: rsc.LoadBalancingPolicy,
278 Methods: make(map[string]MethodConfig),
279 retryThrottling: rsc.RetryThrottling,
280 healthCheckConfig: rsc.HealthCheckConfig,
281 rawJSONString: js,
282 }
283 if rsc.LoadBalancingConfig != nil {
284 for i, lbcfg := range *rsc.LoadBalancingConfig {
285 if len(lbcfg) != 1 {
286 err := fmt.Errorf("invalid loadBalancingConfig: entry %v does not contain exactly 1 policy/config pair: %q", i, lbcfg)
287 grpclog.Warningf(err.Error())
288 return &serviceconfig.ParseResult{Err: err}
289 }
290 var name string
291 var jsonCfg json.RawMessage
292 for name, jsonCfg = range lbcfg {
293 }
294 builder := balancer.Get(name)
295 if builder == nil {
296 continue
297 }
298 sc.lbConfig = &lbConfig{name: name}
299 if parser, ok := builder.(balancer.ConfigParser); ok {
300 var err error
301 sc.lbConfig.cfg, err = parser.ParseConfig(jsonCfg)
302 if err != nil {
303 return &serviceconfig.ParseResult{Err: fmt.Errorf("error parsing loadBalancingConfig for policy %q: %v", name, err)}
304 }
305 } else if string(jsonCfg) != "{}" {
306 grpclog.Warningf("non-empty balancer configuration %q, but balancer does not implement ParseConfig", string(jsonCfg))
307 }
308 break
309 }
310 if sc.lbConfig == nil {
311 // We had a loadBalancingConfig field but did not encounter a
312 // supported policy. The config is considered invalid in this
313 // case.
314 err := fmt.Errorf("invalid loadBalancingConfig: no supported policies found")
315 grpclog.Warningf(err.Error())
316 return &serviceconfig.ParseResult{Err: err}
317 }
318 }
319
320 if rsc.MethodConfig == nil {
321 return &serviceconfig.ParseResult{Config: &sc}
322 }
323 for _, m := range *rsc.MethodConfig {
324 if m.Name == nil {
325 continue
326 }
327 d, err := parseDuration(m.Timeout)
328 if err != nil {
329 grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
330 return &serviceconfig.ParseResult{Err: err}
331 }
332
333 mc := MethodConfig{
334 WaitForReady: m.WaitForReady,
335 Timeout: d,
336 }
337 if mc.retryPolicy, err = convertRetryPolicy(m.RetryPolicy); err != nil {
338 grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
339 return &serviceconfig.ParseResult{Err: err}
340 }
341 if m.MaxRequestMessageBytes != nil {
342 if *m.MaxRequestMessageBytes > int64(maxInt) {
343 mc.MaxReqSize = newInt(maxInt)
344 } else {
345 mc.MaxReqSize = newInt(int(*m.MaxRequestMessageBytes))
346 }
347 }
348 if m.MaxResponseMessageBytes != nil {
349 if *m.MaxResponseMessageBytes > int64(maxInt) {
350 mc.MaxRespSize = newInt(maxInt)
351 } else {
352 mc.MaxRespSize = newInt(int(*m.MaxResponseMessageBytes))
353 }
354 }
355 for _, n := range *m.Name {
356 if path, valid := n.generatePath(); valid {
357 sc.Methods[path] = mc
358 }
359 }
360 }
361
362 if sc.retryThrottling != nil {
363 if mt := sc.retryThrottling.MaxTokens; mt <= 0 || mt > 1000 {
364 return &serviceconfig.ParseResult{Err: fmt.Errorf("invalid retry throttling config: maxTokens (%v) out of range (0, 1000]", mt)}
365 }
366 if tr := sc.retryThrottling.TokenRatio; tr <= 0 {
367 return &serviceconfig.ParseResult{Err: fmt.Errorf("invalid retry throttling config: tokenRatio (%v) may not be negative", tr)}
368 }
369 }
370 return &serviceconfig.ParseResult{Config: &sc}
371}
372
373func convertRetryPolicy(jrp *jsonRetryPolicy) (p *retryPolicy, err error) {
374 if jrp == nil {
375 return nil, nil
376 }
377 ib, err := parseDuration(&jrp.InitialBackoff)
378 if err != nil {
379 return nil, err
380 }
381 mb, err := parseDuration(&jrp.MaxBackoff)
382 if err != nil {
383 return nil, err
384 }
385
386 if jrp.MaxAttempts <= 1 ||
387 *ib <= 0 ||
388 *mb <= 0 ||
389 jrp.BackoffMultiplier <= 0 ||
390 len(jrp.RetryableStatusCodes) == 0 {
391 grpclog.Warningf("grpc: ignoring retry policy %v due to illegal configuration", jrp)
392 return nil, nil
393 }
394
395 rp := &retryPolicy{
396 maxAttempts: jrp.MaxAttempts,
397 initialBackoff: *ib,
398 maxBackoff: *mb,
399 backoffMultiplier: jrp.BackoffMultiplier,
400 retryableStatusCodes: make(map[codes.Code]bool),
401 }
402 if rp.maxAttempts > 5 {
403 // TODO(retry): Make the max maxAttempts configurable.
404 rp.maxAttempts = 5
405 }
406 for _, code := range jrp.RetryableStatusCodes {
407 rp.retryableStatusCodes[code] = true
408 }
409 return rp, nil
410}
411
412func min(a, b *int) *int {
413 if *a < *b {
414 return a
415 }
416 return b
417}
418
419func getMaxSize(mcMax, doptMax *int, defaultVal int) *int {
420 if mcMax == nil && doptMax == nil {
421 return &defaultVal
422 }
423 if mcMax != nil && doptMax != nil {
424 return min(mcMax, doptMax)
425 }
426 if mcMax != nil {
427 return mcMax
428 }
429 return doptMax
430}
431
432func newInt(b int) *int {
433 return &b
434}