113 lines
3.1 KiB
Go
113 lines
3.1 KiB
Go
package queue
|
|
|
|
import (
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
type Metrics struct {
|
|
received atomic.Int64
|
|
processed atomic.Int64
|
|
retried atomic.Int64
|
|
failed atomic.Int64
|
|
deadLettered atomic.Int64
|
|
duplicates atomic.Int64
|
|
malformed atomic.Int64
|
|
ackFailures atomic.Int64
|
|
retryFailures atomic.Int64
|
|
deadLetterFailures atomic.Int64
|
|
lastQueueDepth atomic.Int64
|
|
totalProcessingNanos atomic.Int64
|
|
lastProcessingNanos atomic.Int64
|
|
}
|
|
|
|
type MetricsSnapshot struct {
|
|
Received int64 `json:"received"`
|
|
Processed int64 `json:"processed"`
|
|
Retried int64 `json:"retried"`
|
|
Failed int64 `json:"failed"`
|
|
DeadLettered int64 `json:"dead_lettered"`
|
|
Duplicates int64 `json:"duplicates"`
|
|
Malformed int64 `json:"malformed"`
|
|
AckFailures int64 `json:"ack_failures"`
|
|
RetryFailures int64 `json:"retry_failures"`
|
|
DeadLetterFailures int64 `json:"dead_letter_failures"`
|
|
ApproximateQueueDepth int64 `json:"approximate_queue_depth"`
|
|
LastProcessingLatency time.Duration `json:"last_processing_latency"`
|
|
AverageProcessingLatency time.Duration `json:"average_processing_latency"`
|
|
}
|
|
|
|
func NewMetrics() *Metrics {
|
|
return &Metrics{}
|
|
}
|
|
|
|
func (m *Metrics) RecordReceive(n int) {
|
|
m.received.Add(int64(n))
|
|
}
|
|
|
|
func (m *Metrics) RecordProcessed(duration time.Duration) {
|
|
m.processed.Add(1)
|
|
m.lastProcessingNanos.Store(duration.Nanoseconds())
|
|
m.totalProcessingNanos.Add(duration.Nanoseconds())
|
|
}
|
|
|
|
func (m *Metrics) RecordRetry() {
|
|
m.retried.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) RecordFailure() {
|
|
m.failed.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) RecordDeadLetter() {
|
|
m.deadLettered.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) RecordDuplicate() {
|
|
m.duplicates.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) RecordMalformed() {
|
|
m.malformed.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) RecordAckFailure() {
|
|
m.ackFailures.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) RecordRetryFailure() {
|
|
m.retryFailures.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) RecordDeadLetterFailure() {
|
|
m.deadLetterFailures.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) SetQueueDepth(depth int64) {
|
|
m.lastQueueDepth.Store(depth)
|
|
}
|
|
|
|
func (m *Metrics) Snapshot() MetricsSnapshot {
|
|
processed := m.processed.Load()
|
|
total := m.totalProcessingNanos.Load()
|
|
avg := time.Duration(0)
|
|
if processed > 0 {
|
|
avg = time.Duration(total / processed)
|
|
}
|
|
return MetricsSnapshot{
|
|
Received: m.received.Load(),
|
|
Processed: processed,
|
|
Retried: m.retried.Load(),
|
|
Failed: m.failed.Load(),
|
|
DeadLettered: m.deadLettered.Load(),
|
|
Duplicates: m.duplicates.Load(),
|
|
Malformed: m.malformed.Load(),
|
|
AckFailures: m.ackFailures.Load(),
|
|
RetryFailures: m.retryFailures.Load(),
|
|
DeadLetterFailures: m.deadLetterFailures.Load(),
|
|
ApproximateQueueDepth: m.lastQueueDepth.Load(),
|
|
LastProcessingLatency: time.Duration(m.lastProcessingNanos.Load()),
|
|
AverageProcessingLatency: avg,
|
|
}
|
|
}
|