sync upstream

Author: yandex-cloud-bot

pkg/retry/v1/config.go

@@ -0,0 +1,107 @@
+package retry
+
+import (
+	"strings"
+	"time"
+
+	"google.golang.org/grpc/codes"
+)
+
+//revive:enable:var-naming
+
+type RetryConfig struct {
+	mc map[nameConfig]*methodConfig
+}
+
+type RetryOption func(c *RetryConfig)
+
+type grpcRetryPolicy struct {
+	MethodConfig []*methodConfig `json:"methodConfig"`
+}
+
+type methodConfig struct {
+	NameConfig      []nameConfig    `json:"name"`
+	RetryPolicy     retryPolicy     `json:"retryPolicy"`
+	RetryThrottling retryThrottling `json:"retryThrottling"`
+	WaitForReady    bool            `json:"waitForReady"`
+}
+
+type nameConfig struct {
+	Service string `json:"service,omitempty"`
+	Method  string `json:"method,omitempty"`
+}
+
+type retryPolicy struct {
+	MaxAttempts          int      `json:"maxAttempts"`
+	InitialBackoff       Duration `json:"initialBackoff"`
+	MaxBackoff           Duration `json:"maxBackoff"`
+	BackoffMultiplier    float64  `json:"backoffMultiplier"`
+	RetryableStatusCodes []string `json:"retryableStatusCodes"`
+}
+
+type retryThrottling struct {
+	MaxTokens  int     `json:"maxTokens"`
+	TokenRatio float64 `json:"tokenRatio"`
+}
+
+type GRPCKeepAliveConfig struct {
+	Time                time.Duration `yaml:"time" validate:"required"`
+	Timeout             time.Duration `yaml:"timeout" validate:"required"`
+	PermitWithoutStream bool          `yaml:"permit_without_stream"`
+}
+
+func DefaultRetryConfig() *RetryConfig {
+	return &RetryConfig{}
+}
+
+func DefaultNameConfig() nameConfig {
+	return nameConfig{}
+}
+
+func NewNameConfig(service, method string) nameConfig {
+	return nameConfig{
+		Service: service,
+		Method:  method,
+	}
+}
+
+func WithDefaultRetryConfig() RetryOption {
+	return func(c *RetryConfig) {
+		c = defaultRetryConfig()
+	}
+}
+
+func WithRetries(nm nameConfig, n int) RetryOption {
+	return func(c *RetryConfig) {
+		if c == nil {
+			c = defaultRetryConfig()
+		}
+
+		if _, ok := c.mc[nm]; !ok {
+			c.mc[nm] = defaultMethodConfig()
+			c.mc[nm].NameConfig = []nameConfig{nm}
+		}
+
+		c.mc[nm].RetryPolicy.MaxAttempts = n
+	}
+}
+
+func WithRetryableStatusCodes(nm nameConfig, codes ...codes.Code) RetryOption {
+	return func(c *RetryConfig) {
+		if c == nil {
+			c = defaultRetryConfig()
+		}
+
+		names := make([]string, len(codes))
+		for i, code := range codes {
+			names[i] = strings.ToUpper(code.String())
+		}
+
+		if _, ok := c.mc[nm]; !ok {
+			c.mc[nm] = defaultMethodConfig()
+			c.mc[nm].NameConfig = []nameConfig{nm}
+		}
+
+		c.mc[nm].RetryPolicy.RetryableStatusCodes = names
+	}
+}

pkg/retry/v1/default.go

@@ -0,0 +1,34 @@
+package retry
+
+import "time"
+
+func defaultNameConfig() nameConfig {
+	return nameConfig{}
+}
+
+func defaultRetryConfig() *RetryConfig {
+	return &RetryConfig{
+		mc: map[nameConfig]*methodConfig{
+			// default value for all services and methods
+			defaultNameConfig(): defaultMethodConfig(),
+		},
+	}
+}
+
+func defaultMethodConfig() *methodConfig {
+	return &methodConfig{
+		NameConfig: []nameConfig{{}},
+		RetryPolicy: retryPolicy{
+			MaxAttempts:          4,
+			InitialBackoff:       Duration(time.Millisecond * 100),
+			MaxBackoff:           Duration(time.Second * 20),
+			BackoffMultiplier:    2,
+			RetryableStatusCodes: []string{"UNAVAILABLE"},
+		},
+		RetryThrottling: retryThrottling{
+			MaxTokens:  100,
+			TokenRatio: 0.1,
+		},
+		WaitForReady: true,
+	}
+}

pkg/retry/v1/dial_option.go

@@ -0,0 +1,54 @@
+package retry
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"google.golang.org/grpc"
+)
+
+// RetryDialOption returns DialOption that configures the default
+// service config, which will be used in cases where:
+//
+// 1. User provide service config in SDK build after RetryDialOption function.
+// In this case retries configuration override user's service config.
+//
+// 2. User don't get service config through xDS. In that case service config
+// from xDS override retries configuration.
+//
+// It's recommended to use default configuration (DefaultRetryDialOption)
+// for retries in SDK to avoid retry amplification.
+func RetryDialOption(opts ...RetryOption) (grpc.DialOption, error) {
+	config := defaultRetryConfig()
+	for _, opt := range opts {
+		opt(config)
+	}
+
+	if config.mc == nil {
+		return nil, fmt.Errorf("can't provide retry config with this options")
+	}
+
+	mc := []*methodConfig{}
+	for _, v := range config.mc {
+		mc = append(mc, v)
+	}
+
+	c, err := json.Marshal(&grpcRetryPolicy{MethodConfig: mc})
+	if err != nil {
+		return nil, err
+	}
+
+	return grpc.WithDefaultServiceConfig(string(c)), nil
+}
+
+// DefaultRetryDialOption returns DialOption that configures the default
+// service config, which will be used in cases where:
+//
+// 1. User provide service config in SDK build after DefaultRetryDialOption function.
+// In this case retries configuration override user's service config.
+//
+// 2. User don't get service config through xDS. In that case service config
+// from xDS override retries configuration.
+func DefaultRetryDialOption() (grpc.DialOption, error) {
+	return RetryDialOption(WithDefaultRetryConfig())
+}

pkg/retry/v1/duration.go

@@ -0,0 +1,37 @@
+package retry
+
+import (
+	"fmt"
+	"strings"
+	"time"
+)
+
+// Duration defines JSON marshal and unmarshal methods to conform to the
+// protobuf JSON spec defined [here].
+//
+// [here]: https://protobuf.dev/reference/protobuf/google.protobuf/#duration
+type Duration time.Duration
+
+func (d Duration) String() string {
+	return fmt.Sprint(time.Duration(d))
+}
+
+// MarshalJSON converts from d to a JSON string output.
+func (d Duration) MarshalJSON() ([]byte, error) {
+	ns := time.Duration(d).Nanoseconds()
+	sec := ns / int64(time.Second)
+	ns = ns % int64(time.Second)
+
+	var sign string
+	if sec < 0 || ns < 0 {
+		sign, sec, ns = "-", -1*sec, -1*ns
+	}
+
+	// Generated output always contains 0, 3, 6, or 9 fractional digits,
+	// depending on required precision.
+	str := fmt.Sprintf("%s%d.%09d", sign, sec, ns)
+	str = strings.TrimSuffix(str, "000")
+	str = strings.TrimSuffix(str, "000")
+	str = strings.TrimSuffix(str, ".000")
+	return []byte(fmt.Sprintf("\"%ss\"", str)), nil
+}