server/handler: register missing http handlers related to ingest param (#63099) (#63121)

ref #61553

Author: ti-chi-bot

br/pkg/lightning/backend/local/rate_limiter_param.go

@@ -46,33 +46,29 @@ var (
 // InitializeRateLimiterParam initializes the rate limiter params.
 func InitializeRateLimiterParam(m *meta.Meta, logger *zap.Logger) error {
 	err := initializeVariables(
-		m.GetIngestMaxBatchSplitRanges,
-		defaultMaxBatchSplitRanges,
-		&CurrentMaxBatchSplitRanges,
+		m.GetIngestMaxBatchSplitRanges, m.SetIngestMaxBatchSplitRanges,
+		defaultMaxBatchSplitRanges, &CurrentMaxBatchSplitRanges,
 		logger, "maxBatchSplitRanges")
 	if err != nil {
 		return err
 	}
 	err = initializeVariables(
-		m.GetIngestMaxSplitRangesPerSec,
-		defaultSplitRangesPerSec,
-		&CurrentMaxSplitRangesPerSec,
+		m.GetIngestMaxSplitRangesPerSec, m.SetIngestMaxSplitRangesPerSec,
+		defaultSplitRangesPerSec, &CurrentMaxSplitRangesPerSec,
 		logger, "maxSplitRangesPerSec")
 	if err != nil {
 		return err
 	}
 	err = initializeVariables(
-		m.GetIngestMaxInflight,
-		defaultMaxIngestInflight,
-		&CurrentMaxIngestInflight,
+		m.GetIngestMaxInflight, m.SetIngestMaxInflight,
+		defaultMaxIngestInflight, &CurrentMaxIngestInflight,
 		logger, "maxIngestInflight")
 	if err != nil {
 		return err
 	}
 	err = initializeVariables(
-		m.GetIngestMaxPerSec,
-		defaultMaxIngestPerSec,
-		&CurrentMaxIngestPerSec,
+		m.GetIngestMaxPerSec, m.SetIngestMaxPerSec,
+		defaultMaxIngestPerSec, &CurrentMaxIngestPerSec,
 		logger, "maxIngestPerSec")
 	if err != nil {
 		return err
@@ -82,6 +78,7 @@ func InitializeRateLimiterParam(m *meta.Meta, logger *zap.Logger) error {
 
 func initializeVariables[T comparable](
 	metaGetter func() (v T, isNull bool, err error),
+	metaSetter func(v T) error,
 	defaultVal T,
 	globalVar *atomic.Pointer[T],
 	logger *zap.Logger,
@@ -92,11 +89,17 @@ func initializeVariables[T comparable](
 		return errors.Annotatef(err, "failed to read %s value from meta store", varName)
 	}
 	var zero T
-	if isNull || val == zero {
+	if isNull {
+		err = metaSetter(defaultVal)
+		if err != nil {
+			return errors.Annotatef(err, "failed to set %s value to meta store", varName)
+		}
 		val = defaultVal
 		logger.Info("meta kv not found in meta store, initialized to default and persisted",
 			zap.String("key", varName),
 			zap.Any("value", defaultVal))
+	} else if val == zero {
+		val = defaultVal
 	} else {
 		logger.Info("loaded value from meta store",
 			zap.String("key", varName),

pkg/server/handler/tests/BUILD.bazel

@@ -10,7 +10,7 @@ go_test(
     ],
     flaky = True,
     race = "on",
-    shard_count = 38,
+    shard_count = 39,
     deps = [
         "//pkg/config",
         "//pkg/ddl",

pkg/server/handler/tests/http_handler_serial_test.go

@@ -684,3 +684,53 @@ func TestTTL(t *testing.T) {
 	require.Nil(t, obj)
 	require.EqualError(t, err, "http status: 400 Bad Request, table test_ttl.t2 not exists")
 }
+
+func TestIngestParam(t *testing.T) {
+	ts := createBasicHTTPHandlerTestSuite()
+	ts.startServer(t)
+	defer ts.stopServer(t)
+
+	testCases := []struct {
+		url         string
+		defaultVal  any
+		modVal      any
+		expectedVal any
+	}{
+		{"/ingest/max-batch-split-ranges", float64(2048), 1000, float64(1000)},
+		{"/ingest/max-split-ranges-per-sec", float64(0), 2000, float64(2000)},
+		{"/ingest/max-ingest-inflight", float64(0), 1000, float64(1000)},
+		{"/ingest/max-ingest-per-sec", float64(0), 2000, float64(2000)},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.url, func(t *testing.T) {
+			resp, err := ts.FetchStatus(tc.url)
+			require.NoError(t, err)
+			defer func() { require.NoError(t, resp.Body.Close()) }()
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+			decoder := json.NewDecoder(resp.Body)
+			var payload struct {
+				Value  float64 `json:"value"`
+				IsNull bool    `json:"is_null"`
+			}
+			err = decoder.Decode(&payload)
+			require.NoError(t, err)
+			require.Equal(t, tc.defaultVal, payload.Value)
+
+			resp, err = ts.PostStatus(tc.url, "", bytes.NewBuffer([]byte(fmt.Sprintf(`{"value": %v}`, tc.modVal))))
+			require.NoError(t, err)
+			require.NotNil(t, resp)
+			defer func() { require.NoError(t, resp.Body.Close()) }()
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+
+			resp, err = ts.FetchStatus(tc.url)
+			require.NoError(t, err)
+			defer func() { require.NoError(t, resp.Body.Close()) }()
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+			decoder = json.NewDecoder(resp.Body)
+			err = decoder.Decode(&payload)
+			require.NoError(t, err)
+			require.Equal(t, tc.expectedVal, payload.Value)
+		})
+	}
+}

pkg/server/handler/tikvhandler/tikv_handler.go

@@ -2114,11 +2114,12 @@ func (h IngestConcurrencyHandler) ServeHTTP(w http.ResponseWriter, req *http.Req
 	default:
 		handler.WriteError(w, errors.Errorf("unsupported ingest parameter: %s", h.param))
 	}
+	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnOthers)
 	switch req.Method {
 	case http.MethodGet:
 		var respValue float64
 		var respIsNull bool
-		err := kv.RunInNewTxn(context.Background(), h.Store.(kv.Storage), false, func(_ context.Context, txn kv.Transaction) error {
+		err := kv.RunInNewTxn(ctx, h.Store.(kv.Storage), false, func(_ context.Context, txn kv.Transaction) error {
 			m := meta.NewMeta(txn)
 			var getErr error
 			respValue, respIsNull, getErr = getter(m)
@@ -2148,7 +2149,7 @@ func (h IngestConcurrencyHandler) ServeHTTP(w http.ResponseWriter, req *http.Req
 			handler.WriteError(w, errors.New("value must be >= 0"))
 			return
 		}
-		err := kv.RunInNewTxn(context.Background(), h.Store.(kv.Storage), true, func(_ context.Context, txn kv.Transaction) error {
+		err := kv.RunInNewTxn(ctx, h.Store.(kv.Storage), true, func(_ context.Context, txn kv.Transaction) error {
 			m := meta.NewMeta(txn)
 			return setter(m, newValue)
 		})

pkg/server/http_status.go

@@ -257,6 +257,16 @@ func (s *Server) startHTTPServer() {
 	// HTTP path for upgrade operations.
 	router.Handle("/upgrade/{op}", handler.NewClusterUpgradeHandler(tikvHandlerTool.Store.(kv.Storage))).Name("upgrade operations")
 
+	// HTTP path for ingest configurations
+	router.Handle("/ingest/max-batch-split-ranges", tikvhandler.NewIngestConcurrencyHandler(
+		tikvHandlerTool, tikvhandler.IngestParamMaxBatchSplitRanges)).Name("IngestMaxBatchSplitRanges")
+	router.Handle("/ingest/max-split-ranges-per-sec", tikvhandler.NewIngestConcurrencyHandler(
+		tikvHandlerTool, tikvhandler.IngestParamMaxSplitRangesPerSec)).Name("IngestMaxSplitRangesPerSec")
+	router.Handle("/ingest/max-ingest-inflight", tikvhandler.NewIngestConcurrencyHandler(
+		tikvHandlerTool, tikvhandler.IngestParamMaxInflight)).Name("IngestMaxInflight")
+	router.Handle("/ingest/max-ingest-per-sec", tikvhandler.NewIngestConcurrencyHandler(
+		tikvHandlerTool, tikvhandler.IngestParamMaxPerSecond)).Name("IngestMaxPerSec")
+
 	if s.cfg.Store == "tikv" {
 		// HTTP path for tikv.
 		router.Handle("/tables/{db}/{table}/regions", tikvhandler.NewTableHandler(tikvHandlerTool, tikvhandler.OpTableRegions))