kafka: fix metrics

downstreamadapter/sink/cloudstorage/sink.go

@@ -328,17 +328,7 @@ func (s *sink) AddCheckpointTs(ts uint64) {
 }
 
 func (s *sink) sendCheckpointTs(ctx context.Context) error {
-	var (
-		keyspace   = s.changefeedID.Keyspace()
-		changefeed = s.changefeedID.Name()
-	)
-	checkpointTsMessageDuration := metrics.CheckpointTsMessageDuration.WithLabelValues(keyspace, changefeed)
-	checkpointTsMessageCount := metrics.CheckpointTsMessageCount.WithLabelValues(keyspace, changefeed)
-	defer func() {
-		metrics.CheckpointTsMessageDuration.DeleteLabelValues(keyspace, changefeed)
-		metrics.CheckpointTsMessageCount.DeleteLabelValues(keyspace, changefeed)
-	}()
-
+	keyspace, changefeed := s.changefeedID.Keyspace(), s.changefeedID.Name()
 	var checkpoint uint64
 	for {
 		select {
@@ -376,9 +366,6 @@ func (s *sink) sendCheckpointTs(ctx context.Context) error {
 		}
 		s.lastSendCheckpointTsTime = time.Now()
 		s.lastCheckpointTs.Store(checkpoint)
-
-		checkpointTsMessageCount.Inc()
-		checkpointTsMessageDuration.Observe(time.Since(start).Seconds())
 	}
 }
 

downstreamadapter/sink/kafka/sink.go

@@ -503,13 +503,6 @@ func (s *sink) AddCheckpointTs(ts uint64) {
 }
 
 func (s *sink) sendCheckpoint(ctx context.Context) error {
-	checkpointTsMessageDuration := metrics.CheckpointTsMessageDuration.WithLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
-	checkpointTsMessageCount := metrics.CheckpointTsMessageCount.WithLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
-	defer func() {
-		metrics.CheckpointTsMessageDuration.DeleteLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
-		metrics.CheckpointTsMessageCount.DeleteLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
-	}()
-
 	var (
 		msg          *codecCommon.Message
 		partitionNum int32
@@ -524,7 +517,6 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 				return nil
 			}
 
-			start := time.Now()
 			msg, err = s.comp.encoder.EncodeCheckpointEvent(ts)
 			if err != nil {
 				return err
@@ -561,8 +553,6 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 					}
 				}
 			}
-			checkpointTsMessageCount.Inc()
-			checkpointTsMessageDuration.Observe(time.Since(start).Seconds())
 		}
 	}
 }

downstreamadapter/sink/metrics/mq.go

@@ -16,65 +16,25 @@ package metrics
 import (
 	"sync"
 
-	"github.com/pingcap/ticdc/pkg/sink/codec"
-	"github.com/pingcap/ticdc/pkg/sink/kafka"
-	"github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-var (
-	// WorkerSendMessageDuration records the duration of flushing a group messages.
-	WorkerSendMessageDuration = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "mq_worker_send_message_duration",
-			Help:      "Send Message duration(s) for MQ worker.",
-			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{"namespace", "changefeed"})
-	// WorkerBatchSize record the size of each batched messages.
-	WorkerBatchSize = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "mq_worker_batch_size",
-			Help:      "Batch size for MQ worker.",
-			Buckets:   prometheus.ExponentialBuckets(4, 2, 10), // 4 ~ 2048
-		}, []string{"namespace", "changefeed"})
-	// WorkerBatchDuration record the time duration cost on batch messages.
-	WorkerBatchDuration = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "mq_worker_batch_duration",
-			Help:      "Batch duration for MQ worker.",
-			Buckets:   prometheus.ExponentialBuckets(0.004, 2, 10), // 4ms ~ 2s
-		}, []string{"namespace", "changefeed"})
-)
-
 var (
 	mqServerRegistryMu sync.RWMutex
-	// mqServerRegistry is shared by all MQ sinks on the node. Bootstrap can now
-	// create multiple changefeeds concurrently, so both reads and the fallback
-	// initialization must be synchronized to avoid racing on the global pointer.
+	// mqServerRegistry is shared by all MQ sinks on the node. Bootstrap can
+	// create multiple changefeeds concurrently, so reads and initialization
+	// must be synchronized.
 	mqServerRegistry *prometheus.Registry
 )
 
