mysql: move MySQL-only metrics into the MySQL sink package

downstreamadapter/sink/mysql/causality/conflict_detector.go

@@ -21,7 +21,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/errors"
-	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/ticdc/pkg/sink/mysql"
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/pingcap/ticdc/utils/chann"
 	"github.com/prometheus/client_golang/prometheus"
@@ -61,7 +61,7 @@ func New(
 		resolvedTxnCaches:            make([]txnCache, opt.Count),
 		slots:                        NewSlots(numSlots),
 		notifiedNodes:                chann.NewUnlimitedChannelDefault[func()](),
-		metricConflictDetectDuration: metrics.ConflictDetectDuration.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()),
+		metricConflictDetectDuration: mysql.ConflictDetectDuration.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()),
 
 		changefeedID: changefeedID,
 	}
@@ -75,7 +75,7 @@ func New(
 
 func (d *ConflictDetector) Run(ctx context.Context) error {
 	defer func() {
-		metrics.ConflictDetectDuration.DeleteLabelValues(d.changefeedID.Keyspace(), d.changefeedID.Name())
+		mysql.ConflictDetectDuration.DeleteLabelValues(d.changefeedID.Keyspace(), d.changefeedID.Name())
 		d.closeCache()
 	}()
 

downstreamadapter/sink/mysql/sink.go

@@ -227,18 +227,18 @@ func (s *Sink) runDMLWriter(ctx context.Context, idx int) error {
 	keyspace := s.changefeedID.Keyspace()
 	changefeed := s.changefeedID.Name()
 
-	workerBatchFlushDuration := metrics.WorkerBatchFlushDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-	workerFlushDuration := metrics.WorkerFlushDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-	workerTotalDuration := metrics.WorkerTotalDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-	workerHandledRows := metrics.WorkerHandledRows.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-	workerEventRowCount := metrics.WorkerEventRowCount.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerBatchFlushDuration := mysql.WorkerBatchFlushDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerFlushDuration := mysql.WorkerFlushDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerTotalDuration := mysql.WorkerTotalDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerHandledRows := mysql.WorkerHandledRows.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerEventRowCount := mysql.WorkerEventRowCount.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
 
 	defer func() {
-		metrics.WorkerFlushDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-		metrics.WorkerTotalDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-		metrics.WorkerHandledRows.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-		metrics.WorkerBatchFlushDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-		metrics.WorkerEventRowCount.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerFlushDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerTotalDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerHandledRows.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerBatchFlushDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerEventRowCount.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
 	}()
 
 	inputCh := s.conflictDetector.GetOutChByCacheID(idx)
@@ -457,6 +457,7 @@ func (s *Sink) Close() {
 		s.activeActiveSyncStatsCollector.Close()
 	}
 	s.statistics.Close()
+	mysql.DeleteDMLEventRowsAffectedMetrics(s.changefeedID)
 
 	metrics.ChangefeedDownstreamIsTiDBGauge.DeleteLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
 }

pkg/metrics/changefeed.go

@@ -26,30 +26,30 @@ var (
 			Subsystem: "maintainer",
 			Name:      "checkpoint_ts",
 			Help:      "checkpoint ts of maintainer",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	MaintainerCheckpointTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "maintainer",
 			Name:      "checkpoint_ts_lag",
 			Help:      "checkpoint ts lag of maintainer in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	MaintainerResolvedTsGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "maintainer",
 			Name:      "resolved_ts",
 			Help:      "resolved ts of maintainer",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	MaintainerResolvedTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "maintainer",
 			Name:      "resolved_ts_lag",
 			Help:      "resolved ts lag of maintainer in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	CoordinatorCounter = prometheus.NewCounter(
 		prometheus.CounterOpts{
@@ -65,15 +65,15 @@ var (
 			Subsystem: "changefeed",
 			Name:      "maintainer_counter",
 			Help:      "The counter of changefeed maintainer",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	ChangefeedStatusGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "owner",
 			Name:      "status",
 			Help:      "The status of changefeeds",
-		}, []string{getKeyspaceLabel(), "changefeed", "keyspace_id"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "keyspace_id"})
 
 	// ChangefeedErrorInfoGauge records the current warning or failed reason and its occurrence time
 	// for each changefeed.
@@ -83,7 +83,7 @@ var (
 			Subsystem: "owner",
 			Name:      "changefeed_error_info",
 			Help:      "The current warning or failed reason and occurrence time of changefeeds",
-		}, []string{getKeyspaceLabel(), "changefeed", "state", "error_time", "code", "message"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "state", "error_time", "code", "message"})
 
 	// ChangefeedOperationTimeGauge records a bounded set of recent user initiated
 	// changefeed operation timestamps for the Grafana investigation panel.
@@ -93,15 +93,15 @@ var (
 			Subsystem: "owner",
 			Name:      "changefeed_operation_time",
 			Help:      "Recent user initiated changefeed operation timestamps in Unix milliseconds",
-		}, []string{getKeyspaceLabel(), "changefeed", "operation", "result", "username", "details", "error", "event_id"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "operation", "result", "username", "details", "error", "event_id"})
 
 	ChangefeedCheckpointTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "owner",
 			Name:      "checkpoint_ts_lag",
 			Help:      "changefeed checkpoint ts lag in changefeeds in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed", "keyspace_id"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "keyspace_id"})
 
 	// it's a metrics used in a large number of tcms, we should always keep this metrics
 	ChangefeedCheckpointTsGauge = prometheus.NewGaugeVec(
@@ -110,7 +110,7 @@ var (
 			Subsystem: "owner",
 			Name:      "checkpoint_ts",
 			Help:      "checkpoint ts of changefeeds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// ChangefeedDownstreamInfoGauge is a metric with a constant '1' value,
 	// labeled by the downstream type of each changefeed.
@@ -123,7 +123,7 @@ var (
 			Subsystem: "owner",
 			Name:      "changefeed_downstream_info",
 			Help:      "Downstream type information of changefeeds exposed as labels.",
-		}, []string{getKeyspaceLabel(), "changefeed", "downstream_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "downstream_type"})
 
 	// ChangefeedDownstreamIsTiDBGauge indicates whether the downstream of a
 	// MySQL-compatible sink is confirmed to be TiDB (1 means yes).
@@ -138,7 +138,7 @@ var (
 			Subsystem: "sink",
 			Name:      "changefeed_downstream_is_tidb",
 			Help:      "Whether the downstream of a changefeed is confirmed to be TiDB (1 means yes).",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 )
 
 func DeleteChangefeedCheckpointMetrics(keyspace, changefeed string, keyspaceID uint32) {

pkg/metrics/ddl.go

@@ -27,7 +27,7 @@ var (
 			Name:      "handle_duration",
 			Help:      "Bucketed histogram of handling time (s) of a ddl.",
 			Buckets:   prometheus.ExponentialBuckets(0.01, 2, 18),
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// ExecDDLHistogram records the execution time of a DDL.
 	ExecDDLHistogram = prometheus.NewHistogramVec(
@@ -37,7 +37,7 @@ var (
 			Name:      "exec_duration",
 			Help:      "Bucketed histogram of processing time (s) of a ddl.",
 			Buckets:   prometheus.ExponentialBuckets(0.01, 2, 18),
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// ExecDDLRunningGauge records the count of running DDL.
 	ExecDDLRunningGauge = prometheus.NewGaugeVec(
@@ -46,7 +46,7 @@ var (
 			Subsystem: "ddl",
 			Name:      "exec_running",
 			Help:      "Total count of running ddl.",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// ExecDDLBlockingGauge records the count of blocking DDL.
 	ExecDDLBlockingGauge = prometheus.NewGaugeVec(
@@ -55,7 +55,7 @@ var (
 			Subsystem: "ddl",
 			Name:      "exec_blocking",
 			Help:      "Total count of blocking ddl.",
-		}, []string{getKeyspaceLabel(), "changefeed", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "mode"})
 
 	// ExecDDLCounter records the execution count of different DDL types
 	ExecDDLCounter = prometheus.NewCounterVec(
@@ -64,7 +64,7 @@ var (
 			Subsystem: "ddl",
 			Name:      "execution",
 			Help:      "Total execution count of different DDL types.",
-		}, []string{getKeyspaceLabel(), "changefeed", "ddl_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "ddl_type"})
 )
 
 func initDDLMetrics(registry *prometheus.Registry) {

pkg/metrics/dispatcher.go

@@ -22,23 +22,23 @@ var (
 			Subsystem: "dispatchermanagermanager",
 			Name:      "event_dispatcher_manager_count",
 			Help:      "The number of event dispatcher managers",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	TableTriggerEventDispatcherGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "dispatchermanager",
 			Name:      "table_trigger_dispatcher_count",
 			Help:      "The number of table dispatchers",
-		}, []string{getKeyspaceLabel(), "changefeed", "event_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "event_type"})
 
 	EventDispatcherGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "dispatchermanager",
 			Name:      "table_dispatcher_count",
 			Help:      "The number of table dispatchers",
-		}, []string{getKeyspaceLabel(), "changefeed", "event_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "event_type"})
 
 	CreateDispatcherDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -47,55 +47,55 @@ var (
 			Name:      "create_dispatcher_duration",
 			Help:      "Bucketed histogram of create dispatcher time (s) for table span.",
 			Buckets:   prometheus.ExponentialBuckets(0.000001, 2, 20), // 1us~524ms
-		}, []string{getKeyspaceLabel(), "changefeed", "event_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "event_type"})
 
 	DispatcherManagerResolvedTsGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "dispatchermanager",
 			Name:      "resolved_ts",
 			Help:      "Resolved ts of event dispatcher manager(changefeed)",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	DispatcherManagerResolvedTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "dispatchermanager",
 			Name:      "resolved_ts_lag",
 			Help:      "Resolved ts lag of event dispatcher manager(changefeed) in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	DispatcherManagerCheckpointTsGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "dispatchermanager",
 			Name:      "checkpoint_ts",
 			Help:      "Checkpoint ts of event dispatcher manager(changefeed)",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	DispatcherManagerCheckpointTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "dispatchermanager",
 			Name:      "checkpoint_ts_lag",
 			Help:      "Checkpoint ts lag of event dispatcher manager(changefeed) in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	DispatcherManagerBlockStatusesChanLenGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "dispatchermanager",
 			Name:      "block_statuses_chan_len",
 			Help:      "length of dispatcher manager block statuses channel",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	HandleDispatcherRequsetCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "ticdc",
 			Subsystem: "sink",
 			Name:      "handle_dispatcher_request",
 			Help:      "Total count of dispatcher request.",
-		}, []string{getKeyspaceLabel(), "changefeed", "type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 
 	DispatcherReceivedEventCount = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace: "ticdc",

pkg/metrics/dynamic_stream.go

@@ -21,7 +21,7 @@ var (
 			Namespace: "ticdc",
 			Subsystem: "dynamic_stream",
 			Name:      "memory_usage",
-		}, []string{"module", "type", getKeyspaceLabel(), "area"})
+		}, []string{"module", "type", GetKeyspaceLabel(), "area"})
 	DynamicStreamEventChanSize = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",

pkg/metrics/init.go

@@ -47,7 +47,8 @@ func InitMetrics(registry *prometheus.Registry) {
 	initDDLMetrics(registry)
 }
 
-func getKeyspaceLabel() string {
+// GetKeyspaceLabel returns the keyspace label name used by TiCDC metrics.
+func GetKeyspaceLabel() string {
 	if kerneltype.IsNextGen() {
 		return "keyspace_name"
 	}

pkg/metrics/log_coordinator.go

@@ -22,14 +22,14 @@ var (
 			Subsystem: "owner",
 			Name:      "resolved_ts",
 			Help:      "resolved ts of changefeeds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	ChangefeedResolvedTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "owner",
 			Name:      "resolved_ts_lag",
 			Help:      "resolved ts lag of changefeeds in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 )
 
 func initLogCoordinatorMetrics(registry *prometheus.Registry) {

pkg/metrics/maintainer.go

@@ -23,31 +23,31 @@ var (
 			Name:      "handle_event_duration",
 			Help:      "Bucketed histogram of maintainer handle event time (s).",
 			Buckets:   prometheus.ExponentialBuckets(0.01 /* 10 ms */, 2, 18),
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	MaintainerEventChLenGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "maintainer",
 			Name:      "event_ch_len",
 			Help:      "length of maintainer event channel",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	OperatorCount = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "maintainer",
 			Name:      "created_count",
 			Help:      "number of created operators",
-		}, []string{getKeyspaceLabel(), "changefeed", "type", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type", "mode"})
 
 	TotalOperatorCount = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "maintainer",
 			Name:      "total_operator_count",
 			Help:      "number of total operators",
-		}, []string{getKeyspaceLabel(), "changefeed", "type", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type", "mode"})
 
 	OperatorDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -56,7 +56,7 @@ var (
 			Name:      "finish_operators_duration_seconds",
 			Help:      "Bucketed histogram of processing time (s) of finished operator.",
 			Buckets:   []float64{0.5, 1, 2, 4, 8, 16, 20, 40, 60, 90, 120, 180, 240, 300, 480, 600, 720, 900, 1200, 1800, 3600},
-		}, []string{getKeyspaceLabel(), "changefeed", "type", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type", "mode"})
 )
 
 func initMaintainerMetrics(registry *prometheus.Registry) {