metrics

package
v1.0.21 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 16, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

Documentation

Overview

Package metrics provides Prometheus metrics integration for mink.

This package enables observability through Prometheus metrics for event sourcing operations, including command execution, event store operations, and projections.

Basic usage: 

metrics := metrics.New()
// Register with Prometheus
prometheus.MustRegister(metrics.Collectors()...)

// Use with command bus
bus := mink.NewCommandBus()
bus.Use(metrics.CommandMiddleware())

// Use with event store
tracedStore := metrics.WrapEventStore(adapter)

The metrics collected include:

  • Command execution counts and durations
  • Event store operations (append, load, subscribe)
  • Projection processing metrics
  • Error counts by type

Index

Constants

View Source 
const (
	LabelCommandType    = "command_type"
	LabelAggregateType  = "aggregate_type"
	LabelEventType      = "event_type"
	LabelProjectionName = "projection_name"
	LabelOperation      = "operation"
	LabelStatus         = "status"
	LabelErrorType      = "error_type"
	LabelStreamID       = "stream_id"
	LabelService        = "service"
)

Default metric labels.

View Source 
const (
	StatusSuccess = "success"
	StatusError   = "error"
)

Status values.

View Source 
const (
	OperationAppend    = "append"
	OperationLoad      = "load"
	OperationSubscribe = "subscribe"
)

Operation values.

Variables

This section is empty.

Functions

This section is empty.

Types

type EventStoreMiddleware 

type EventStoreMiddleware struct {
	// contains filtered or unexported fields
}

EventStoreMiddleware wraps an EventStoreAdapter with metrics.

func (*EventStoreMiddleware) Append 

func (em *EventStoreMiddleware) Append(ctx context.Context, streamID string, events []adapters.EventRecord, expectedVersion int64) ([]adapters.StoredEvent, error)

Append stores events with metrics.

func (*EventStoreMiddleware) Close 

func (em *EventStoreMiddleware) Close() error

Close closes the adapter.

func (*EventStoreMiddleware) GetLastPosition 

func (em *EventStoreMiddleware) GetLastPosition(ctx context.Context) (uint64, error)

GetLastPosition returns the last global position with metrics.

func (*EventStoreMiddleware) GetStreamInfo 

func (em *EventStoreMiddleware) GetStreamInfo(ctx context.Context, streamID string) (*adapters.StreamInfo, error)

GetStreamInfo returns stream metadata with metrics.

func (*EventStoreMiddleware) Initialize 

func (em *EventStoreMiddleware) Initialize(ctx context.Context) error

Initialize initializes the adapter with metrics.

func (*EventStoreMiddleware) Load 

func (em *EventStoreMiddleware) Load(ctx context.Context, streamID string, fromVersion int64) ([]adapters.StoredEvent, error)

Load retrieves events with metrics.

func (*EventStoreMiddleware) LoadFromPosition 

func (em *EventStoreMiddleware) LoadFromPosition(ctx context.Context, fromPosition uint64, limit int) ([]adapters.StoredEvent, error)

LoadFromPosition loads events from a global position with metrics. Returns an error if the underlying adapter doesn't support subscriptions.

func (*EventStoreMiddleware) SubscribeAll 

func (em *EventStoreMiddleware) SubscribeAll(ctx context.Context, fromPosition uint64, opts ...adapters.SubscriptionOptions) (<-chan adapters.StoredEvent, error)

SubscribeAll subscribes to all events with metrics. Returns an error if the underlying adapter doesn't support subscriptions.

func (*EventStoreMiddleware) SubscribeCategory 

func (em *EventStoreMiddleware) SubscribeCategory(ctx context.Context, category string, fromPosition uint64, opts ...adapters.SubscriptionOptions) (<-chan adapters.StoredEvent, error)

SubscribeCategory subscribes to a category with metrics. Returns an error if the underlying adapter doesn't support subscriptions.

func (*EventStoreMiddleware) SubscribeStream 

func (em *EventStoreMiddleware) SubscribeStream(ctx context.Context, streamID string, fromVersion int64, opts ...adapters.SubscriptionOptions) (<-chan adapters.StoredEvent, error)

SubscribeStream subscribes to a stream with metrics. Returns an error if the underlying adapter doesn't support subscriptions.

func (*EventStoreMiddleware) SupportsSubscriptions 

func (em *EventStoreMiddleware) SupportsSubscriptions() bool

SupportsSubscriptions returns true if the underlying adapter supports subscriptions.

type Metrics 

type Metrics struct {
	// contains filtered or unexported fields
}

Metrics holds all Prometheus metrics for mink.

func New 

func New(opts ...MetricsOption) *Metrics

New creates a new Metrics instance with default settings.

func (*Metrics) Collectors 

func (m *Metrics) Collectors() []prometheus.Collector

Collectors returns all Prometheus collectors for registration.

func (*Metrics) CommandDuration 

func (m *Metrics) CommandDuration() *prometheus.HistogramVec

CommandDuration returns the command duration histogram.

func (*Metrics) CommandMiddleware 

func (m *Metrics) CommandMiddleware() mink.Middleware

CommandMiddleware returns middleware that records command metrics.

func (*Metrics) CommandsInFlight 

func (m *Metrics) CommandsInFlight() *prometheus.GaugeVec

CommandsInFlight returns the in-flight commands gauge.

func (*Metrics) CommandsTotal 

func (m *Metrics) CommandsTotal() *prometheus.CounterVec

CommandsTotal returns the commands counter.

func (*Metrics) ErrorsTotal 

func (m *Metrics) ErrorsTotal() *prometheus.CounterVec

ErrorsTotal returns the errors counter.

