Skip to content

Commit edae046

Browse files
sbalabanov-zzsbalabanov
sbalabanov-zz
authored and
sbalabanov
committed
refactor: simplify operation metrics
1 parent 0b689f5 commit edae046

63 files changed

Lines changed: 322 additions & 515 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎platform/extension/counter/mysql/counter.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,8 @@ func NewCounter(db *sql.DB, scope tally.Scope) counter.Counter {
3737
// Next atomically increments the counter for the given domain and returns the new value.
3838
// Uses MySQL's LAST_INSERT_ID() to set the value atomically and read the incremented value.
3939
func (c *mysqlCounter) Next(ctx context.Context, domain string) (ret int64, retErr error) {
40-
op := metrics.Begin(c.scope, "next")
41-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
40+
op := metrics.Begin(c.scope, "next", metrics.StorageLatencyBuckets)
41+
defer func() { op.Complete(retErr) }()
4242
result, err := c.db.ExecContext(ctx,
4343
"INSERT INTO counter (domain, value) VALUES (?, LAST_INSERT_ID(1)) ON DUPLICATE KEY UPDATE value = LAST_INSERT_ID(value + 1)",
4444
domain,

‎platform/extension/messagequeue/mysql/delivery_state_store.go‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -49,11 +49,11 @@ func newDeliveryStateStore(db *sql.DB, logger *zap.SugaredLogger, scope tally.Sc
4949
// — only the lease holder calls MarkDelivered for a given partition, so no concurrent
5050
// mutation can occur between the two statements.
5151
func (s *sqldeliveryStateStore) MarkDelivered(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, visibilityTimeoutMs int64) (_ int, retErr error) {
52-
op := metrics.Begin(s.scope, "mark_delivered",
52+
op := metrics.Begin(s.scope, "mark_delivered", metrics.StorageLatencyBuckets,
5353
metrics.NewTag("topic", topic),
5454
metrics.NewTag("consumer_group", consumerGroup),
5555
metrics.NewTag("partition_key", partitionKey))
56-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
56+
defer func() { op.Complete(retErr) }()
5757

5858
now := time.Now().UnixMilli()
5959
invisibleUntil := now + visibilityTimeoutMs
@@ -90,11 +90,11 @@ func (s *sqldeliveryStateStore) MarkDelivered(ctx context.Context, consumerGroup
9090
// ExtendVisibility extends the visibility timeout for an in-flight message
9191
// without incrementing retry_count. Used by ExtendVisibilityTimeout.
9292
func (s *sqldeliveryStateStore) ExtendVisibility(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, visibilityTimeoutMs int64) (retErr error) {
93-
op := metrics.Begin(s.scope, "extend_visibility",
93+
op := metrics.Begin(s.scope, "extend_visibility", metrics.StorageLatencyBuckets,
9494
metrics.NewTag("topic", topic),
9595
metrics.NewTag("consumer_group", consumerGroup),
9696
metrics.NewTag("partition_key", partitionKey))
97-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
97+
defer func() { op.Complete(retErr) }()
9898

9999
now := time.Now().UnixMilli()
100100
invisibleUntil := now + visibilityTimeoutMs
@@ -124,11 +124,11 @@ func (s *sqldeliveryStateStore) ExtendVisibility(ctx context.Context, consumerGr
124124

125125
// MarkAcked sets acked = TRUE to indicate this group has processed the message.
126126
func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (retErr error) {
127-
op := metrics.Begin(s.scope, "mark_acked",
127+
op := metrics.Begin(s.scope, "mark_acked", metrics.StorageLatencyBuckets,
128128
metrics.NewTag("topic", topic),
129129
metrics.NewTag("consumer_group", consumerGroup),
130130
metrics.NewTag("partition_key", partitionKey))
131-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
131+
defer func() { op.Complete(retErr) }()
132132

133133
_, err := s.db.ExecContext(ctx, fmt.Sprintf(`
134134
INSERT INTO %s (consumer_group, topic, partition_key, message_offset, acked, invisible_until, retry_count)
@@ -147,11 +147,11 @@ func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, to
147147
// MarkNacked sets invisible_until = now + delay to schedule redelivery.
148148
// retry_count is NOT incremented here — it is incremented by MarkDelivered on redelivery.
149149
func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) (retErr error) {
150-
op := metrics.Begin(s.scope, "mark_nacked",
150+
op := metrics.Begin(s.scope, "mark_nacked", metrics.StorageLatencyBuckets,
151151
metrics.NewTag("topic", topic),
152152
metrics.NewTag("consumer_group", consumerGroup),
153153
metrics.NewTag("partition_key", partitionKey))
154-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
154+
defer func() { op.Complete(retErr) }()
155155

156156
now := time.Now().UnixMilli()
157157
invisibleUntil := now + delayMs
@@ -174,11 +174,11 @@ func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, t
174174
// GetDeliveryState returns the full delivery state for a message offset.
175175
// Returns (state, found, error). found=false means no row (never delivered).
176176
func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (_ DeliveryState, _ bool, retErr error) {
177-
op := metrics.Begin(s.scope, "get_delivery_state",
177+
op := metrics.Begin(s.scope, "get_delivery_state", metrics.StorageLatencyBuckets,
178178
metrics.NewTag("topic", topic),
179179
metrics.NewTag("consumer_group", consumerGroup),
180180
metrics.NewTag("partition_key", partitionKey))
181-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
181+
defer func() { op.Complete(retErr) }()
182182

183183
var state DeliveryState
184184
err := s.db.QueryRowContext(ctx, fmt.Sprintf(`
@@ -201,11 +201,11 @@ func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGr
201201
// offsets are the actual message offsets above the current watermark (from messageStore).
202202
// Returns the new watermark (highest contiguous acked offset from currentWatermark).
203203
func (s *sqldeliveryStateStore) AdvanceWatermark(ctx context.Context, consumerGroup, topic, partitionKey string, currentWatermark int64, offsets []int64) (_ int64, retErr error) {
204-
op := metrics.Begin(s.scope, "advance_watermark",
204+
op := metrics.Begin(s.scope, "advance_watermark", metrics.StorageLatencyBuckets,
205205
metrics.NewTag("topic", topic),
206206
metrics.NewTag("consumer_group", consumerGroup),
207207
metrics.NewTag("partition_key", partitionKey))
208-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
208+
defer func() { op.Complete(retErr) }()
209209

210210
if len(offsets) == 0 {
211211
return currentWatermark, nil

‎platform/extension/messagequeue/mysql/message_store.go‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -62,8 +62,8 @@ func (s *sqlmessageStore) Insert(ctx context.Context, topic string, messages []e
6262
// second Cancel RPC for the same request) without surfacing 1062 duplicate-key
6363
// errors.
6464
func (s *sqlmessageStore) InsertDelayed(ctx context.Context, topic string, messages []entityqueue.Message, visibleAfterMs int64) (retErr error) {
65-
op := metrics.Begin(s.scope, "insert", metrics.NewTag("topic", topic))
66-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
65+
op := metrics.Begin(s.scope, "insert", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
66+
defer func() { op.Complete(retErr) }()
6767

6868
if len(messages) == 0 {
6969
return nil
@@ -132,8 +132,8 @@ func (s *sqlmessageStore) InsertDelayed(ctx context.Context, topic string, messa
132132

133133
// Delete deletes a message by topic, partition key, and ID
134134
func (s *sqlmessageStore) Delete(ctx context.Context, topic string, partitionKey string, messageID string) (retErr error) {
135-
op := metrics.Begin(s.scope, "delete", metrics.NewTag("topic", topic))
136-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
135+
op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
136+
defer func() { op.Complete(retErr) }()
137137

138138
_, err := s.db.ExecContext(ctx, fmt.Sprintf(`
139139
DELETE FROM %s WHERE topic = ? AND partition_key = ? AND id = ?
@@ -151,8 +151,8 @@ func (s *sqlmessageStore) Delete(ctx context.Context, topic string, partitionKey
151151
// (published via InsertDelayed) that should not yet be surfaced to subscribers.
152152
// Messages are fetched from the immutable log; no per-message mutation occurs.
153153
func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, partitionKey string, currentOffset int64, nowMs int64, limit int) (_ []messageRow, retErr error) {
154-
op := metrics.Begin(s.scope, "fetch", metrics.NewTag("topic", topic))
155-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
154+
op := metrics.Begin(s.scope, "fetch", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
155+
defer func() { op.Complete(retErr) }()
156156

157157
rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
158158
SELECT offset, id, payload, metadata, partition_key, published_at, failed_at, failure_count, last_error, original_topic
@@ -227,8 +227,8 @@ func (s *sqlmessageStore) FetchByOffset(ctx context.Context, topic string, parti
227227
// The message is inserted back into queue_messages table with the DLQ topic (original + suffix)
228228
// This allows DLQ messages to be consumed using the normal subscriber
229229
func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partitionKey string, messageID string, failureCount int, lastError string, dlqTopicSuffix string) (retErr error) {
230-
op := metrics.Begin(s.scope, "move_to_dlq", metrics.NewTag("topic", topic))
231-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
230+
op := metrics.Begin(s.scope, "move_to_dlq", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
231+
defer func() { op.Complete(retErr) }()
232232

233233
// Construct DLQ topic name
234234
dlqTopic := topic + dlqTopicSuffix
@@ -300,8 +300,8 @@ func (s *sqlmessageStore) MoveToDLQ(ctx context.Context, topic string, partition
300300
// free of cross-table queries.
301301
// Returns the number of rows deleted.
302302
func (s *sqlmessageStore) GarbageCollect(ctx context.Context, topic string, partitionKey string, minAckedOffset int64) (_ int64, retErr error) {
303-
op := metrics.Begin(s.scope, "gc", metrics.NewTag("topic", topic))
304-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
303+
op := metrics.Begin(s.scope, "gc", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
304+
defer func() { op.Complete(retErr) }()
305305

306306
if minAckedOffset == 0 {
307307
return 0, nil
@@ -342,8 +342,8 @@ func (s *sqlmessageStore) GarbageCollect(ctx context.Context, topic string, part
342342

343343
// GetOffsetsAbove returns message offsets above afterOffset for a partition, ordered ascending.
344344
func (s *sqlmessageStore) GetOffsetsAbove(ctx context.Context, topic string, partitionKey string, afterOffset int64, limit int) (_ []int64, retErr error) {
345-
op := metrics.Begin(s.scope, "get_offsets_above", metrics.NewTag("topic", topic))
346-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
345+
op := metrics.Begin(s.scope, "get_offsets_above", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
346+
defer func() { op.Complete(retErr) }()
347347

348348
rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
349349
SELECT offset FROM %s

‎platform/extension/messagequeue/mysql/offset_store.go‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -40,11 +40,11 @@ func newOffsetStore(db *sql.DB, scope tally.Scope) offsetStore {
4040

4141
// Initialize creates an offset entry for a topic+partition if it doesn't exist
4242
func (s *sqloffsetStore) Initialize(ctx context.Context, topic string, partitionKey string, consumerGroup string) (retErr error) {
43-
op := metrics.Begin(s.scope, "initialize",
43+
op := metrics.Begin(s.scope, "initialize", metrics.StorageLatencyBuckets,
4444
metrics.NewTag("topic", topic),
4545
metrics.NewTag("partition_key", partitionKey),
4646
metrics.NewTag("consumer_group", consumerGroup))
47-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
47+
defer func() { op.Complete(retErr) }()
4848

4949
now := time.Now().UnixMilli()
5050

@@ -63,11 +63,11 @@ func (s *sqloffsetStore) Initialize(ctx context.Context, topic string, partition
6363

6464
// GetAckedOffset returns the current acked offset for a topic+partition
6565
func (s *sqloffsetStore) GetAckedOffset(ctx context.Context, topic string, partitionKey string, consumerGroup string) (_ int64, retErr error) {
66-
op := metrics.Begin(s.scope, "get_acked_offset",
66+
op := metrics.Begin(s.scope, "get_acked_offset", metrics.StorageLatencyBuckets,
6767
metrics.NewTag("topic", topic),
6868
metrics.NewTag("partition_key", partitionKey),
6969
metrics.NewTag("consumer_group", consumerGroup))
70-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
70+
defer func() { op.Complete(retErr) }()
7171

7272
var offset int64
7373
err := s.db.QueryRowContext(ctx, fmt.Sprintf(`
@@ -88,11 +88,11 @@ func (s *sqloffsetStore) GetAckedOffset(ctx context.Context, topic string, parti
8888

8989
// UpdateAckedOffset updates the offset_acked for a topic+partition (only if new offset is greater)
9090
func (s *sqloffsetStore) UpdateAckedOffset(ctx context.Context, topic string, partitionKey string, offset int64, consumerGroup string) (retErr error) {
91-
op := metrics.Begin(s.scope, "update_acked_offset",
91+
op := metrics.Begin(s.scope, "update_acked_offset", metrics.StorageLatencyBuckets,
9292
metrics.NewTag("topic", topic),
9393
metrics.NewTag("partition_key", partitionKey),
9494
metrics.NewTag("consumer_group", consumerGroup))
95-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
95+
defer func() { op.Complete(retErr) }()
9696

9797
now := time.Now().UnixMilli()
9898

@@ -112,10 +112,10 @@ func (s *sqloffsetStore) UpdateAckedOffset(ctx context.Context, topic string, pa
112112
// GetMinAckedOffset returns the minimum offset_acked across all consumer groups
113113
// for a topic+partition. Returns (0, false, nil) if no offset rows exist.
114114
func (s *sqloffsetStore) GetMinAckedOffset(ctx context.Context, topic string, partitionKey string) (_ int64, _ bool, retErr error) {
115-
op := metrics.Begin(s.scope, "get_min_acked_offset",
115+
op := metrics.Begin(s.scope, "get_min_acked_offset", metrics.StorageLatencyBuckets,
116116
metrics.NewTag("topic", topic),
117117
metrics.NewTag("partition_key", partitionKey))
118-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
118+
defer func() { op.Complete(retErr) }()
119119

120120
var minOffset int64
121121
err := s.db.QueryRowContext(ctx, fmt.Sprintf(`

‎platform/extension/messagequeue/mysql/partition_lease_store.go‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,8 @@ func newPartitionLeaseStore(db *sql.DB, logger *zap.SugaredLogger, scope tally.S
4444

4545
// TryAcquireLease attempts to acquire or renew a lease for a partition
4646
func (s *sqlpartitionLeaseStore) TryAcquireLease(ctx context.Context, topic string, partitionKey string, subscriberName string, consumerGroup string, leaseDurationMs int64) (_ bool, retErr error) {
47-
op := metrics.Begin(s.scope, "try_acquire_lease", metrics.NewTag("topic", topic))
48-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
47+
op := metrics.Begin(s.scope, "try_acquire_lease", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
48+
defer func() { op.Complete(retErr) }()
4949

5050
now := currentTimeMillis()
5151
staleThreshold := now - leaseDurationMs
@@ -93,8 +93,8 @@ func (s *sqlpartitionLeaseStore) TryAcquireLease(ctx context.Context, topic stri
9393

9494
// RenewLease renews the lease for a partition owned by this worker
9595
func (s *sqlpartitionLeaseStore) RenewLease(ctx context.Context, topic string, partitionKey string, subscriberName string, consumerGroup string, leaseDurationMs int64) (retErr error) {
96-
op := metrics.Begin(s.scope, "renew_lease", metrics.NewTag("topic", topic))
97-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
96+
op := metrics.Begin(s.scope, "renew_lease", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
97+
defer func() { op.Complete(retErr) }()
9898

9999
now := currentTimeMillis()
100100

@@ -127,8 +127,8 @@ func (s *sqlpartitionLeaseStore) RenewLease(ctx context.Context, topic string, p
127127

128128
// ReleaseLease releases the lease for a partition owned by this worker
129129
func (s *sqlpartitionLeaseStore) ReleaseLease(ctx context.Context, topic string, partitionKey string, subscriberName string, consumerGroup string) (retErr error) {
130-
op := metrics.Begin(s.scope, "release_lease", metrics.NewTag("topic", topic))
131-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
130+
op := metrics.Begin(s.scope, "release_lease", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
131+
defer func() { op.Complete(retErr) }()
132132

133133
result, err := s.db.ExecContext(ctx, fmt.Sprintf(`
134134
DELETE FROM %s
@@ -162,8 +162,8 @@ func (s *sqlpartitionLeaseStore) ReleaseLease(ctx context.Context, topic string,
162162

163163
// GetLeasedPartitions returns all partitions currently leased by this worker
164164
func (s *sqlpartitionLeaseStore) GetLeasedPartitions(ctx context.Context, topic string, subscriberName string, consumerGroup string) (_ []string, retErr error) {
165-
op := metrics.Begin(s.scope, "get_leased_partitions", metrics.NewTag("topic", topic))
166-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
165+
op := metrics.Begin(s.scope, "get_leased_partitions", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
166+
defer func() { op.Complete(retErr) }()
167167

168168
rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
169169
SELECT partition_key FROM %s
@@ -200,8 +200,8 @@ func (s *sqlpartitionLeaseStore) GetLeasedPartitions(ctx context.Context, topic
200200
// Returns the number of new leases acquired and the full list of discovered partitions.
201201
// maxPartitions limits how many total partitions this subscriber can own (0 = unlimited)
202202
func (s *sqlpartitionLeaseStore) DiscoverAndAcquirePartitions(ctx context.Context, topic string, subscriberName string, consumerGroup string, leaseDurationMs int64, maxPartitions int) (_ int, _ []string, retErr error) {
203-
op := metrics.Begin(s.scope, "discover_and_acquire", metrics.NewTag("topic", topic))
204-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
203+
op := metrics.Begin(s.scope, "discover_and_acquire", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
204+
defer func() { op.Complete(retErr) }()
205205

206206
// Query distinct partition_keys from messages table.
207207
// No LIMIT is applied because all partitions must be discoverable for fair

‎platform/extension/messagequeue/mysql/publisher.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,8 +46,8 @@ func NewPublisher(logger *zap.SugaredLogger, scope tally.Scope, messageStore mes
4646

4747
// Publish sends a message to the specified topic
4848
func (p *publisher) Publish(ctx context.Context, topic string, message entityqueue.Message) (retErr error) {
49-
op := metrics.Begin(p.scope, "publish", metrics.NewTag("topic", topic))
50-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
49+
op := metrics.Begin(p.scope, "publish", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
50+
defer func() { op.Complete(retErr) }()
5151

5252
// Check if closed (under lock)
5353
p.mu.RLock()
@@ -72,8 +72,8 @@ func (p *publisher) Publish(ctx context.Context, topic string, message entityque
7272
// now + delayMs; FetchByOffset skips it until that timestamp.
7373
// delayMs <= 0 is equivalent to Publish.
7474
func (p *publisher) PublishAfter(ctx context.Context, topic string, message entityqueue.Message, delayMs int64) (retErr error) {
75-
op := metrics.Begin(p.scope, "publish_after", metrics.NewTag("topic", topic))
76-
defer func() { op.Complete(retErr, metrics.StorageLatencyBuckets) }()
75+
op := metrics.Begin(p.scope, "publish_after", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
76+
defer func() { op.Complete(retErr) }()
7777

7878
p.mu.RLock()
7979
closed := p.closed

0 commit comments

Comments
 (0)