ceph-csi/vendor/google.golang.org/grpc/service_config.go

355 lines
11 KiB
Go
Raw Normal View History

2018-01-09 18:57:14 +00:00
/*
*
* Copyright 2017 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package grpc
import (
"encoding/json"
"errors"
2018-01-09 18:57:14 +00:00
"fmt"
"reflect"
2018-01-09 18:57:14 +00:00
"time"
"google.golang.org/grpc/balancer"
2018-11-26 18:23:56 +00:00
"google.golang.org/grpc/codes"
"google.golang.org/grpc/internal"
"google.golang.org/grpc/internal/balancer/gracefulswitch"
internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
"google.golang.org/grpc/serviceconfig"
2018-01-09 18:57:14 +00:00
)
const maxInt = int(^uint(0) >> 1)
// MethodConfig defines the configuration recommended by the service providers for a
// particular method.
2018-07-18 14:47:22 +00:00
//
// Deprecated: Users should not use this struct. Service config should be received
2018-01-09 18:57:14 +00:00
// through name resolver, as specified here
// https://github.com/grpc/grpc/blob/master/doc/service_config.md
type MethodConfig = internalserviceconfig.MethodConfig
2018-01-09 18:57:14 +00:00
// ServiceConfig is provided by the service provider and contains parameters for how
// clients that connect to the service should behave.
2018-07-18 14:47:22 +00:00
//
// Deprecated: Users should not use this struct. Service config should be received
2018-01-09 18:57:14 +00:00
// through name resolver, as specified here
// https://github.com/grpc/grpc/blob/master/doc/service_config.md
type ServiceConfig struct {
serviceconfig.Config
// lbConfig is the service config's load balancing configuration. If
// lbConfig and LB are both present, lbConfig will be used.
lbConfig serviceconfig.LoadBalancingConfig
2018-11-26 18:23:56 +00:00
// Methods contains a map for the methods in this service. If there is an
// exact match for a method (i.e. /service/method) in the map, use the
// corresponding MethodConfig. If there's no exact match, look for the
// default config for the service (/service/) and use the corresponding
// MethodConfig if it exists. Otherwise, the method has no MethodConfig to
// use.
2018-01-09 18:57:14 +00:00
Methods map[string]MethodConfig
2018-07-18 14:47:22 +00:00
2018-11-26 18:23:56 +00:00
// If a retryThrottlingPolicy is provided, gRPC will automatically throttle
// retry attempts and hedged RPCs when the clients ratio of failures to
// successes exceeds a threshold.
//
// For each server name, the gRPC client will maintain a token_count which is
// initially set to maxTokens, and can take values between 0 and maxTokens.
//
// Every outgoing RPC (regardless of service or method invoked) will change
// token_count as follows:
//
// - Every failed RPC will decrement the token_count by 1.
// - Every successful RPC will increment the token_count by tokenRatio.
//
// If token_count is less than or equal to maxTokens / 2, then RPCs will not
// be retried and hedged RPCs will not be sent.
retryThrottling *retryThrottlingPolicy
// healthCheckConfig must be set as one of the requirement to enable LB channel
// health check.
healthCheckConfig *healthCheckConfig
// rawJSONString stores service config json string that get parsed into
// this service config struct.
rawJSONString string
}
// healthCheckConfig defines the go-native version of the LB channel health check config.
type healthCheckConfig struct {
// serviceName is the service name to use in the health-checking request.
ServiceName string
2018-11-26 18:23:56 +00:00
}
type jsonRetryPolicy struct {
MaxAttempts int
InitialBackoff internalserviceconfig.Duration
MaxBackoff internalserviceconfig.Duration
2018-11-26 18:23:56 +00:00
BackoffMultiplier float64
RetryableStatusCodes []codes.Code
}
// retryThrottlingPolicy defines the go-native version of the retry throttling
// policy defined by the service config here:
// https://github.com/grpc/proposal/blob/master/A6-client-retries.md#integration-with-service-config
type retryThrottlingPolicy struct {
// The number of tokens starts at maxTokens. The token_count will always be
// between 0 and maxTokens.
//
// This field is required and must be greater than zero.
MaxTokens float64
// The amount of tokens to add on each successful RPC. Typically this will
// be some number between 0 and 1, e.g., 0.1.
//
// This field is required and must be greater than zero. Up to 3 decimal
// places are supported.
TokenRatio float64
2018-01-09 18:57:14 +00:00
}
type jsonName struct {
Service string
Method string
2018-01-09 18:57:14 +00:00
}
var (
errDuplicatedName = errors.New("duplicated name")
errEmptyServiceNonEmptyMethod = errors.New("cannot combine empty 'service' and non-empty 'method'")
)
func (j jsonName) generatePath() (string, error) {
if j.Service == "" {
if j.Method != "" {
return "", errEmptyServiceNonEmptyMethod
}
return "", nil
2018-01-09 18:57:14 +00:00
}
res := "/" + j.Service + "/"
if j.Method != "" {
res += j.Method
2018-01-09 18:57:14 +00:00
}
return res, nil
2018-01-09 18:57:14 +00:00
}
// TODO(lyuxuan): delete this struct after cleaning up old service config implementation.
type jsonMC struct {
Name *[]jsonName
WaitForReady *bool
Timeout *internalserviceconfig.Duration
2018-01-09 18:57:14 +00:00
MaxRequestMessageBytes *int64
MaxResponseMessageBytes *int64
2018-11-26 18:23:56 +00:00
RetryPolicy *jsonRetryPolicy
2018-01-09 18:57:14 +00:00
}
// TODO(lyuxuan): delete this struct after cleaning up old service config implementation.
type jsonSC struct {
2018-11-26 18:23:56 +00:00
LoadBalancingPolicy *string
LoadBalancingConfig *json.RawMessage
2018-11-26 18:23:56 +00:00
MethodConfig *[]jsonMC
RetryThrottling *retryThrottlingPolicy
HealthCheckConfig *healthCheckConfig
2018-01-09 18:57:14 +00:00
}
func init() {
internal.ParseServiceConfig = parseServiceConfig
}
func parseServiceConfig(js string) *serviceconfig.ParseResult {
if len(js) == 0 {
return &serviceconfig.ParseResult{Err: fmt.Errorf("no JSON service config provided")}
}
2018-01-09 18:57:14 +00:00
var rsc jsonSC
err := json.Unmarshal([]byte(js), &rsc)
if err != nil {
logger.Warningf("grpc: unmarshalling service config %s: %v", js, err)
return &serviceconfig.ParseResult{Err: err}
2018-01-09 18:57:14 +00:00
}
sc := ServiceConfig{
Methods: make(map[string]MethodConfig),
retryThrottling: rsc.RetryThrottling,
healthCheckConfig: rsc.HealthCheckConfig,
rawJSONString: js,
2018-01-09 18:57:14 +00:00
}
c := rsc.LoadBalancingConfig
if c == nil {
name := PickFirstBalancerName
if rsc.LoadBalancingPolicy != nil {
name = *rsc.LoadBalancingPolicy
}
if balancer.Get(name) == nil {
name = PickFirstBalancerName
}
cfg := []map[string]any{{name: struct{}{}}}
strCfg, err := json.Marshal(cfg)
if err != nil {
return &serviceconfig.ParseResult{Err: fmt.Errorf("unexpected error marshaling simple LB config: %w", err)}
}
r := json.RawMessage(strCfg)
c = &r
}
cfg, err := gracefulswitch.ParseConfig(*c)
if err != nil {
return &serviceconfig.ParseResult{Err: err}
}
sc.lbConfig = cfg
2018-01-09 18:57:14 +00:00
if rsc.MethodConfig == nil {
return &serviceconfig.ParseResult{Config: &sc}
2018-01-09 18:57:14 +00:00
}
paths := map[string]struct{}{}
2018-01-09 18:57:14 +00:00
for _, m := range *rsc.MethodConfig {
if m.Name == nil {
continue
}
mc := MethodConfig{
WaitForReady: m.WaitForReady,
Timeout: (*time.Duration)(m.Timeout),
2018-01-09 18:57:14 +00:00
}
if mc.RetryPolicy, err = convertRetryPolicy(m.RetryPolicy); err != nil {
logger.Warningf("grpc: unmarshalling service config %s: %v", js, err)
return &serviceconfig.ParseResult{Err: err}
2018-11-26 18:23:56 +00:00
}
2018-01-09 18:57:14 +00:00
if m.MaxRequestMessageBytes != nil {
if *m.MaxRequestMessageBytes > int64(maxInt) {
mc.MaxReqSize = newInt(maxInt)
} else {
mc.MaxReqSize = newInt(int(*m.MaxRequestMessageBytes))
}
}
if m.MaxResponseMessageBytes != nil {
if *m.MaxResponseMessageBytes > int64(maxInt) {
mc.MaxRespSize = newInt(maxInt)
} else {
mc.MaxRespSize = newInt(int(*m.MaxResponseMessageBytes))
}
}
for i, n := range *m.Name {
path, err := n.generatePath()
if err != nil {
logger.Warningf("grpc: error unmarshalling service config %s due to methodConfig[%d]: %v", js, i, err)
return &serviceconfig.ParseResult{Err: err}
}
if _, ok := paths[path]; ok {
err = errDuplicatedName
logger.Warningf("grpc: error unmarshalling service config %s due to methodConfig[%d]: %v", js, i, err)
return &serviceconfig.ParseResult{Err: err}
2018-01-09 18:57:14 +00:00
}
paths[path] = struct{}{}
sc.Methods[path] = mc
2018-01-09 18:57:14 +00:00
}
}
2018-11-26 18:23:56 +00:00
if sc.retryThrottling != nil {
if mt := sc.retryThrottling.MaxTokens; mt <= 0 || mt > 1000 {
return &serviceconfig.ParseResult{Err: fmt.Errorf("invalid retry throttling config: maxTokens (%v) out of range (0, 1000]", mt)}
}
if tr := sc.retryThrottling.TokenRatio; tr <= 0 {
return &serviceconfig.ParseResult{Err: fmt.Errorf("invalid retry throttling config: tokenRatio (%v) may not be negative", tr)}
2018-11-26 18:23:56 +00:00
}
}
return &serviceconfig.ParseResult{Config: &sc}
2018-01-09 18:57:14 +00:00
}
func convertRetryPolicy(jrp *jsonRetryPolicy) (p *internalserviceconfig.RetryPolicy, err error) {
2018-11-26 18:23:56 +00:00
if jrp == nil {
return nil, nil
}
if jrp.MaxAttempts <= 1 ||
jrp.InitialBackoff <= 0 ||
jrp.MaxBackoff <= 0 ||
2018-11-26 18:23:56 +00:00
jrp.BackoffMultiplier <= 0 ||
len(jrp.RetryableStatusCodes) == 0 {
logger.Warningf("grpc: ignoring retry policy %v due to illegal configuration", jrp)
2018-11-26 18:23:56 +00:00
return nil, nil
}
rp := &internalserviceconfig.RetryPolicy{
MaxAttempts: jrp.MaxAttempts,
InitialBackoff: time.Duration(jrp.InitialBackoff),
MaxBackoff: time.Duration(jrp.MaxBackoff),
BackoffMultiplier: jrp.BackoffMultiplier,
RetryableStatusCodes: make(map[codes.Code]bool),
2018-11-26 18:23:56 +00:00
}
if rp.MaxAttempts > 5 {
2018-11-26 18:23:56 +00:00
// TODO(retry): Make the max maxAttempts configurable.
rp.MaxAttempts = 5
2018-11-26 18:23:56 +00:00
}
for _, code := range jrp.RetryableStatusCodes {
rp.RetryableStatusCodes[code] = true
2018-11-26 18:23:56 +00:00
}
return rp, nil
}
2018-01-09 18:57:14 +00:00
func min(a, b *int) *int {
if *a < *b {
return a
}
return b
}
func getMaxSize(mcMax, doptMax *int, defaultVal int) *int {
if mcMax == nil && doptMax == nil {
return &defaultVal
}
if mcMax != nil && doptMax != nil {
return min(mcMax, doptMax)
}
if mcMax != nil {
return mcMax
}
return doptMax
}
func newInt(b int) *int {
return &b
}
func init() {
internal.EqualServiceConfigForTesting = equalServiceConfig
}
// equalServiceConfig compares two configs. The rawJSONString field is ignored,
// because they may diff in white spaces.
//
// If any of them is NOT *ServiceConfig, return false.
func equalServiceConfig(a, b serviceconfig.Config) bool {
if a == nil && b == nil {
return true
}
aa, ok := a.(*ServiceConfig)
if !ok {
return false
}
bb, ok := b.(*ServiceConfig)
if !ok {
return false
}
aaRaw := aa.rawJSONString
aa.rawJSONString = ""
bbRaw := bb.rawJSONString
bb.rawJSONString = ""
defer func() {
aa.rawJSONString = aaRaw
bb.rawJSONString = bbRaw
}()
// Using reflect.DeepEqual instead of cmp.Equal because many balancer
// configs are unexported, and cmp.Equal cannot compare unexported fields
// from unexported structs.
return reflect.DeepEqual(aa, bb)
}