add maintainer handle event time metrics and api error handler

Author: sdojjy

api/middleware/middleware.go

@@ -15,13 +15,15 @@ package middleware
 
 import (
 	"bufio"
-	"github.com/flowbehappy/tigate/pkg/node"
 	"net/http"
 	"strconv"
 	"time"
 
+	"github.com/flowbehappy/tigate/pkg/node"
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tiflow/cdc/api"
+	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/httputil"
 	"go.uber.org/zap"
@@ -39,6 +41,27 @@ const (
 // ClientVersionHeader is the header name of client version
 const ClientVersionHeader = "X-client-version"
 
+// ErrorHandleMiddleware puts the error into response
+func ErrorHandleMiddleware() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		c.Next()
+		// because we will return immediately after an error occurs in http_handler
+		// there wil be only one error in c.Errors
+		lastError := c.Errors.Last()
+		if lastError != nil {
+			err := lastError.Err
+			// put the error into response
+			if api.IsHTTPBadRequestError(err) {
+				c.IndentedJSON(http.StatusBadRequest, model.NewHTTPError(err))
+			} else {
+				c.IndentedJSON(http.StatusInternalServerError, model.NewHTTPError(err))
+			}
+			c.Abort()
+			return
+		}
+	}
+}
+
 // LogMiddleware logs the api requests
 func LogMiddleware() gin.HandlerFunc {
 	return func(c *gin.Context) {

api/v2/api.go

@@ -34,6 +34,7 @@ func RegisterOpenAPIV2Routes(router *gin.Engine, api OpenAPIV2) {
 	v2 := router.Group("/api/v2")
 
 	v2.Use(middleware.LogMiddleware())
+	v2.Use(middleware.ErrorHandleMiddleware())
 
 	v2.GET("status", api.serverStatus)
 	// For compatibility with the old API,

maintainer/barrier.go

@@ -138,7 +138,7 @@ func (b *Barrier) handleNoStateHeartbeat(dispatcherID common.DispatcherID, check
 		event.advancedDispatchers[dispatcherID] = true
 		// all dispatcher reported heartbeat, select one to write
 		if !event.selected && event.allDispatcherReported() {
-			distacherStatus := &heartbeatpb.DispatcherStatus{
+			dispatcherStatus := &heartbeatpb.DispatcherStatus{
 				InfluencedDispatchers: &heartbeatpb.InfluencedDispatchers{
 					InfluenceType: heartbeatpb.InfluenceType_Normal,
 					DispatcherIDs: []*heartbeatpb.DispatcherID{dispatcherID.ToPB()},
@@ -149,7 +149,7 @@ func (b *Barrier) handleNoStateHeartbeat(dispatcherID common.DispatcherID, check
 				}}
 			event.writerDispatcher = dispatcherID
 			event.selected = true
-			return nil, distacherStatus, nil
+			return nil, dispatcherStatus, nil
 		}
 	}
 

maintainer/maintainer.go

@@ -15,7 +15,6 @@ package maintainer
 
 import (
 	"encoding/json"
-	"fmt"
 	"math"
 	"sync"
 	"time"
@@ -100,6 +99,7 @@ type Maintainer struct {
 	scheduledTaskGauge             prometheus.Gauge
 	runningTaskGauge               prometheus.Gauge
 	tableCountGauge                prometheus.Gauge
+	handleEventDuration            prometheus.Observer
 }
 
 // NewMaintainer create the maintainer for the changefeed
@@ -143,6 +143,7 @@ func NewMaintainer(cfID model.ChangeFeedID,
 		scheduledTaskGauge:             metrics.ScheduleTaskGuage.WithLabelValues(cfID.Namespace, cfID.ID),
 		runningTaskGauge:               metrics.RunningScheduleTaskGauge.WithLabelValues(cfID.Namespace, cfID.ID),
 		tableCountGauge:                metrics.TableGauge.WithLabelValues(cfID.Namespace, cfID.ID),
+		handleEventDuration:            metrics.MaintainerHandleEventDuration.WithLabelValues(cfID.Namespace, cfID.ID),
 	}
 	m.bootstrapper = NewBootstrapper(m.id.ID, m.getNewBootstrapFn())
 	m.barrier = NewBarrier(m.scheduler)
@@ -159,12 +160,13 @@ func (m *Maintainer) HandleEvent(event *Event) bool {
 	start := time.Now()
 	defer func() {
 		duration := time.Since(start)
-		if duration > time.Millisecond*500 {
+		if duration > time.Second {
 			log.Info("maintainer is too slow",
 				zap.String("id", m.id.String()),
 				zap.Int("type", event.eventType),
 				zap.Duration("duration", duration))
 		}
+		m.handleEventDuration.Observe(duration.Seconds())
 	}()
 	if m.state == heartbeatpb.ComponentState_Stopped {
 		log.Warn("maintainer is not stopped, ignore",
@@ -274,6 +276,7 @@ func (m *Maintainer) cleanupMetrics() {
 	metrics.ScheduleTaskGuage.DeleteLabelValues(m.id.Namespace, m.id.ID)
 	metrics.RunningScheduleTaskGauge.DeleteLabelValues(m.id.Namespace, m.id.ID)
 	metrics.TableGauge.DeleteLabelValues(m.id.Namespace, m.id.ID)
+	metrics.MaintainerHandleEventDuration.DeleteLabelValues(m.id.Namespace, m.id.ID)
 }
 
 func (m *Maintainer) onInit() bool {
@@ -407,7 +410,7 @@ func (m *Maintainer) updateMetrics() {
 	m.changefeedStatusGauge.Set(float64(m.state))
 }
 
-// send message to remote, todo: use a io thread pool
+// send message to remote
 func (m *Maintainer) sendMessages(msgs []*messaging.TargetMessage) {
 	for _, msg := range msgs {
 		err := m.mc.SendCommand(msg)
@@ -628,15 +631,15 @@ func (m *Maintainer) onPeriodTask() {
 	}
 	// send scheduling messages
 	m.handleResendMessage()
-	m.printStatus()
+	m.collectMetrics()
 	m.calCheckpointTs()
 	SubmitScheduledEvent(m.taskScheduler, m.stream, &Event{
 		changefeedID: m.id.ID,
 		eventType:    EventPeriod,
 	}, time.Now().Add(time.Millisecond*500))
 }
 
-func (m *Maintainer) printStatus() {
+func (m *Maintainer) collectMetrics() {
 	if time.Since(m.lastPrintStatusTime) > time.Second*20 {
 		tableStates := make(map[scheduler.SchedulerStatus]int)
 		total := m.scheduler.TaskSize()
@@ -650,20 +653,6 @@ func (m *Maintainer) printStatus() {
 		for state, count := range tableStates {
 			metrics.TableStateGauge.WithLabelValues(m.id.Namespace, m.id.ID, state.String()).Set(float64(count))
 		}
-
-		var taskDistribution string
-		for nodeID, _ := range m.bootstrapper.GetAllNodes() {
-			taskDistribution = fmt.Sprintf("%s, %s=%d",
-				taskDistribution, nodeID, m.scheduler.GetTaskSizeByNodeID(nodeID))
-		}
-		log.Info("table span status",
-			zap.String("distribution", taskDistribution),
-			zap.String("changefeed", m.id.ID),
-			zap.Int("total", total),
-			zap.Int("absent", tableStates[scheduler.SchedulerStatusAbsent]),
-			zap.Int("commiting", tableStates[scheduler.SchedulerStatusCommiting]),
-			zap.Int("working", tableStates[scheduler.SchedulerStatusWorking]),
-			zap.Int("removing", tableStates[scheduler.SchedulerStatusRemoving]))
 		m.lastPrintStatusTime = time.Now()
 	}
 }

pkg/metrics/init.go

@@ -28,4 +28,5 @@ func InitMetrics(registry *prometheus.Registry) {
 	InitEventStoreMetrics(registry)
 	InitSchemaStoreMetrics(registry)
 	InitEventServiceMetrics(registry)
+	InitMaintainerMetrics(registry)
 }

pkg/metrics/maintainer.go

@@ -0,0 +1,31 @@
+// Copyright 2024 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 metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var (
+	MaintainerHandleEventDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "ticdc",
+			Subsystem: "maintainer",
+			Name:      "handle_event_duration",
+			Help:      "Bucketed histogram of maintainer handle event time (s).",
+			Buckets:   prometheus.ExponentialBuckets(0.01 /* 10 ms */, 2, 18),
+		}, []string{"namespace", "changefeed"})
+)
+
+func InitMaintainerMetrics(registry *prometheus.Registry) {
+	registry.MustRegister(MaintainerHandleEventDuration)
+}