func (*Metrics) EventStoreOperationDuration 

func (m *Metrics) EventStoreOperationDuration() *prometheus.HistogramVec

EventStoreOperationDuration returns the event store duration histogram.

func (*Metrics) EventStoreOperationsTotal 

func (m *Metrics) EventStoreOperationsTotal() *prometheus.CounterVec

EventStoreOperationsTotal returns the event store operations counter.

func (*Metrics) EventsAppendedTotal 

func (m *Metrics) EventsAppendedTotal() *prometheus.CounterVec

EventsAppendedTotal returns the events appended counter.

func (*Metrics) EventsLoadedTotal 

func (m *Metrics) EventsLoadedTotal() *prometheus.CounterVec

EventsLoadedTotal returns the events loaded counter.

func (*Metrics) MustRegister 

func (m *Metrics) MustRegister()

MustRegister registers all collectors with the default registry. Panics if registration fails.

func (*Metrics) OutboxBatchDuration 

func (m *Metrics) OutboxBatchDuration() prometheus.Histogram

OutboxBatchDuration returns the outbox batch duration histogram.

func (*Metrics) OutboxDeadLetteredTotal 

func (m *Metrics) OutboxDeadLetteredTotal() prometheus.Counter

OutboxDeadLetteredTotal returns the outbox dead-lettered counter.

func (*Metrics) OutboxFailedTotal 

func (m *Metrics) OutboxFailedTotal() *prometheus.CounterVec

OutboxFailedTotal returns the outbox failed counter.

func (*Metrics) OutboxPendingMessages 

func (m *Metrics) OutboxPendingMessages() prometheus.Gauge

OutboxPendingMessages returns the outbox pending messages gauge.

func (*Metrics) OutboxProcessedTotal 

func (m *Metrics) OutboxProcessedTotal() *prometheus.CounterVec

OutboxProcessedTotal returns the outbox processed counter.

func (*Metrics) ProjectionCheckpoint 

func (m *Metrics) ProjectionCheckpoint() *prometheus.GaugeVec

ProjectionCheckpoint returns the projection checkpoint gauge.

func (*Metrics) ProjectionDuration 

func (m *Metrics) ProjectionDuration() *prometheus.HistogramVec

ProjectionDuration returns the projection duration histogram.

func (*Metrics) ProjectionLag 

func (m *Metrics) ProjectionLag() *prometheus.GaugeVec

ProjectionLag returns the projection lag gauge.

func (*Metrics) ProjectionsProcessedTotal 

func (m *Metrics) ProjectionsProcessedTotal() *prometheus.CounterVec

ProjectionsProcessedTotal returns the projections processed counter.

func (*Metrics) RecordBatchDuration 

func (m *Metrics) RecordBatchDuration(duration time.Duration)

RecordBatchDuration records the duration of an outbox batch processing cycle.

func (*Metrics) RecordError 

func (m *Metrics) RecordError(errorType string)

RecordError records a custom error.

func (*Metrics) RecordMessageDeadLettered 

func (m *Metrics) RecordMessageDeadLettered()

RecordMessageDeadLettered records an outbox message moved to dead letter.

func (*Metrics) RecordMessageFailed 

func (m *Metrics) RecordMessageFailed(destination string)

RecordMessageFailed records an outbox message delivery failure.

func (*Metrics) RecordMessageProcessed 

func (m *Metrics) RecordMessageProcessed(destination string, success bool)

RecordMessageProcessed records an outbox message processing result.

func (*Metrics) RecordPendingMessages 

func (m *Metrics) RecordPendingMessages(count int64)

RecordPendingMessages records the current count of pending outbox messages.

func (*Metrics) RecordProjectionCheckpoint 

func (m *Metrics) RecordProjectionCheckpoint(projectionName string, position uint64)

RecordProjectionCheckpoint records the checkpoint position for a projection.

func (*Metrics) RecordProjectionLag 

func (m *Metrics) RecordProjectionLag(projectionName string, lag int64)

RecordProjectionLag records the current lag for a projection.

func (*Metrics) Register 

func (m *Metrics) Register(registry prometheus.Registerer) error

Register registers all collectors with the given registry.

func (*Metrics) WrapEventStore 

func (m *Metrics) WrapEventStore(adapter adapters.EventStoreAdapter) *EventStoreMiddleware

WrapEventStore wraps an adapter with metrics collection.

func (*Metrics) WrapProjection 

func (m *Metrics) WrapProjection(projection mink.InlineProjection) *ProjectionMiddleware

WrapProjection wraps an inline projection with metrics collection.

type MetricsOption 

type MetricsOption func(*Metrics)

MetricsOption configures Metrics.

func WithMetricsServiceName 

func WithMetricsServiceName(name string) MetricsOption

WithMetricsServiceName sets the service name label.

func WithNamespace 

func WithNamespace(namespace string) MetricsOption

WithNamespace sets the Prometheus namespace.

func WithSubsystem 

func WithSubsystem(subsystem string) MetricsOption

WithSubsystem sets the Prometheus subsystem.

type ProjectionMiddleware 

type ProjectionMiddleware struct {
	// contains filtered or unexported fields
}

ProjectionMiddleware wraps an inline projection with metrics.

func (*ProjectionMiddleware) Apply 

func (pm *ProjectionMiddleware) Apply(ctx context.Context, event mink.StoredEvent) error

Apply applies an event with metrics.

func (*ProjectionMiddleware) HandledEvents 

func (pm *ProjectionMiddleware) HandledEvents() []string

HandledEvents returns the handled event types.

func (*ProjectionMiddleware) Name 

func (pm *ProjectionMiddleware) Name() string

Name returns the projection name.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.