From 918306c727097d201b3217419f9281b4b184cd6e Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 4 Aug 2026 17:59:17 +0800 Subject: [PATCH 1/2] remove checkpoint ts and count metrics --- downstreamadapter/sink/cloudstorage/sink.go | 15 +----- downstreamadapter/sink/kafka/sink.go | 10 ---- downstreamadapter/sink/metrics/mq.go | 52 ++------------------- downstreamadapter/sink/pulsar/sink.go | 11 ----- pkg/metrics/sink.go | 19 -------- 5 files changed, 6 insertions(+), 101 deletions(-) diff --git a/downstreamadapter/sink/cloudstorage/sink.go b/downstreamadapter/sink/cloudstorage/sink.go index 297c5f1e06..2be2380780 100644 --- a/downstreamadapter/sink/cloudstorage/sink.go +++ b/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()) } } diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 2133e1c6ba..d60b843b24 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/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()) } } } diff --git a/downstreamadapter/sink/metrics/mq.go b/downstreamadapter/sink/metrics/mq.go index 3bdca8f5bf..ce718f244e 100644 --- a/downstreamadapter/sink/metrics/mq.go +++ b/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) } diff --git a/downstreamadapter/sink/pulsar/sink.go b/downstreamadapter/sink/pulsar/sink.go index 9895541233..7db478ec1a 100644 --- a/downstreamadapter/sink/pulsar/sink.go +++ b/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()) } } } diff --git a/pkg/metrics/sink.go b/pkg/metrics/sink.go index 925cb718ad..73ef980542 100644 --- a/pkg/metrics/sink.go +++ b/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) From 617db028a180ffea10fcf92565188b3cbb228930 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Wed, 5 Aug 2026 11:37:34 +0800 Subject: [PATCH 2/2] fix metrics name export --- pkg/sink/kafka/metrics.go | 18 +++++++++--------- pkg/sink/kafka/metrics_collector.go | 16 ++++++++-------- 2 files changed, 17 insertions(+), 17 deletions(-) diff --git a/pkg/sink/kafka/metrics.go b/pkg/sink/kafka/metrics.go index d3f88055e2..1bd9c6a60b 100644 --- a/pkg/sink/kafka/metrics.go +++ b/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) diff --git a/pkg/sink/kafka/metrics_collector.go b/pkg/sink/kafka/metrics_collector.go index 96f969dd39..99e7e9ba63 100644 --- a/pkg/sink/kafka/metrics_collector.go +++ b/pkg/sink/kafka/metrics_collector.go @@ -129,7 +129,7 @@ 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()) } @@ -137,7 +137,7 @@ func (m *saramaMetricsCollector) collectBrokerMetrics() { requestRateMetric := m.registry.Get( getBrokerMetricName(requestRateMetricNamePrefix, brokerID)) if meter, ok := requestRateMetric.(metrics.Meter); ok { - RequestRateGauge. + requestRateGauge. WithLabelValues(keyspace, changefeedID, brokerID). Set(meter.Snapshot().Rate1()) } @@ -145,10 +145,10 @@ func (m *saramaMetricsCollector) collectBrokerMetrics() { 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)