Merge branch 'master' into merge-atomic

nolouch · Apr 11, 2024 · 3637dee · 3637dee
2 parents b1275b1 + f0eb74b
commit 3637dee
Show file tree

Hide file tree

Showing 51 changed files with 1,056 additions and 150 deletions.
diff --git a/Makefile b/Makefile
@@ -80,7 +80,7 @@ BUILD_BIN_PATH := $(ROOT_PATH)/bin
 
 build: pd-server pd-ctl pd-recover
 
-tools: pd-tso-bench pd-heartbeat-bench regions-dump stores-dump pd-api-bench
+tools: pd-tso-bench pd-heartbeat-bench regions-dump stores-dump pd-api-bench pd-ut
 
 PD_SERVER_DEP :=
 ifeq ($(SWAGGER), 1)
@@ -108,7 +108,6 @@ pd-server-basic:
 .PHONY: pre-build build tools pd-server pd-server-basic
 
 # Tools
-
 pd-ctl:
  cd tools && GOEXPERIMENT=$(BUILD_GOEXPERIMENT) CGO_ENABLED=$(BUILD_TOOL_CGO_ENABLED) go build -gcflags '$(GCFLAGS)' -ldflags '$(LDFLAGS)' -o $(BUILD_BIN_PATH)/pd-ctl pd-ctl/main.go
 pd-tso-bench:
@@ -127,8 +126,12 @@ regions-dump:
  cd tools && CGO_ENABLED=0 go build -gcflags '$(GCFLAGS)' -ldflags '$(LDFLAGS)' -o $(BUILD_BIN_PATH)/regions-dump regions-dump/main.go
 stores-dump:
  cd tools && CGO_ENABLED=0 go build -gcflags '$(GCFLAGS)' -ldflags '$(LDFLAGS)' -o $(BUILD_BIN_PATH)/stores-dump stores-dump/main.go
+pd-ut: pd-xprog
+ cd tools && GOEXPERIMENT=$(BUILD_GOEXPERIMENT) CGO_ENABLED=$(BUILD_TOOL_CGO_ENABLED) go build -gcflags '$(GCFLAGS)' -ldflags '$(LDFLAGS)' -o $(BUILD_BIN_PATH)/pd-ut pd-ut/ut.go
+pd-xprog:
+ cd tools && GOEXPERIMENT=$(BUILD_GOEXPERIMENT) CGO_ENABLED=$(BUILD_TOOL_CGO_ENABLED) go build -tags xprog -gcflags '$(GCFLAGS)' -ldflags '$(LDFLAGS)' -o $(BUILD_BIN_PATH)/xprog pd-ut/xprog.go
 
-.PHONY: pd-ctl pd-tso-bench pd-recover pd-analysis pd-heartbeat-bench simulator regions-dump stores-dump pd-api-bench
+.PHONY: pd-ctl pd-tso-bench pd-recover pd-analysis pd-heartbeat-bench simulator regions-dump stores-dump pd-api-bench pd-ut
 
 #### Docker image ####
 
@@ -225,6 +228,12 @@ failpoint-disable: install-tools
 
 #### Test ####
 
+ut: pd-ut
+ @$(FAILPOINT_ENABLE)
+ ./bin/pd-ut run --race
+ @$(CLEAN_UT_BINARY)
+ @$(FAILPOINT_DISABLE)
+
 PACKAGE_DIRECTORIES := $(subst $(PD_PKG)/,,$(PACKAGES))
 TEST_PKGS := $(filter $(shell find . -iname "*_test.go" -exec dirname {} \; | \
  sort -u | sed -e "s/^\./github.com\/tikv\/pd/"),$(PACKAGES))
@@ -303,13 +312,16 @@ split:
 
 clean: failpoint-disable clean-test clean-build
 
+CLEAN_UT_BINARY := find . -name '*.test.bin'| xargs rm -f
+
 clean-test:
  # Cleaning test tmp...
  rm -rf /tmp/test_pd*
  rm -rf /tmp/pd-tests*
  rm -rf /tmp/test_etcd*
  rm -f $(REAL_CLUSTER_TEST_PATH)/playground.log
  go clean -testcache
