*: integrate circuitbreaker for get region calls to PD (#58737)

close #58780

pkg/domain/BUILD.bazel

@@ -111,6 +111,7 @@ go_library(
         "@com_github_tikv_pd_client//:client",
         "@com_github_tikv_pd_client//http",
         "@com_github_tikv_pd_client//opt",
+        "@com_github_tikv_pd_client//pkg/circuitbreaker",
         "@com_github_tikv_pd_client//resource_group/controller",
         "@io_etcd_go_etcd_client_v3//:client",
         "@io_etcd_go_etcd_client_v3//concurrency",

pkg/domain/domain_sysvars.go

@@ -23,8 +23,10 @@ import (
 	"github.com/pingcap/tidb/pkg/meta"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
+	"github.com/tikv/client-go/v2/tikv"
 	pd "github.com/tikv/pd/client"
 	"github.com/tikv/pd/client/opt"
+	"github.com/tikv/pd/client/pkg/circuitbreaker"
 )
 
 // initDomainSysVars() is called when a domain is initialized.
@@ -46,6 +48,8 @@ func (do *Domain) initDomainSysVars() {
 	variable.SetLowResolutionTSOUpdateInterval = do.setLowResolutionTSOUpdateInterval
 
 	variable.ChangeSchemaCacheSize = do.changeSchemaCacheSize
+
+	variable.ChangePDMetadataCircuitBreakerErrorRateThresholdPct = changePDMetadataCircuitBreakerErrorRateThresholdPct
 }
 
 // setStatsCacheCapacity sets statsCache cap
@@ -151,3 +155,9 @@ func (do *Domain) changeSchemaCacheSize(ctx context.Context, size uint64) error
 	do.infoCache.Data.SetCacheCapacity(size)
 	return nil
 }
+
+func changePDMetadataCircuitBreakerErrorRateThresholdPct(errorRatePct uint32) {
+	tikv.ChangePDRegionMetaCircuitBreakerSettings(func(config *circuitbreaker.Settings) {
+		config.ErrorRateThresholdPct = errorRatePct
+	})
+}

pkg/sessionctx/vardef/tidb_vars.go

@@ -1216,6 +1216,9 @@ const (
 	// TiDBTSOClientRPCMode controls how the TSO client performs the TSO RPC requests. It internally controls the
 	// concurrency of the RPC. This variable provides an approach to tune the latency of getting timestamps from PD.
 	TiDBTSOClientRPCMode = "tidb_tso_client_rpc_mode"
+	// TiDBCircuitBreakerPDMetadataErrorRateThresholdPct variable is used to set percent of errors to trip the circuit breaker for get region calls to PD
+	// https://github.com/tikv/rfcs/blob/master/text/0115-circuit-breaker.md
+	TiDBCircuitBreakerPDMetadataErrorRateThresholdPct = "tidb_cb_pd_metadata_error_rate_threshold_pct"
 )
 
 // TiDB intentional limits, can be raised in the future.
@@ -1567,6 +1570,7 @@ const (
 	DefOptEnableProjectionPushDown                    = true
 	DefTiDBEnableSharedLockPromotion                  = false
 	DefTiDBTSOClientRPCMode                           = TSOClientRPCModeDefault
+	DefTiDBCircuitBreakerPDMetaErrorRatePct           = 0
 )
 
 // Process global variables.

pkg/sessionctx/variable/sysvar.go

@@ -3448,6 +3448,14 @@ var defaultSysVars = []*SysVar{
 			return (*SetPDClientDynamicOption.Load())(vardef.TiDBTSOClientRPCMode, val)
 		},
 	},
+	{Scope: vardef.ScopeGlobal, Name: vardef.TiDBCircuitBreakerPDMetadataErrorRateThresholdPct, Value: strconv.Itoa(vardef.DefTiDBCircuitBreakerPDMetaErrorRatePct), Type: vardef.TypeUnsigned, MinValue: 0, MaxValue: 100,
+		SetGlobal: func(_ context.Context, s *SessionVars, val string) error {
+			if ChangePDMetadataCircuitBreakerErrorRateThresholdPct != nil {
+				ChangePDMetadataCircuitBreakerErrorRateThresholdPct(uint32(tidbOptPositiveInt32(val, vardef.DefTiDBCircuitBreakerPDMetaErrorRatePct)))
+			}
+			return nil
+		},
+	},
 }
 
 // GlobalSystemVariableInitialValue gets the default value for a system variable including ones that are dynamically set (e.g. based on the store)

pkg/sessionctx/variable/sysvar_test.go

@@ -1743,6 +1743,30 @@ func TestTiDBSchemaCacheSize(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestTiDBCircuitBreakerPDMetadataErrorRateThresholdPct(t *testing.T) {
+	sv := GetSysVar(vardef.TiDBCircuitBreakerPDMetadataErrorRateThresholdPct)
+	vars := NewSessionVars(nil)
+
+	// Too low, will get raised to the min value
+	val, err := sv.Validate(vars, "-1", vardef.ScopeGlobal)
+	require.NoError(t, err)
+	require.Equal(t, strconv.FormatInt(GetSysVar(vardef.TiDBCircuitBreakerPDMetadataErrorRateThresholdPct).MinValue, 10), val)
+	warn := vars.StmtCtx.GetWarnings()[0].Err
+	require.Equal(t, "[variable:1292]Truncated incorrect tidb_cb_pd_metadata_error_rate_threshold_pct value: '-1'", warn.Error())
+
+	// Too high, will get lowered to the max value
+	val, err = sv.Validate(vars, "101", vardef.ScopeGlobal)
+	require.NoError(t, err)
+	require.Equal(t, strconv.FormatUint(GetSysVar(vardef.TiDBCircuitBreakerPDMetadataErrorRateThresholdPct).MaxValue, 10), val)
+	warn = vars.StmtCtx.GetWarnings()[1].Err
+	require.Equal(t, "[variable:1292]Truncated incorrect tidb_cb_pd_metadata_error_rate_threshold_pct value: '101'", warn.Error())
+
+	// valid
+	val, err = sv.Validate(vars, "10", vardef.ScopeGlobal)
+	require.NoError(t, err)
+	require.Equal(t, "10", val)
+}
+
 func TestEnableWindowFunction(t *testing.T) {
 	vars := NewSessionVars(nil)
 	require.Equal(t, vars.EnableWindowFunction, vardef.DefEnableWindowFunction)

pkg/sessionctx/variable/tidb_vars.go

@@ -54,6 +54,8 @@ var (
 	EnableStatsOwner func() error = nil
 	// DisableStatsOwner is the func registered by stats to disable running stats in this instance.
 	DisableStatsOwner func() error = nil
+	// ChangePDMetadataCircuitBreakerErrorRateThresholdPct changes the error rate threshold of the PD metadata circuit breaker.
+	ChangePDMetadataCircuitBreakerErrorRateThresholdPct func(uint32) = nil
 )
 
 // Hooks functions for Cluster Resource Control.