-// InitMQMetrics registers all metrics in this file.
+// InitMQMetrics configures the registry used by MQ client metrics.
 func InitMQMetrics(registry *prometheus.Registry) {
 	mqServerRegistryMu.Lock()
 	mqServerRegistry = registry
 	mqServerRegistryMu.Unlock()
-
-	registry.MustRegister(WorkerSendMessageDuration)
-	registry.MustRegister(WorkerBatchSize)
-	registry.MustRegister(WorkerBatchDuration)
-	claimcheck.InitMetrics(registry)
-	codec.InitMetrics(registry)
-	kafka.InitMetrics(registry)
 }
 
-// GetMQMetricRegistry for add pulsar default metrics
+// GetMQMetricRegistry returns the registry used by MQ client metrics.
 func GetMQMetricRegistry() *prometheus.Registry {
 	mqServerRegistryMu.RLock()
 	registry := mqServerRegistry
@@ -85,8 +45,6 @@ func GetMQMetricRegistry() *prometheus.Registry {
 
 	mqServerRegistryMu.Lock()
 	defer mqServerRegistryMu.Unlock()
-	// Make sure registry is not nil when MQ sink metrics are first requested
-	// before the server metrics bootstrap wires in the shared registry.
 	if mqServerRegistry == nil {
 		mqServerRegistry = prometheus.DefaultRegisterer.(*prometheus.Registry)
 	}

downstreamadapter/sink/pulsar/sink.go

@@ -283,13 +283,6 @@ func (s *sink) SetTableSchemaStore(tableSchemaStore *commonEvent.TableSchemaStor
 }
 
 func (s *sink) sendCheckpoint(ctx context.Context) error {
-	checkpointTsMessageDuration := metrics.CheckpointTsMessageDuration.WithLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
-	checkpointTsMessageCount := metrics.CheckpointTsMessageCount.WithLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
-
-	defer func() {
-		metrics.CheckpointTsMessageDuration.DeleteLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
-		metrics.CheckpointTsMessageCount.DeleteLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
-	}()
 	var (
 		msg *common.Message
 		err error
@@ -306,7 +299,6 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 				return nil
 			}
 
-			start := time.Now()
 			msg, err = s.comp.encoder.EncodeCheckpointEvent(ts)
 			if err != nil {
 				return errors.Trace(err)
@@ -344,9 +336,6 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 					}
 				}
 			}
-
-			checkpointTsMessageCount.Inc()
-			checkpointTsMessageDuration.Observe(time.Since(start).Seconds())
 		}
 	}
 }

pkg/metrics/sink.go

@@ -144,23 +144,6 @@ var (
 			Help:      "Batch duration for MQ worker.",
 			Buckets:   prometheus.ExponentialBuckets(0.004, 2, 10), // 4ms ~ 2s
 		}, []string{GetKeyspaceLabel(), "changefeed"})
-
-	CheckpointTsMessageDuration = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "mq_checkpoint_ts_message_duration",
-			Help:      "Duration of sending checkpoint ts message.",
-			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{GetKeyspaceLabel(), "changefeed"})
-
-	CheckpointTsMessageCount = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "mq_checkpoint_ts_message_count",
-			Help:      "Number of checkpoint ts messages sent.",
-		}, []string{GetKeyspaceLabel(), "changefeed"})
 )
 
 // InitMetrics registers all metrics in this file.
@@ -183,8 +166,6 @@ func initSinkMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(WorkerSendMessageDuration)
 	registry.MustRegister(WorkerBatchSize)
 	registry.MustRegister(WorkerBatchDuration)
-	registry.MustRegister(CheckpointTsMessageDuration)
-	registry.MustRegister(CheckpointTsMessageCount)
 
 	// pulsar sink metrics
 	initPulsarMetrics(registry)

pkg/sink/kafka/metrics.go

@@ -28,25 +28,25 @@ var (
 			Help: "The current number of in-flight requests" +
 				" awaiting a response for all brokers.",
 		}, []string{"namespace", "changefeed", "broker"})
