policy.go

/*
 * Copyright 2021 CloudWeGo 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 retry

import (
	"context"
	"fmt"

	"github.com/cloudwego/kitex/pkg/klog"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
)

// Type is retry type include FailureType, BackupType
type Type int

// retry types
const (
	FailureType Type = iota
	BackupType
	MixedType
)

// String prints human readable information.
func (t Type) String() string {
	switch t {
	case FailureType:
		return "Failure"
	case BackupType:
		return "Backup"
	case MixedType:
		return "Mixed"
	}
	return ""
}

// BuildFailurePolicy is used to build Policy with *FailurePolicy
func BuildFailurePolicy(p *FailurePolicy) Policy {
	if p == nil {
		return Policy{}
	}
	return Policy{Enable: true, Type: FailureType, FailurePolicy: p}
}

// BuildBackupRequest is used to build Policy with *BackupPolicy
func BuildBackupRequest(p *BackupPolicy) Policy {
	if p == nil {
		return Policy{}
	}
	return Policy{Enable: true, Type: BackupType, BackupPolicy: p}
}

// BuildMixedPolicy is used to build Policy with *MixedPolicy
func BuildMixedPolicy(p *MixedPolicy) Policy {
	if p == nil {
		return Policy{}
	}
	return Policy{Enable: true, Type: MixedType, MixedPolicy: p}
}

// Policy contains all retry policies
// DON'T FORGET to update Equals() and DeepCopy() if you add new fields
type Policy struct {
	Enable bool `json:"enable"`
	// 0 is failure retry, 1 is backup
	Type Type `json:"type"`
	// notice: only one retry policy can be enabled, which one depend on Policy.Type
	FailurePolicy *FailurePolicy `json:"failure_policy,omitempty"`
	BackupPolicy  *BackupPolicy  `json:"backup_policy,omitempty"`
	MixedPolicy   *MixedPolicy   `json:"mixed_policy,omitempty"`
}

func (p *Policy) DeepCopy() *Policy {
	if p == nil {
		return nil
	}
	return &Policy{
		Enable:        p.Enable,
		Type:          p.Type,
		FailurePolicy: p.FailurePolicy.DeepCopy(),
		BackupPolicy:  p.BackupPolicy.DeepCopy(),
		MixedPolicy:   p.MixedPolicy.DeepCopy(),
	}
}

// FailurePolicy for failure retry
// DON'T FORGET to update Equals() and DeepCopy() if you add new fields
type FailurePolicy struct {
	StopPolicy        StopPolicy         `json:"stop_policy"`
	BackOffPolicy     *BackOffPolicy     `json:"backoff_policy,omitempty"`
	RetrySameNode     bool               `json:"retry_same_node"`
	ShouldResultRetry *ShouldResultRetry `json:"-"`

	// Extra is not used directly by kitex. It's used for better integrating your own config source.
	// After loading FailurePolicy from your config source, `Extra` can be decoded into a user-defined schema,
	// with which, more complex strategies can be implemented, such as modifying the `ShouldResultRetry`.
	Extra string `json:"extra"`
}

// BackupPolicy for backup request
// DON'T FORGET to update Equals() and DeepCopy() if you add new fields
type BackupPolicy struct {
	RetryDelayMS  uint32     `json:"retry_delay_ms"`
	StopPolicy    StopPolicy `json:"stop_policy"`
	RetrySameNode bool       `json:"retry_same_node"`
}

// MixedPolicy for failure retry
// DON'T FORGET to update Equals() and DeepCopy() if you add new fields
type MixedPolicy struct {
	RetryDelayMS uint32 `json:"retry_delay_ms"`
	FailurePolicy
}

// StopPolicy is a group policies to decide when stop retry
type StopPolicy struct {
	MaxRetryTimes    int      `json:"max_retry_times"`
	MaxDurationMS    uint32   `json:"max_duration_ms"`
	DisableChainStop bool     `json:"disable_chain_stop"`
	DDLStop          bool     `json:"ddl_stop"`
	CBPolicy         CBPolicy `json:"cb_policy"`
}

const (
	defaultCBErrRate = 0.1
	cbMinSample      = 10
)

// CBPolicy is the circuit breaker policy
type CBPolicy struct {
	ErrorRate float64 `json:"error_rate"`
}

// BackOffPolicy is the BackOff policy.
// DON'T FORGET to update Equals() and DeepCopy() if you add new fields
type BackOffPolicy struct {
	BackOffType BackOffType               `json:"backoff_type"`
	CfgItems    map[BackOffCfgKey]float64 `json:"cfg_items,omitempty"`
}

// BackOffType means the BackOff type.
type BackOffType string

// all back off types
const (
	NoneBackOffType   BackOffType = "none"
	FixedBackOffType  BackOffType = "fixed"
	RandomBackOffType BackOffType = "random"
)

// BackOffCfgKey represents the keys for BackOff.
type BackOffCfgKey string

// the keys of all back off configs
const (
	FixMSBackOffCfgKey      BackOffCfgKey = "fix_ms"
	MinMSBackOffCfgKey      BackOffCfgKey = "min_ms"
	MaxMSBackOffCfgKey      BackOffCfgKey = "max_ms"
	InitialMSBackOffCfgKey  BackOffCfgKey = "initial_ms"
	MultiplierBackOffCfgKey BackOffCfgKey = "multiplier"
)

// ShouldResultRetry is used for specifying which error or resp need to be retried
type ShouldResultRetry struct {
	// ErrorRetryWithCtx is added in v0.10.0, passing ctx is more convenient for user
	ErrorRetryWithCtx func(ctx context.Context, err error, ri rpcinfo.RPCInfo) bool
	// RespRetryWithCtx is added in v0.10.0, passing ctx is more convenient for user
	RespRetryWithCtx func(ctx context.Context, resp interface{}, ri rpcinfo.RPCInfo) bool

	// Deprecated: please use ErrorRetryWithCtx instead of ErrorRetry
	ErrorRetry func(err error, ri rpcinfo.RPCInfo) bool
	// Deprecated: please use RespRetryWithCtx instead of RespRetry
	RespRetry func(resp interface{}, ri rpcinfo.RPCInfo) bool

	// disable the default timeout retry in specific scenarios (e.g. the requests are not non-idempotent)
	NotRetryForTimeout bool
}

// Equals to check if policy is equal
func (p Policy) Equals(np Policy) bool {
	if p.Enable != np.Enable {
		return false
	}
	if p.Type != np.Type {
		return false
	}
	if !p.FailurePolicy.Equals(np.FailurePolicy) {
		return false
	}
	if !p.BackupPolicy.Equals(np.BackupPolicy) {
		return false
	}
	if !p.MixedPolicy.Equals(np.MixedPolicy) {
		return false
	}
	return true
}

// Equals to check if BackOffPolicy is equal.
func (p *BackOffPolicy) Equals(np *BackOffPolicy) bool {
	if p == nil {
		return np == nil
	}
	if np == nil {
		return false
	}
	if p.BackOffType != np.BackOffType {
		return false
	}
	if len(p.CfgItems) != len(np.CfgItems) {
		return false
	}
	for k := range p.CfgItems {
		if p.CfgItems[k] != np.CfgItems[k] {
			return false
		}
	}

	return true
}

func (p *BackOffPolicy) DeepCopy() *BackOffPolicy {
	if p == nil {
		return nil
	}
	return &BackOffPolicy{
		BackOffType: p.BackOffType,
		CfgItems:    p.copyCfgItems(),
	}
}

func (p *BackOffPolicy) copyCfgItems() map[BackOffCfgKey]float64 {
	if p.CfgItems == nil {
		return nil
	}
	cfgItems := make(map[BackOffCfgKey]float64, len(p.CfgItems))
	for k, v := range p.CfgItems {
		cfgItems[k] = v
	}
	return cfgItems
}

func (rr *ShouldResultRetry) IsValid() bool {
	return rr.ErrorRetryWithCtx != nil || rr.RespRetryWithCtx != nil || rr.RespRetry != nil || rr.ErrorRetry != nil
}

func checkCBErrorRate(p *CBPolicy) error {
	if p.ErrorRate <= 0 || p.ErrorRate > 0.3 {
		return fmt.Errorf("invalid retry circuit breaker rate, errRate=%0.2f", p.ErrorRate)
	}
	return nil
}

func checkStopPolicy(sp *StopPolicy, maxRetryTimes int, retryer Retryer) error {
	rt := sp.MaxRetryTimes
	// 0 is valid, it means stop retry
	if rt < 0 || rt > maxRetryTimes {
		return fmt.Errorf("invalid MaxRetryTimes[%d]", rt)
	}
	if e := checkCBErrorRate(&sp.CBPolicy); e != nil {
		sp.CBPolicy.ErrorRate = defaultCBErrRate
		klog.Warnf("KITEX: %s retry - %s, use default %0.2f", retryer.Type(), e.Error(), defaultCBErrRate)
	}
	return nil
}