1/* 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} 435