feat(metrics): record eventsourced pg outbox publishes and retries
otelsetup / test (push) Skipped
otelsetup / vulnerabilities (push) Skipped
pre-commit / pre-commit (push) Skipped
otelsetup / vulnerabilities (pull_request) Successful in 53s
otelsetup / test (pull_request) Successful in 1m9s
pre-commit / pre-commit (pull_request) Successful in 2m48s
otelsetup / test (push) Skipped
otelsetup / vulnerabilities (push) Skipped
pre-commit / pre-commit (push) Skipped
otelsetup / vulnerabilities (pull_request) Successful in 53s
otelsetup / test (pull_request) Successful in 1m9s
pre-commit / pre-commit (pull_request) Successful in 2m48s
NewEventsourcedMetrics now handles pg.OutboxPublish (eventsourced.outbox.publish.duration, by event.type and success) and pg.OutboxRetry (eventsourced.outbox.retries, by event.type and permanent), so a service using the pg/v2 outbox can alert on failing or abandoned publishes. OutboxBatch and OutboxCleanup stay ignored. otelsetup now imports codeberg.org/eventsourced/pg/v2 (v2.1.1). Every current consumer already depends on it; a release raises their minimum to v2.1.1. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013DJouG8ZZZxKtodF9kDvzj
This commit is contained in:
4 files changed
+79
-9
No files matched your search
+17
-2
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
|
||||
"codeberg.org/eventsourced/eventsourced"
|
||||
"codeberg.org/eventsourced/pg/v2"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/metric"
|
||||
@@ -40,6 +41,8 @@ type eventsourcedMetrics struct {
|
||||
snapshotStoreDur metric.Float64Histogram
|
||||
snapshotLoadDur metric.Float64Histogram
|
||||
idempotencyCheck metric.Int64Counter
|
||||
outboxPublishDur metric.Float64Histogram
|
||||
outboxRetries metric.Int64Counter
|
||||
}
|
||||
|
||||
// NewEventsourcedMetrics builds an eventsourced.MetricsRecorder that records to
|
||||
@@ -78,6 +81,8 @@ func NewEventsourcedMetrics() (eventsourced.MetricsRecorder, error) {
|
||||
snapshotStoreDur: hist("eventsourced.snapshot.store.duration", "Time taken to persist a snapshot."),
|
||||
snapshotLoadDur: hist("eventsourced.snapshot.load.duration", "Time taken to load a snapshot."),
|
||||
idempotencyCheck: counter("eventsourced.idempotency.checks", "Number of command idempotency lookups."),
|
||||
outboxPublishDur: hist("eventsourced.outbox.publish.duration", "Time taken to publish an event from the outbox."),
|
||||
outboxRetries: counter("eventsourced.outbox.retries", "Failed outbox publishes; permanent ones are never retried."),
|
||||
}
|
||||
if err := errors.Join(errs...); err != nil {
|
||||
return nil, err
|
||||
@@ -86,8 +91,8 @@ func NewEventsourcedMetrics() (eventsourced.MetricsRecorder, error) {
|
||||
}
|
||||
|
||||
// Record implements eventsourced.MetricsRecorder. Metric types the adapter does
|
||||
// not recognise (for example pg outbox metrics when the outbox is not enabled)
|
||||
// are ignored.
|
||||
// not recognise are ignored, and so are pg.OutboxBatch and pg.OutboxCleanup: a failed
|
||||
// publish already shows in the publish and retry instruments.
|
||||
func (e *eventsourcedMetrics) Record(ctx context.Context, raw eventsourced.Metric) {
|
||||
switch m := raw.(type) {
|
||||
case eventsourced.CommandDuration:
|
||||
@@ -119,5 +124,15 @@ func (e *eventsourcedMetrics) Record(ctx context.Context, raw eventsourced.Metri
|
||||
attribute.String("aggregate.type", m.AggregateType),
|
||||
attribute.Bool("hit", m.Hit),
|
||||
))
|
||||
case pg.OutboxPublish:
|
||||
e.outboxPublishDur.Record(ctx, m.Duration.Seconds(), metric.WithAttributes(
|
||||
attribute.String("event.type", m.EventType),
|
||||
attribute.Bool("success", m.Success),
|
||||
))
|
||||
case pg.OutboxRetry:
|
||||
e.outboxRetries.Add(ctx, 1, metric.WithAttributes(
|
||||
attribute.String("event.type", m.EventType),
|
||||
attribute.Bool("permanent", m.Permanent),
|
||||
))
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user