logpuller: add http debug interface

api/v2/api.go

@@ -16,17 +16,19 @@ package v2
 import (
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/ticdc/api/middleware"
+	"github.com/pingcap/ticdc/logservice/logpuller"
 	"github.com/pingcap/ticdc/pkg/server"
 )
 
 // OpenAPIV2 provides CDC v2 APIs
 type OpenAPIV2 struct {
-	server server.Server
+	server              server.Server
+	pullerDebugProvider logpuller.DebugInfoProvider
 }
 
 // NewOpenAPIV2 creates a new OpenAPIV2.
 func NewOpenAPIV2(c server.Server) OpenAPIV2 {
-	return OpenAPIV2{c}
+	return OpenAPIV2{server: c}
 }
 
 // RegisterOpenAPIV2Routes registers routes for OpenAPI
@@ -51,6 +53,11 @@ func RegisterOpenAPIV2Routes(router *gin.Engine, api OpenAPIV2) {
 	debugGroup.POST("/failpoints", api.EnableFailpoint)
 	debugGroup.DELETE("/failpoints", api.DisableFailpoint)
 	debugGroup.GET("/failpoints", api.ListFailpoints)
+	debugGroup.GET("/puller", api.GetPullerDebugInfo)
+	debugGroup.GET(
+		"/puller/subscriptions/:subscription_id/regions/:region_id",
+		api.GetPullerDebugRegion,
+	)
 
 	coordinatorMiddleware := middleware.ForwardToCoordinatorMiddleware(api.server)
 	authenticateMiddleware := middleware.AuthenticateMiddleware(api.server)

api/v2/puller_debug.go

@@ -0,0 +1,115 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v2
+
+import (
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/pingcap/ticdc/logservice/logpuller"
+	apiutil "github.com/pingcap/ticdc/pkg/api"
+	appcontext "github.com/pingcap/ticdc/pkg/common/context"
+	cerror "github.com/pingcap/ticdc/pkg/errors"
+)
+
+const (
+	defaultPullerDebugLimit = 1
+	maxPullerDebugLimit     = 20
+)
+
+// GetPullerDebugInfo returns the slowest local subscriptions and Regions.
+func (h *OpenAPIV2) GetPullerDebugInfo(c *gin.Context) {
+	provider, ok := h.getPullerDebugProvider()
+	if !ok {
+		writePullerDebugError(c, http.StatusServiceUnavailable, "log puller is not initialized")
+		return
+	}
+	subscriptionLimit, ok := parsePullerDebugLimit(c, "subscription_limit")
+	if !ok {
+		return
+	}
+	regionLimit, ok := parsePullerDebugLimit(c, "region_limit")
+	if !ok {
+		return
+	}
+	c.IndentedJSON(http.StatusOK, provider.GetPullerDebugInfo(logpuller.PullerDebugOptions{
+		SubscriptionLimit: subscriptionLimit,
+		RegionLimit:       regionLimit,
+	}))
+}
+
+// GetPullerDebugRegion returns one Region owned by a local subscription.
+func (h *OpenAPIV2) GetPullerDebugRegion(c *gin.Context) {
+	provider, ok := h.getPullerDebugProvider()
+	if !ok {
+		writePullerDebugError(c, http.StatusServiceUnavailable, "log puller is not initialized")
+		return
+	}
+	subID, ok := parsePullerDebugUint(c, "subscription_id")
+	if !ok {
+		return
+	}
+	regionID, ok := parsePullerDebugUint(c, "region_id")
+	if !ok {
+		return
+	}
+	detail, found := provider.GetPullerDebugRegion(
+		logpuller.SubscriptionID(subID), regionID)
+	if !found {
+		writePullerDebugError(c, http.StatusNotFound, "puller region not found")
+		return
+	}
+	c.IndentedJSON(http.StatusOK, detail)
+}
+
+func (h *OpenAPIV2) getPullerDebugProvider() (logpuller.DebugInfoProvider, bool) {
+	if h.pullerDebugProvider != nil {
+		return h.pullerDebugProvider, true
+	}
+	return appcontext.TryGetService[logpuller.DebugInfoProvider](appcontext.SubscriptionClient)
+}
+
+func parsePullerDebugLimit(c *gin.Context, name string) (int, bool) {
+	raw := strings.TrimSpace(c.Query(name))
+	if raw == "" {
+		return defaultPullerDebugLimit, true
+	}
+	value, err := strconv.Atoi(raw)
+	if err != nil || value <= 0 || value > maxPullerDebugLimit {
+		writePullerDebugInvalidParam(c, name+" must be between 1 and 20")
+		return 0, false
+	}
+	return value, true
+}
+
+func parsePullerDebugUint(c *gin.Context, name string) (uint64, bool) {
+	raw := strings.TrimSpace(c.Param(name))
+	value, err := strconv.ParseUint(raw, 10, 64)
+	if err != nil || value == 0 {
+		writePullerDebugInvalidParam(c, name+" must be a positive integer")
+		return 0, false
+	}
+	return value, true
+}
+
+func writePullerDebugInvalidParam(c *gin.Context, message string) {
+	err := cerror.ErrAPIInvalidParam.GenWithStackByArgs(message)
+	c.IndentedJSON(http.StatusBadRequest, apiutil.NewHTTPError(err))
+}
+
+func writePullerDebugError(c *gin.Context, status int, message string) {
+	c.IndentedJSON(status, apiutil.HTTPError{Error: message})
+}

api/v2/puller_debug_test.go

@@ -0,0 +1,108 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// 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 v2
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/golang/mock/gomock"
+	"github.com/pingcap/ticdc/logservice/logpuller"
+	logpullermock "github.com/pingcap/ticdc/logservice/logpuller/mock"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPullerDebugAPI(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	ctrl := gomock.NewController(t)
+	provider := logpullermock.NewMockDebugInfoProvider(ctrl)
+	router := newPullerDebugTestRouter(OpenAPIV2{pullerDebugProvider: provider})
+
+	options := logpuller.PullerDebugOptions{SubscriptionLimit: 2, RegionLimit: 3}
+	provider.EXPECT().GetPullerDebugInfo(options).Return(logpuller.PullerDebugInfo{
+		SlowSubscriptions: []logpuller.PullerSubscriptionDebugInfo{{
+			SubscriptionID: 1,
+			SlowRegions: []logpuller.PullerRegionDebugInfo{{
+				RegionID: 11,
+			}},
+		}},
+	})
+	response := performPullerDebugRequest(
+		router, "/debug/puller?subscription_limit=2&region_limit=3")
+	require.Equal(t, http.StatusOK, response.Code)
+	var info logpuller.PullerDebugInfo
+	require.NoError(t, json.Unmarshal(response.Body.Bytes(), &info))
+	require.Equal(t, logpuller.SubscriptionID(1),
+		info.SlowSubscriptions[0].SubscriptionID)
+	require.Equal(t, uint64(11), info.SlowSubscriptions[0].SlowRegions[0].RegionID)
+
+	provider.EXPECT().GetPullerDebugRegion(
+		logpuller.SubscriptionID(1), uint64(11)).
+		Return(logpuller.PullerRegionDebugDetail{
+			SubscriptionID: 1,
+			Region:         logpuller.PullerRegionDebugInfo{RegionID: 11},
+		}, true)
+	response = performPullerDebugRequest(
+		router, "/debug/puller/subscriptions/1/regions/11")
+	require.Equal(t, http.StatusOK, response.Code)
+	var detail logpuller.PullerRegionDebugDetail
+	require.NoError(t, json.Unmarshal(response.Body.Bytes(), &detail))
+	require.Equal(t, uint64(11), detail.Region.RegionID)
+}
+
+func TestPullerDebugAPIValidationAndNotFound(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	ctrl := gomock.NewController(t)
+	provider := logpullermock.NewMockDebugInfoProvider(ctrl)
+	router := newPullerDebugTestRouter(OpenAPIV2{pullerDebugProvider: provider})
+
+	response := performPullerDebugRequest(
+		router, "/debug/puller?subscription_limit=21")
+	require.Equal(t, http.StatusBadRequest, response.Code)
+
+	response = performPullerDebugRequest(
+		router, "/debug/puller/subscriptions/not-a-number/regions/11")
+	require.Equal(t, http.StatusBadRequest, response.Code)
+
+	provider.EXPECT().GetPullerDebugRegion(
+		logpuller.SubscriptionID(999), uint64(11)).
+		Return(logpuller.PullerRegionDebugDetail{}, false)
+	response = performPullerDebugRequest(
+		router, "/debug/puller/subscriptions/999/regions/11")
+	require.Equal(t, http.StatusNotFound, response.Code)
+}
+
+func newPullerDebugTestRouter(api OpenAPIV2) *gin.Engine {
+	router := gin.New()
+	router.GET("/debug/puller", api.GetPullerDebugInfo)
+	router.GET(
+		"/debug/puller/subscriptions/:subscription_id/regions/:region_id",
+		api.GetPullerDebugRegion,
+	)
+	return router
+}
+
+func performPullerDebugRequest(
+	router *gin.Engine,
+	path string,
+) *httptest.ResponseRecorder {
+	recorder := httptest.NewRecorder()
+	request := httptest.NewRequest(http.MethodGet, path, nil)
+	router.ServeHTTP(recorder, request)
+	return recorder
+}

logservice/eventstore/pebble.go

@@ -42,7 +42,7 @@ func newPebbleOptions(dbNum int) *pebble.Options {
 
 		MaxOpenFiles: maxOpenFilesPerDB,
 
-		MaxConcurrentCompactions: func() int { return 6 },
+		MaxConcurrentCompactions: func() int { return 3 },
 
 		// Decrease compaction frequency
 		L0CompactionThreshold:     20,