-	// OutgoingByteRateGauge for outgoing events.
+	// outgoingByteRateGauge is for outgoing events.
 	// Meter mark for each request's size in bytes.
-	OutgoingByteRateGauge = prometheus.NewGaugeVec(
+	outgoingByteRateGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "sink",
 			Name:      "kafka_producer_outgoing_byte_rate",
 			Help:      "Bytes/second written off all brokers.",
 		}, []string{"namespace", "changefeed", "broker"})
-	// RequestRateGauge Meter mark by 1 for each request.
-	RequestRateGauge = prometheus.NewGaugeVec(
+	// requestRateGauge is marked by 1 for each request.
+	requestRateGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "sink",
 			Name:      "kafka_producer_request_rate",
 			Help:      "Requests/second sent to all brokers.",
 		}, []string{"namespace", "changefeed", "broker"})
-	// RequestLatencyGauge Histogram update by `requestLatency`.
-	RequestLatencyGauge = prometheus.NewGaugeVec(
+	// requestLatencyGauge is updated by the request latency histogram.
+	requestLatencyGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "sink",
@@ -84,9 +84,9 @@ var (
 func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(compressionRatioGauge)
 	registry.MustRegister(recordsPerRequestGauge)
-	registry.MustRegister(OutgoingByteRateGauge)
-	registry.MustRegister(RequestRateGauge)
-	registry.MustRegister(RequestLatencyGauge)
+	registry.MustRegister(outgoingByteRateGauge)
+	registry.MustRegister(requestRateGauge)
+	registry.MustRegister(requestLatencyGauge)
 	registry.MustRegister(requestsInFlightGauge)
 	registry.MustRegister(responseRateGauge)
 

pkg/sink/kafka/metrics_collector.go

@@ -129,26 +129,26 @@ func (m *saramaMetricsCollector) collectBrokerMetrics() {
 		outgoingByteRateMetric := m.registry.Get(
 			getBrokerMetricName(outgoingByteRateMetricNamePrefix, brokerID))
 		if meter, ok := outgoingByteRateMetric.(metrics.Meter); ok {
-			OutgoingByteRateGauge.
+			outgoingByteRateGauge.
 				WithLabelValues(keyspace, changefeedID, brokerID).
 				Set(meter.Snapshot().Rate1())
 		}
 
 		requestRateMetric := m.registry.Get(
 			getBrokerMetricName(requestRateMetricNamePrefix, brokerID))
 		if meter, ok := requestRateMetric.(metrics.Meter); ok {
-			RequestRateGauge.
+			requestRateGauge.
 				WithLabelValues(keyspace, changefeedID, brokerID).
 				Set(meter.Snapshot().Rate1())
 		}
 
 		requestLatencyMetric := m.registry.Get(
 			getBrokerMetricName(requestLatencyInMsMetricNamePrefix, brokerID))
 		if histogram, ok := requestLatencyMetric.(metrics.Histogram); ok {
-			RequestLatencyGauge.
+			requestLatencyGauge.
 				WithLabelValues(keyspace, changefeedID, brokerID, avg).
 				Set(histogram.Snapshot().Mean() / 1000)
-			RequestLatencyGauge.
+			requestLatencyGauge.
 				WithLabelValues(keyspace, changefeedID, brokerID, p99).
 				Set(histogram.Snapshot().Percentile(0.99) / 1000)
 		}
@@ -192,13 +192,13 @@ func (m *saramaMetricsCollector) cleanupBrokerMetrics() {
 	changefeedID := m.changefeedID.Name()
 	for id := range m.brokers {
 		brokerID := strconv.Itoa(int(id))
-		OutgoingByteRateGauge.
+		outgoingByteRateGauge.
 			DeleteLabelValues(keyspace, changefeedID, brokerID)
-		RequestRateGauge.
+		requestRateGauge.
 			DeleteLabelValues(keyspace, changefeedID, brokerID)
-		RequestLatencyGauge.
+		requestLatencyGauge.
 			DeleteLabelValues(keyspace, changefeedID, brokerID, avg)
-		RequestLatencyGauge.
+		requestLatencyGauge.
 			DeleteLabelValues(keyspace, changefeedID, brokerID, p99)
 		requestsInFlightGauge.
 			DeleteLabelValues(keyspace, changefeedID, brokerID)