Serge Bazanski | cc25bdf | 2018-10-25 14:02:58 +0200 | [diff] [blame^] | 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 | |
| 19 | package grpc |
| 20 | |
| 21 | import ( |
| 22 | "encoding/json" |
| 23 | "fmt" |
| 24 | "strconv" |
| 25 | "strings" |
| 26 | "time" |
| 27 | |
| 28 | "google.golang.org/grpc/codes" |
| 29 | "google.golang.org/grpc/grpclog" |
| 30 | ) |
| 31 | |
| 32 | const maxInt = int(^uint(0) >> 1) |
| 33 | |
| 34 | // MethodConfig defines the configuration recommended by the service providers for a |
| 35 | // particular method. |
| 36 | // |
| 37 | // Deprecated: Users should not use this struct. Service config should be received |
| 38 | // through name resolver, as specified here |
| 39 | // https://github.com/grpc/grpc/blob/master/doc/service_config.md |
| 40 | type MethodConfig struct { |
| 41 | // WaitForReady indicates whether RPCs sent to this method should wait until |
| 42 | // the connection is ready by default (!failfast). The value specified via the |
| 43 | // gRPC client API will override the value set here. |
| 44 | WaitForReady *bool |
| 45 | // Timeout is the default timeout for RPCs sent to this method. The actual |
| 46 | // deadline used will be the minimum of the value specified here and the value |
| 47 | // set by the application via the gRPC client API. If either one is not set, |
| 48 | // then the other will be used. If neither is set, then the RPC has no deadline. |
| 49 | Timeout *time.Duration |
| 50 | // MaxReqSize is the maximum allowed payload size for an individual request in a |
| 51 | // stream (client->server) in bytes. The size which is measured is the serialized |
| 52 | // payload after per-message compression (but before stream compression) in bytes. |
| 53 | // The actual value used is the minimum of the value specified here and the value set |
| 54 | // by the application via the gRPC client API. If either one is not set, then the other |
| 55 | // will be used. If neither is set, then the built-in default is used. |
| 56 | MaxReqSize *int |
| 57 | // MaxRespSize is the maximum allowed payload size for an individual response in a |
| 58 | // stream (server->client) in bytes. |
| 59 | MaxRespSize *int |
| 60 | // RetryPolicy configures retry options for the method. |
| 61 | retryPolicy *retryPolicy |
| 62 | } |
| 63 | |
| 64 | // ServiceConfig is provided by the service provider and contains parameters for how |
| 65 | // clients that connect to the service should behave. |
| 66 | // |
| 67 | // Deprecated: Users should not use this struct. Service config should be received |
| 68 | // through name resolver, as specified here |
| 69 | // https://github.com/grpc/grpc/blob/master/doc/service_config.md |
| 70 | type ServiceConfig struct { |
| 71 | // LB is the load balancer the service providers recommends. The balancer specified |
| 72 | // via grpc.WithBalancer will override this. |
| 73 | LB *string |
| 74 | |
| 75 | // Methods contains a map for the methods in this service. If there is an |
| 76 | // exact match for a method (i.e. /service/method) in the map, use the |
| 77 | // corresponding MethodConfig. If there's no exact match, look for the |
| 78 | // default config for the service (/service/) and use the corresponding |
| 79 | // MethodConfig if it exists. Otherwise, the method has no MethodConfig to |
| 80 | // use. |
| 81 | Methods map[string]MethodConfig |
| 82 | |
| 83 | // If a retryThrottlingPolicy is provided, gRPC will automatically throttle |
| 84 | // retry attempts and hedged RPCs when the client’s ratio of failures to |
| 85 | // successes exceeds a threshold. |
| 86 | // |
| 87 | // For each server name, the gRPC client will maintain a token_count which is |
| 88 | // initially set to maxTokens, and can take values between 0 and maxTokens. |
| 89 | // |
| 90 | // Every outgoing RPC (regardless of service or method invoked) will change |
| 91 | // token_count as follows: |
| 92 | // |
| 93 | // - Every failed RPC will decrement the token_count by 1. |
| 94 | // - Every successful RPC will increment the token_count by tokenRatio. |
| 95 | // |
| 96 | // If token_count is less than or equal to maxTokens / 2, then RPCs will not |
| 97 | // be retried and hedged RPCs will not be sent. |
| 98 | retryThrottling *retryThrottlingPolicy |
| 99 | } |
| 100 | |
| 101 | // retryPolicy defines the go-native version of the retry policy defined by the |
| 102 | // service config here: |
| 103 | // https://github.com/grpc/proposal/blob/master/A6-client-retries.md#integration-with-service-config |
| 104 | type retryPolicy struct { |
| 105 | // MaxAttempts is the maximum number of attempts, including the original RPC. |
| 106 | // |
| 107 | // This field is required and must be two or greater. |
| 108 | maxAttempts int |
| 109 | |
| 110 | // Exponential backoff parameters. The initial retry attempt will occur at |
| 111 | // random(0, initialBackoffMS). In general, the nth attempt will occur at |
| 112 | // random(0, |
| 113 | // min(initialBackoffMS*backoffMultiplier**(n-1), maxBackoffMS)). |
| 114 | // |
| 115 | // These fields are required and must be greater than zero. |
| 116 | initialBackoff time.Duration |
| 117 | maxBackoff time.Duration |
| 118 | backoffMultiplier float64 |
| 119 | |
| 120 | // The set of status codes which may be retried. |
| 121 | // |
| 122 | // Status codes are specified as strings, e.g., "UNAVAILABLE". |
| 123 | // |
| 124 | // This field is required and must be non-empty. |
| 125 | // Note: a set is used to store this for easy lookup. |
| 126 | retryableStatusCodes map[codes.Code]bool |
| 127 | } |
| 128 | |
| 129 | type jsonRetryPolicy struct { |
| 130 | MaxAttempts int |
| 131 | InitialBackoff string |
| 132 | MaxBackoff string |
| 133 | BackoffMultiplier float64 |
| 134 | RetryableStatusCodes []codes.Code |
| 135 | } |
| 136 | |
| 137 | // retryThrottlingPolicy defines the go-native version of the retry throttling |
| 138 | // policy defined by the service config here: |
| 139 | // https://github.com/grpc/proposal/blob/master/A6-client-retries.md#integration-with-service-config |
| 140 | type retryThrottlingPolicy struct { |
| 141 | // The number of tokens starts at maxTokens. The token_count will always be |
| 142 | // between 0 and maxTokens. |
| 143 | // |
| 144 | // This field is required and must be greater than zero. |
| 145 | MaxTokens float64 |
| 146 | // The amount of tokens to add on each successful RPC. Typically this will |
| 147 | // be some number between 0 and 1, e.g., 0.1. |
| 148 | // |
| 149 | // This field is required and must be greater than zero. Up to 3 decimal |
| 150 | // places are supported. |
| 151 | TokenRatio float64 |
| 152 | } |
| 153 | |
| 154 | func parseDuration(s *string) (*time.Duration, error) { |
| 155 | if s == nil { |
| 156 | return nil, nil |
| 157 | } |
| 158 | if !strings.HasSuffix(*s, "s") { |
| 159 | return nil, fmt.Errorf("malformed duration %q", *s) |
| 160 | } |
| 161 | ss := strings.SplitN((*s)[:len(*s)-1], ".", 3) |
| 162 | if len(ss) > 2 { |
| 163 | return nil, fmt.Errorf("malformed duration %q", *s) |
| 164 | } |
| 165 | // hasDigits is set if either the whole or fractional part of the number is |
| 166 | // present, since both are optional but one is required. |
| 167 | hasDigits := false |
| 168 | var d time.Duration |
| 169 | if len(ss[0]) > 0 { |
| 170 | i, err := strconv.ParseInt(ss[0], 10, 32) |
| 171 | if err != nil { |
| 172 | return nil, fmt.Errorf("malformed duration %q: %v", *s, err) |
| 173 | } |
| 174 | d = time.Duration(i) * time.Second |
| 175 | hasDigits = true |
| 176 | } |
| 177 | if len(ss) == 2 && len(ss[1]) > 0 { |
| 178 | if len(ss[1]) > 9 { |
| 179 | return nil, fmt.Errorf("malformed duration %q", *s) |
| 180 | } |
| 181 | f, err := strconv.ParseInt(ss[1], 10, 64) |
| 182 | if err != nil { |
| 183 | return nil, fmt.Errorf("malformed duration %q: %v", *s, err) |
| 184 | } |
| 185 | for i := 9; i > len(ss[1]); i-- { |
| 186 | f *= 10 |
| 187 | } |
| 188 | d += time.Duration(f) |
| 189 | hasDigits = true |
| 190 | } |
| 191 | if !hasDigits { |
| 192 | return nil, fmt.Errorf("malformed duration %q", *s) |
| 193 | } |
| 194 | |
| 195 | return &d, nil |
| 196 | } |
| 197 | |
| 198 | type jsonName struct { |
| 199 | Service *string |
| 200 | Method *string |
| 201 | } |
| 202 | |
| 203 | func (j jsonName) generatePath() (string, bool) { |
| 204 | if j.Service == nil { |
| 205 | return "", false |
| 206 | } |
| 207 | res := "/" + *j.Service + "/" |
| 208 | if j.Method != nil { |
| 209 | res += *j.Method |
| 210 | } |
| 211 | return res, true |
| 212 | } |
| 213 | |
| 214 | // TODO(lyuxuan): delete this struct after cleaning up old service config implementation. |
| 215 | type jsonMC struct { |
| 216 | Name *[]jsonName |
| 217 | WaitForReady *bool |
| 218 | Timeout *string |
| 219 | MaxRequestMessageBytes *int64 |
| 220 | MaxResponseMessageBytes *int64 |
| 221 | RetryPolicy *jsonRetryPolicy |
| 222 | } |
| 223 | |
| 224 | // TODO(lyuxuan): delete this struct after cleaning up old service config implementation. |
| 225 | type jsonSC struct { |
| 226 | LoadBalancingPolicy *string |
| 227 | MethodConfig *[]jsonMC |
| 228 | RetryThrottling *retryThrottlingPolicy |
| 229 | } |
| 230 | |
| 231 | func parseServiceConfig(js string) (ServiceConfig, error) { |
| 232 | if len(js) == 0 { |
| 233 | return ServiceConfig{}, fmt.Errorf("no JSON service config provided") |
| 234 | } |
| 235 | var rsc jsonSC |
| 236 | err := json.Unmarshal([]byte(js), &rsc) |
| 237 | if err != nil { |
| 238 | grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err) |
| 239 | return ServiceConfig{}, err |
| 240 | } |
| 241 | sc := ServiceConfig{ |
| 242 | LB: rsc.LoadBalancingPolicy, |
| 243 | Methods: make(map[string]MethodConfig), |
| 244 | retryThrottling: rsc.RetryThrottling, |
| 245 | } |
| 246 | if rsc.MethodConfig == nil { |
| 247 | return sc, nil |
| 248 | } |
| 249 | |
| 250 | for _, m := range *rsc.MethodConfig { |
| 251 | if m.Name == nil { |
| 252 | continue |
| 253 | } |
| 254 | d, err := parseDuration(m.Timeout) |
| 255 | if err != nil { |
| 256 | grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err) |
| 257 | return ServiceConfig{}, err |
| 258 | } |
| 259 | |
| 260 | mc := MethodConfig{ |
| 261 | WaitForReady: m.WaitForReady, |
| 262 | Timeout: d, |
| 263 | } |
| 264 | if mc.retryPolicy, err = convertRetryPolicy(m.RetryPolicy); err != nil { |
| 265 | grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err) |
| 266 | return ServiceConfig{}, err |
| 267 | } |
| 268 | if m.MaxRequestMessageBytes != nil { |
| 269 | if *m.MaxRequestMessageBytes > int64(maxInt) { |
| 270 | mc.MaxReqSize = newInt(maxInt) |
| 271 | } else { |
| 272 | mc.MaxReqSize = newInt(int(*m.MaxRequestMessageBytes)) |
| 273 | } |
| 274 | } |
| 275 | if m.MaxResponseMessageBytes != nil { |
| 276 | if *m.MaxResponseMessageBytes > int64(maxInt) { |
| 277 | mc.MaxRespSize = newInt(maxInt) |
| 278 | } else { |
| 279 | mc.MaxRespSize = newInt(int(*m.MaxResponseMessageBytes)) |
| 280 | } |
| 281 | } |
| 282 | for _, n := range *m.Name { |
| 283 | if path, valid := n.generatePath(); valid { |
| 284 | sc.Methods[path] = mc |
| 285 | } |
| 286 | } |
| 287 | } |
| 288 | |
| 289 | if sc.retryThrottling != nil { |
| 290 | if sc.retryThrottling.MaxTokens <= 0 || |
| 291 | sc.retryThrottling.MaxTokens >= 1000 || |
| 292 | sc.retryThrottling.TokenRatio <= 0 { |
| 293 | // Illegal throttling config; disable throttling. |
| 294 | sc.retryThrottling = nil |
| 295 | } |
| 296 | } |
| 297 | return sc, nil |
| 298 | } |
| 299 | |
| 300 | func convertRetryPolicy(jrp *jsonRetryPolicy) (p *retryPolicy, err error) { |
| 301 | if jrp == nil { |
| 302 | return nil, nil |
| 303 | } |
| 304 | ib, err := parseDuration(&jrp.InitialBackoff) |
| 305 | if err != nil { |
| 306 | return nil, err |
| 307 | } |
| 308 | mb, err := parseDuration(&jrp.MaxBackoff) |
| 309 | if err != nil { |
| 310 | return nil, err |
| 311 | } |
| 312 | |
| 313 | if jrp.MaxAttempts <= 1 || |
| 314 | *ib <= 0 || |
| 315 | *mb <= 0 || |
| 316 | jrp.BackoffMultiplier <= 0 || |
| 317 | len(jrp.RetryableStatusCodes) == 0 { |
| 318 | grpclog.Warningf("grpc: ignoring retry policy %v due to illegal configuration", jrp) |
| 319 | return nil, nil |
| 320 | } |
| 321 | |
| 322 | rp := &retryPolicy{ |
| 323 | maxAttempts: jrp.MaxAttempts, |
| 324 | initialBackoff: *ib, |
| 325 | maxBackoff: *mb, |
| 326 | backoffMultiplier: jrp.BackoffMultiplier, |
| 327 | retryableStatusCodes: make(map[codes.Code]bool), |
| 328 | } |
| 329 | if rp.maxAttempts > 5 { |
| 330 | // TODO(retry): Make the max maxAttempts configurable. |
| 331 | rp.maxAttempts = 5 |
| 332 | } |
| 333 | for _, code := range jrp.RetryableStatusCodes { |
| 334 | rp.retryableStatusCodes[code] = true |
| 335 | } |
| 336 | return rp, nil |
| 337 | } |
| 338 | |
| 339 | func min(a, b *int) *int { |
| 340 | if *a < *b { |
| 341 | return a |
| 342 | } |
| 343 | return b |
| 344 | } |
| 345 | |
| 346 | func getMaxSize(mcMax, doptMax *int, defaultVal int) *int { |
| 347 | if mcMax == nil && doptMax == nil { |
| 348 | return &defaultVal |
| 349 | } |
| 350 | if mcMax != nil && doptMax != nil { |
| 351 | return min(mcMax, doptMax) |
| 352 | } |
| 353 | if mcMax != nil { |
| 354 | return mcMax |
| 355 | } |
| 356 | return doptMax |
| 357 | } |
| 358 | |
| 359 | func newInt(b int) *int { |
| 360 | return &b |
| 361 | } |