+ @$(CLEAN_UT_BINARY)
 
 clean-build:
  # Cleaning building files...

diff --git a/client/resource_group/controller/config.go b/client/resource_group/controller/config.go
@@ -52,6 +52,10 @@ const (
  defaultTargetPeriod = 5 * time.Second
  // defaultMaxWaitDuration is the max duration to wait for the token before throwing error.
  defaultMaxWaitDuration = 30 * time.Second
+ // defaultWaitRetryTimes is the times to retry when waiting for the token.
+ defaultWaitRetryTimes = 10
+ // defaultWaitRetryInterval is the interval to retry when waiting for the token.
+ defaultWaitRetryInterval = 50 * time.Millisecond
 )
 
 const (
@@ -85,6 +89,12 @@ type Config struct {
  // LTBMaxWaitDuration is the max wait time duration for local token bucket.
  LTBMaxWaitDuration Duration `toml:"ltb-max-wait-duration" json:"ltb-max-wait-duration"`
 
+ // WaitRetryInterval is the interval to retry when waiting for the token.
+ WaitRetryInterval Duration `toml:"wait-retry-interval" json:"wait-retry-interval"`
+
+ // WaitRetryTimes is the times to retry when waiting for the token.
+ WaitRetryTimes int `toml:"wait-retry-times" json:"wait-retry-times"`
+
  // RequestUnit is the configuration determines the coefficients of the RRU and WRU cost.
  // This configuration should be modified carefully.
  RequestUnit RequestUnitConfig `toml:"request-unit" json:"request-unit"`
@@ -98,6 +108,8 @@ func DefaultConfig() *Config {
  return &Config{
  DegradedModeWaitDuration: NewDuration(defaultDegradedModeWaitDuration),
  LTBMaxWaitDuration: NewDuration(defaultMaxWaitDuration),
+ WaitRetryInterval: NewDuration(defaultWaitRetryInterval),
+ WaitRetryTimes: defaultWaitRetryTimes,
  RequestUnit: DefaultRequestUnitConfig(),
  EnableControllerTraceLog: false,
  }
@@ -155,6 +167,8 @@ type RUConfig struct {
 
  // some config for client
  LTBMaxWaitDuration time.Duration
+ WaitRetryInterval time.Duration
+ WaitRetryTimes int
  DegradedModeWaitDuration time.Duration
 }
 
@@ -176,6 +190,8 @@ func GenerateRUConfig(config *Config) *RUConfig {
  WriteBytesCost: RequestUnit(config.RequestUnit.WriteCostPerByte),
  CPUMsCost: RequestUnit(config.RequestUnit.CPUMsCost),
  LTBMaxWaitDuration: config.LTBMaxWaitDuration.Duration,
+ WaitRetryInterval: config.WaitRetryInterval.Duration,
+ WaitRetryTimes: config.WaitRetryTimes,
  DegradedModeWaitDuration: config.DegradedModeWaitDuration.Duration,
  }
 }
diff --git a/client/resource_group/controller/controller.go b/client/resource_group/controller/controller.go
@@ -39,8 +39,6 @@ import (
 
 const (
  controllerConfigPath = "resource_group/controller"
- maxRetry = 10
- retryInterval = 50 * time.Millisecond
  maxNotificationChanLen = 200
  needTokensAmplification = 1.1
  trickleReserveDuration = 1250 * time.Millisecond
@@ -105,6 +103,20 @@ func WithMaxWaitDuration(d time.Duration) ResourceControlCreateOption {
  }
 }
 
+// WithWaitRetryInterval is the option to set the retry interval when waiting for the token.
+func WithWaitRetryInterval(d time.Duration) ResourceControlCreateOption {
+ return func(controller *ResourceGroupsController) {
+ controller.ruConfig.WaitRetryInterval = d
+ }
+}
+
+// WithWaitRetryTimes is the option to set the times to retry when waiting for the token.
+func WithWaitRetryTimes(times int) ResourceControlCreateOption {
+ return func(controller *ResourceGroupsController) {
+ controller.ruConfig.WaitRetryTimes = times
+ }
+}
+
 var _ ResourceGroupKVInterceptor = (*ResourceGroupsController)(nil)
 
 // ResourceGroupsController implements ResourceGroupKVInterceptor.
@@ -186,7 +198,7 @@ func loadServerConfig(ctx context.Context, provider ResourceGroupProvider) (*Con
  log.Warn("[resource group controller] server does not save config, load config failed")
  return DefaultConfig(), nil
  }
- config := &Config{}
+ config := DefaultConfig()
  err = json.Unmarshal(kvs[0].GetValue(), config)
  if err != nil {
  return nil, err
@@ -367,7 +379,7 @@ func (c *ResourceGroupsController) Start(ctx context.Context) {
  }
  for _, item := range resp {
  cfgRevision = item.Kv.ModRevision
- config := &Config{}
+ config := DefaultConfig()
  if err := json.Unmarshal(item.Kv.Value, config); err != nil {
  continue
  }
@@ -1206,7 +1218,7 @@ func (gc *groupCostController) onRequestWait(
  var i int
  var d time.Duration
  retryLoop:
- for i = 0; i < maxRetry; i++ {
+ for i = 0; i < gc.mainCfg.WaitRetryTimes; i++ {
  switch gc.mode {
  case rmpb.GroupMode_RawMode:
  res := make([]*Reservation, 0, len(requestResourceLimitTypeList))
@@ -1230,8 +1242,8 @@ func (gc *groupCostController) onRequestWait(
  }
  }
  gc.requestRetryCounter.Inc()
- time.Sleep(retryInterval)
- waitDuration += retryInterval
+ time.Sleep(gc.mainCfg.WaitRetryInterval)
+ waitDuration += gc.mainCfg.WaitRetryInterval
  }
  if err != nil {
  gc.failedRequestCounter.Inc()

diff --git a/client/resource_group/controller/util_test.go b/client/resource_group/controller/util_test.go
@@ -27,7 +27,6 @@ type example struct {
 }
 
 func TestDurationJSON(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  example := &example{}
 
@@ -41,7 +40,6 @@ func TestDurationJSON(t *testing.T) {
 }
 
 func TestDurationTOML(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  example := &example{}
 

diff --git a/pkg/audit/audit_test.go b/pkg/audit/audit_test.go
@@ -32,7 +32,6 @@ import (
 )
 
 func TestLabelMatcher(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  matcher := &LabelMatcher{"testSuccess"}
  labels1 := &BackendLabels{Labels: []string{"testFail", "testSuccess"}}
@@ -42,7 +41,6 @@ func TestLabelMatcher(t *testing.T) {
 }
 
 func TestPrometheusHistogramBackend(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  serviceAuditHistogramTest := prometheus.NewHistogramVec(
  prometheus.HistogramOpts{
@@ -90,7 +88,6 @@ func TestPrometheusHistogramBackend(t *testing.T) {
 }
 
 func TestLocalLogBackendUsingFile(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  backend := NewLocalLogBackend(true)
  fname := testutil.InitTempFileLogger("info")

diff --git a/pkg/autoscaling/calculation_test.go b/pkg/autoscaling/calculation_test.go
@@ -29,7 +29,6 @@ import (
 )
 
 func TestGetScaledTiKVGroups(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  ctx, cancel := context.WithCancel(context.Background())
  defer cancel()
@@ -214,7 +213,6 @@ func (q *mockQuerier) Query(options *QueryOptions) (QueryResult, error) {
 }
 
 func TestGetTotalCPUUseTime(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  querier := &mockQuerier{}
  instances := []instance{
@@ -237,7 +235,6 @@ func TestGetTotalCPUUseTime(t *testing.T) {
 }
 
 func TestGetTotalCPUQuota(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  querier := &mockQuerier{}
  instances := []instance{
@@ -260,7 +257,6 @@ func TestGetTotalCPUQuota(t *testing.T) {
 }
 
 func TestScaleOutGroupLabel(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  var jsonStr = []byte(`
 {
@@ -303,7 +299,6 @@ func TestScaleOutGroupLabel(t *testing.T) {
 }
 
 func TestStrategyChangeCount(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  var count uint64 = 2
  strategy := &Strategy{

diff --git a/pkg/autoscaling/prometheus_test.go b/pkg/autoscaling/prometheus_test.go
@@ -180,7 +180,6 @@ func (c *normalClient) Do(_ context.Context, req *http.Request) (response *http.
 }
 
 func TestRetrieveCPUMetrics(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  client := &normalClient{
  mockData: make(map[string]*response),
@@ -225,7 +224,6 @@ func (c *emptyResponseClient) Do(_ context.Context, req *http.Request) (r *http.
 }
 
 func TestEmptyResponse(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  client := &emptyResponseClient{}
  querier := NewPrometheusQuerier(client)
@@ -253,7 +251,6 @@ func (c *errorHTTPStatusClient) Do(_ context.Context, req *http.Request) (r *htt
 }
 
 func TestErrorHTTPStatus(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  client := &errorHTTPStatusClient{}
  querier := NewPrometheusQuerier(client)
@@ -279,7 +276,6 @@ func (c *errorPrometheusStatusClient) Do(_ context.Context, req *http.Request) (
 }
 
 func TestErrorPrometheusStatus(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  client := &errorPrometheusStatusClient{}
  querier := NewPrometheusQuerier(client)
@@ -290,7 +286,6 @@ func TestErrorPrometheusStatus(t *testing.T) {
 }
 
 func TestGetInstanceNameFromAddress(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  testCases := []struct {
  address string
@@ -328,7 +323,6 @@ func TestGetInstanceNameFromAddress(t *testing.T) {
 }
 
 func TestGetDurationExpression(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  testCases := []struct {
  duration time.Duration

diff --git a/pkg/balancer/balancer_test.go b/pkg/balancer/balancer_test.go
@@ -22,7 +22,6 @@ import (
 )
 
 func TestBalancerPutAndDelete(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  balancers := []Balancer[uint32]{
  NewRoundRobin[uint32](),
@@ -56,7 +55,6 @@ func TestBalancerPutAndDelete(t *testing.T) {
 }
 
 func TestBalancerDuplicate(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  balancers := []Balancer[uint32]{
  NewRoundRobin[uint32](),
@@ -77,7 +75,6 @@ func TestBalancerDuplicate(t *testing.T) {
 }
 
 func TestRoundRobin(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  balancer := NewRoundRobin[uint32]()
  for i := 0; i < 100; i++ {

diff --git a/pkg/cache/cache_test.go b/pkg/cache/cache_test.go
@@ -25,7 +25,6 @@ import (
 )
 
 func TestExpireRegionCache(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  ctx, cancel := context.WithCancel(context.Background())
  defer cancel()
@@ -121,7 +120,6 @@ func sortIDs(ids []uint64) []uint64 {
 }
 
 func TestLRUCache(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  cache := newLRU(3)
 
@@ -199,7 +197,6 @@ func TestLRUCache(t *testing.T) {
 }
 
 func TestFifoCache(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  cache := NewFIFO(3)
  cache.Put(1, "1")
@@ -227,7 +224,6 @@ func TestFifoCache(t *testing.T) {
 }
 
 func TestFifoFromLastSameElems(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  type testStruct struct {
  value string
@@ -260,7 +256,6 @@ func TestFifoFromLastSameElems(t *testing.T) {
 }
 
 func TestTwoQueueCache(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  cache := newTwoQueue(3)
  cache.Put(1, "1")
@@ -345,7 +340,6 @@ func (pq PriorityQueueItemTest) ID() uint64 {
 }
 
 func TestPriorityQueue(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  testData := []PriorityQueueItemTest{0, 1, 2, 3, 4, 5}
  pq := NewPriorityQueue(0)

diff --git a/pkg/codec/codec_test.go b/pkg/codec/codec_test.go
@@ -21,7 +21,6 @@ import (
 )
 
 func TestDecodeBytes(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  key := "abcdefghijklmnopqrstuvwxyz"
  for i := 0; i < len(key); i++ {
@@ -32,7 +31,6 @@ func TestDecodeBytes(t *testing.T) {
 }
 
 func TestTableID(t *testing.T) {
- t.Parallel()
  re := require.New(t)
  key := EncodeBytes([]byte("t\x80\x00\x00\x00\x00\x00\x00\xff"))
  re.Equal(int64(0xff), key.TableID())