metrics

package

v0.22.0 Latest Latest Go to latest Published: Sep 25, 2026 License: MIT Imports: 13 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/batchstream/sink

Links

Open Source Insights

Documentation ¶

Overview ¶

Package metrics exposes bounded-cardinality Prometheus metrics for Sink.

Index ¶

type BatchObservation
type Metrics
- func New(version string, storeNames ...string) (*Metrics, error)
type RequestQueueObservation
type WritePhaseObservation

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type BatchObservation ¶

type BatchObservation struct {
	Store             string
	Method            string
	Reason            string
	Operations        int
	Bytes             int
	QueueDuration     time.Duration
	ExecutionDuration time.Duration
}

type Metrics ¶

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

func New ¶

func New(version string, storeNames ...string) (*Metrics, error)

func (*Metrics) AdjustBatchQueue ¶

func (m *Metrics) AdjustBatchQueue(store string, method string, operations int, bytes int)

func (*Metrics) AdjustInFlight ¶

func (m *Metrics) AdjustInFlight(requests int)

func (*Metrics) Handler ¶

func (m *Metrics) Handler() http.Handler

func (*Metrics) ObserveBatch ¶

func (m *Metrics) ObserveBatch(observation BatchObservation)

func (*Metrics) ObserveBatchRejected ¶

func (m *Metrics) ObserveBatchRejected(store string, method string, reason string)

func (*Metrics) ObserveForward ¶

func (m *Metrics) ObserveForward(req *forward.ForwardRequest, code codes.Code, elapsed time.Duration)

ObserveForward retains the public method labels when the wire RPC is Forward.

func (*Metrics) ObserveForwardResult ¶

func (m *Metrics) ObserveForwardResult(req *forward.ForwardRequest, resp *forward.ForwardResponse)

ObserveForwardResult records delivered outcomes without retaining documents.

func (*Metrics) ObserveKafkaDeadLetter ¶

func (m *Metrics) ObserveKafkaDeadLetter(store string, count int)

func (*Metrics) ObserveKafkaPublish ¶

func (m *Metrics) ObserveKafkaPublish(store string, duration time.Duration, accepted int, failed int)

func (*Metrics) ObserveKafkaRetry ¶

func (m *Metrics) ObserveKafkaRetry(store string, count int)

func (*Metrics) ObserveKafkaWorker ¶

func (m *Metrics) ObserveKafkaWorker(store string, status string, count int)

func (*Metrics) ObserveMergeConflict ¶

func (m *Metrics) ObserveMergeConflict(store string, count int)

func (*Metrics) ObserveMergeExhausted ¶

func (m *Metrics) ObserveMergeExhausted(store string, count int)

func (*Metrics) ObserveMergeFold ¶

func (m *Metrics) ObserveMergeFold(store string, operations int)

func (*Metrics) ObserveQuarantined ¶

func (m *Metrics) ObserveQuarantined(store string)

func (*Metrics) ObserveRequestQueue ¶

func (m *Metrics) ObserveRequestQueue(observation RequestQueueObservation)

func (*Metrics) ObserveWorkerCommitted ¶

func (m *Metrics) ObserveWorkerCommitted(store string, oldest time.Time)

func (*Metrics) ObserveWorkerPoll ¶

func (m *Metrics) ObserveWorkerPoll(store string, errors int)

func (*Metrics) ObserveWorkerRecovery ¶

func (m *Metrics) ObserveWorkerRecovery(store string)

func (*Metrics) ObserveWritePhase ¶

func (m *Metrics) ObserveWritePhase(observation WritePhaseObservation)

func (*Metrics) ObserveWriteRounds ¶

func (m *Metrics) ObserveWriteRounds(store string, reads, writes int)

func (*Metrics) Register ¶

func (m *Metrics) Register(collector prometheus.Collector) error

func (*Metrics) RequestStore ¶

func (m *Metrics) RequestStore(request any) string

RequestStore uses request addresses so failed operations retain attribution.

func (*Metrics) RequestStores ¶

func (m *Metrics) RequestStores(stores []string) string

RequestStores attributes a request once, even when it targets several stores.

func (*Metrics) SetWorkerOffsetGap ¶

func (m *Metrics) SetWorkerOffsetGap(store string, gap bool)

func (*Metrics) SetWorkerPending ¶

func (m *Metrics) SetWorkerPending(store string, oldest time.Time, count int)

func (*Metrics) StreamServerInterceptor ¶

func (m *Metrics) StreamServerInterceptor() grpc.StreamServerInterceptor

func (*Metrics) UnaryServerInterceptor ¶

func (m *Metrics) UnaryServerInterceptor() grpc.UnaryServerInterceptor

type RequestQueueObservation ¶

type RequestQueueObservation struct {
	Store    string
	Method   string
	Outcome  string
	Duration time.Duration
}

type WritePhaseObservation ¶

type WritePhaseObservation struct {
	// Store identifies the configured store or a fixed request fallback.
	Store      string
	Completion sink.CompletionMode
	Phase      string
	Duration   time.Duration
}

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL