58 lines
1.4 KiB
Go
58 lines
1.4 KiB
Go
package metrics
|
|
|
|
import (
|
|
"github.com/ThreeDotsLabs/watermill/message"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
)
|
|
|
|
var (
|
|
subscriberLabelKeys = []string{
|
|
labelKeyHandlerName,
|
|
labelKeySubscriberName,
|
|
}
|
|
)
|
|
|
|
// SubscriberPrometheusMetricsDecorator decorates a subscriber to capture Prometheus metrics.
|
|
type SubscriberPrometheusMetricsDecorator struct {
|
|
message.Subscriber
|
|
subscriberName string
|
|
subscriberMessagesReceivedTotal *prometheus.CounterVec
|
|
closing chan struct{}
|
|
additionalLabels []MetricLabel
|
|
}
|
|
|
|
func (s SubscriberPrometheusMetricsDecorator) recordMetrics(msg *message.Message) {
|
|
if msg == nil {
|
|
return
|
|
}
|
|
|
|
ctx := msg.Context()
|
|
labels := labelsFromCtx(ctx, subscriberLabelKeys...)
|
|
if labels[labelKeySubscriberName] == "" {
|
|
labels[labelKeySubscriberName] = s.subscriberName
|
|
}
|
|
if labels[labelKeyHandlerName] == "" {
|
|
labels[labelKeyHandlerName] = labelValueNoHandler
|
|
}
|
|
for _, lb := range s.additionalLabels {
|
|
labels[lb.Label] = lb.ComputeValueFn(ctx)
|
|
}
|
|
|
|
go func() {
|
|
if subscribeAlreadyObserved(ctx) {
|
|
// decorator idempotency when applied decorator multiple times
|
|
return
|
|
}
|
|
|
|
select {
|
|
case <-msg.Acked():
|
|
labels[labelAcked] = "acked"
|
|
case <-msg.Nacked():
|
|
labels[labelAcked] = "nacked"
|
|
}
|
|
s.subscriberMessagesReceivedTotal.With(labels).Inc()
|
|
}()
|
|
|
|
msg.SetContext(setSubscribeObservedToCtx(msg.Context()))
|
|
}
|