prometheus

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 10, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package prometheus provides OpenTelemetry-based metrics instrumentation for the cleat workflow engine. Metrics are recorded via OTel instruments and exposed in Prometheus exposition format on the /metrics HTTP endpoint.

Usage: 

m, err := prometheus.New(prometheus.Config{
    WorkerID:  "abc123",
})
if err != nil {
    log.Fatal(err)
}
defer m.Shutdown(context.Background())

m.RecordWorkflowStarted(ctx, "order_processing")
m.RecordWorkflowCompleted(ctx, "order_processing")

http.Handle("/metrics", m.ServeHTTP())

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Config 

type Config struct {
	// WorkerID uniquely identifies this worker instance.
	WorkerID string
	// ServiceName identifies the service for the OTel meter.
	ServiceName string
}

Config holds configuration for the Metrics instance.

type Metrics 

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

Metrics provides OpenTelemetry-based metrics instrumentation for the cleat workflow engine. All methods are safe for concurrent use -- OTel instruments are thread-safe by design.

func New 

func New(cfg Config) (*Metrics, error)

New creates a new Metrics instance with the given configuration. It initialises all OTel instruments and starts a manual reader that will be polled on each /metrics request.

func (*Metrics) AddCompactionEventsDeleted 

func (m *Metrics) AddCompactionEventsDeleted(ctx context.Context, delta int64, extraAttrs ...attribute.KeyValue)

AddCompactionEventsDeleted adds to the compaction-events-deleted counter.

func (*Metrics) AddWorkflowActive 

func (m *Metrics) AddWorkflowActive(ctx context.Context, delta int64, workflowName string, extraAttrs ...attribute.KeyValue)

AddWorkflowActive increments or decrements the active workflow gauge. Pass a positive value to increment, negative to decrement.

func (*Metrics) RecordAmbiguousCall 

func (m *Metrics) RecordAmbiguousCall(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordAmbiguousCall increments the ambiguous-calls counter.

func (*Metrics) RecordBackgroundLoop 

func (m *Metrics) RecordBackgroundLoop(ctx context.Context, loopName, status string, extraAttrs ...attribute.KeyValue)

RecordBackgroundLoop increments the background-loops counter.

func (*Metrics) RecordBackgroundLoopRestart 

func (m *Metrics) RecordBackgroundLoopRestart(ctx context.Context, loopName string, count int64, extraAttrs ...attribute.KeyValue)

RecordBackgroundLoopRestart increments the background-loop-restarts counter.

func (*Metrics) RecordCall 

func (m *Metrics) RecordCall(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordCall increments the durable-call-invocations counter.

func (*Metrics) RecordCallRetry 

func (m *Metrics) RecordCallRetry(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordCallRetry increments the DurableCallWithRetry retry counter. Call once per retry that actually happens (i.e. a failed attempt followed by another attempt), not once per attempt -- so a retry storm across many calls is visible as a rate, distinct from ordinary call volume, which freshCallWithRetry's per-call event history does not surface anywhere this metrics package can see.

func (*Metrics) RecordClaimLatency 

func (m *Metrics) RecordClaimLatency(ctx context.Context, duration time.Duration, workflowName string, extraAttrs ...attribute.KeyValue)

RecordClaimLatency records the time taken to claim a workflow.

func (*Metrics) RecordConcurrencyLimit 

func (m *Metrics) RecordConcurrencyLimit(ctx context.Context, limit int64, extraAttrs ...attribute.KeyValue)

RecordConcurrencyLimit records the current effective concurrency cap. Converts the absolute value to a delta for the UpDownCounter.

func (*Metrics) RecordContinueAsNew 

func (m *Metrics) RecordContinueAsNew(ctx context.Context, reason string, extraAttrs ...attribute.KeyValue)

RecordContinueAsNew increments the continue-as-new counter by reason.

func (*Metrics) RecordDBQueryLatency 

func (m *Metrics) RecordDBQueryLatency(ctx context.Context, duration time.Duration, operation string, extraAttrs ...attribute.KeyValue)

RecordDBQueryLatency records a database query duration.

func (*Metrics) RecordDecryptionError 

func (m *Metrics) RecordDecryptionError(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordDecryptionError increments the decryption-errors counter.

func (*Metrics) RecordDesiredConcurrency 

func (m *Metrics) RecordDesiredConcurrency(ctx context.Context, desired int64, extraAttrs ...attribute.KeyValue)

RecordDesiredConcurrency records the configured --concurrency value. Converts the absolute value to a delta for the UpDownCounter.

func (*Metrics) RecordDispatchLatency 

func (m *Metrics) RecordDispatchLatency(ctx context.Context, duration time.Duration, taskQueue string, extraAttrs ...attribute.KeyValue)

RecordDispatchLatency records time from workflow creation to claim by task_queue.

func (*Metrics) RecordEncryptionError 

func (m *Metrics) RecordEncryptionError(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordEncryptionError increments the encryption-errors counter.

func (*Metrics) RecordEventsDeleted 

func (m *Metrics) RecordEventsDeleted(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

RecordEventsDeleted adds to the events-deleted counter.

func (*Metrics) RecordFreshDuration 

func (m *Metrics) RecordFreshDuration(ctx context.Context, duration time.Duration, defName string, extraAttrs ...attribute.KeyValue)

RecordFreshDuration records the duration of first-run workflow execution by def_name.

func (*Metrics) RecordFreshStep 

func (m *Metrics) RecordFreshStep(ctx context.Context, workflowName string, extraAttrs ...attribute.KeyValue)

RecordFreshStep increments the fresh-steps-executed counter. workflowName is recorded as a label so the runner can scope step counts to the correct variant (matching cleat_workflows_completed_total).

func (*Metrics) RecordHTTPRequest 

func (m *Metrics) RecordHTTPRequest(ctx context.Context, method, path, status string, extraAttrs ...attribute.KeyValue)

RecordHTTPRequest increments the HTTP requests counter.

func (*Metrics) RecordHTTPRequestDuration 

func (m *Metrics) RecordHTTPRequestDuration(ctx context.Context, duration time.Duration, method, path string, extraAttrs ...attribute.KeyValue)

RecordHTTPRequestDuration records an HTTP request duration for backendkit apps.

func (*Metrics) RecordMemoryAvailable 

func (m *Metrics) RecordMemoryAvailable(ctx context.Context, bytes int64, extraAttrs ...attribute.KeyValue)

RecordMemoryAvailable records available memory in bytes. Converts the absolute value to a delta for the UpDownCounter.

func (*Metrics) RecordMemoryRSS 

func (m *Metrics) RecordMemoryRSS(ctx context.Context, bytes int64, extraAttrs ...attribute.KeyValue)

RecordMemoryRSS records worker process RSS in bytes. Converts the absolute value to a delta for the UpDownCounter.

func (*Metrics) RecordMemoryTotal 

func (m *Metrics) RecordMemoryTotal(ctx context.Context, bytes int64, extraAttrs ...attribute.KeyValue)

RecordMemoryTotal records total allocatable memory in bytes. Converts the absolute value to a delta for the UpDownCounter.

func (*Metrics) RecordPluginCallDuration 

func (m *Metrics) RecordPluginCallDuration(ctx context.Context, duration time.Duration, pluginName, functionName string, extraAttrs ...attribute.KeyValue)

RecordPluginCallDuration records a plugin call duration by plugin and function.

func (*Metrics) RecordPollWaitDuration 

func (m *Metrics) RecordPollWaitDuration(ctx context.Context, duration time.Duration, extraAttrs ...attribute.KeyValue)

RecordPollWaitDuration records the time spent waiting for work in poll queries.

func (*Metrics) RecordReaperInstanceClaimed 

func (m *Metrics) RecordReaperInstanceClaimed(ctx context.Context, status string, extraAttrs ...attribute.KeyValue)

RecordReaperInstanceClaimed increments the reaper-instances-claimed counter.

func (*Metrics) RecordReplayChecksumFailure 

func (m *Metrics) RecordReplayChecksumFailure(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordReplayChecksumFailure increments the replay-checksum-failures counter.

func (*Metrics) RecordReplayDuration 

func (m *Metrics) RecordReplayDuration(ctx context.Context, duration time.Duration, extraAttrs ...attribute.KeyValue)

RecordReplayDuration records the duration of workflow replay execution.

func (*Metrics) RecordReplayFailure 

func (m *Metrics) RecordReplayFailure(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordReplayFailure increments the replay-failures counter.

func (*Metrics) RecordReplayStep 

func (m *Metrics) RecordReplayStep(ctx context.Context, defName string, extraAttrs ...attribute.KeyValue)

RecordReplayStep increments the replay-steps counter by def_name.

func (*Metrics) RecordWasmCacheHit 

func (m *Metrics) RecordWasmCacheHit(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordWasmCacheHit increments the wasm-cache-hits counter.

func (*Metrics) RecordWasmCacheMiss 

func (m *Metrics) RecordWasmCacheMiss(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordWasmCacheMiss increments the wasm-cache-misses counter.

func (*Metrics) RecordWasmCompileDuration 

func (m *Metrics) RecordWasmCompileDuration(ctx context.Context, duration time.Duration, defName string, extraAttrs ...attribute.KeyValue)

RecordWasmCompileDuration records WASM compile duration by def_name.

func (*Metrics) RecordWasmFuelExhausted 

func (m *Metrics) RecordWasmFuelExhausted(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordWasmFuelExhausted increments the wasm-fuel-exhausted counter.

func (*Metrics) RecordWasmLoadLatency 

func (m *Metrics) RecordWasmLoadLatency(ctx context.Context, duration time.Duration, defName string, extraAttrs ...attribute.KeyValue)

RecordWasmLoadLatency records the time taken to load a WASM module.

func (*Metrics) RecordWorkflowCompleted 

func (m *Metrics) RecordWorkflowCompleted(ctx context.Context, workflowName string, taskQueue string, extraAttrs ...attribute.KeyValue)

RecordWorkflowCompleted increments the workflows-completed counter. workflowName and taskQueue are recorded as labels.

func (*Metrics) RecordWorkflowDuration 

func (m *Metrics) RecordWorkflowDuration(ctx context.Context, duration time.Duration, workflowName string, status string, taskQueue string, extraAttrs ...attribute.KeyValue)

RecordWorkflowDuration records the total execution duration of a workflow.

func (*Metrics) RecordWorkflowFailed 

func (m *Metrics) RecordWorkflowFailed(ctx context.Context, workflowName string, errMsg string, taskQueue string, extraAttrs ...attribute.KeyValue)

RecordWorkflowFailed increments the workflows-failed counter. workflowName, error, and taskQueue are recorded as labels.

func (*Metrics) RecordWorkflowMemoryEstimate 

func (m *Metrics) RecordWorkflowMemoryEstimate(ctx context.Context, defName string, bytes float64, extraAttrs ...attribute.KeyValue)

RecordWorkflowMemoryEstimate records the estimated memory per workflow execution by def_name. Converts the absolute value to a delta for the UpDownCounter.

func (*Metrics) RecordWorkflowStarted 

func (m *Metrics) RecordWorkflowStarted(ctx context.Context, workflowName string, extraAttrs ...attribute.KeyValue)

RecordWorkflowStarted increments the workflows-started counter. workflowName is recorded as a label.

func (*Metrics) RecordWorkflowsClaimed 

func (m *Metrics) RecordWorkflowsClaimed(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

RecordWorkflowsClaimed increments the workflows-claimed counter by count.

func (*Metrics) RecordWorkflowsDeadLettered 

func (m *Metrics) RecordWorkflowsDeadLettered(ctx context.Context, extraAttrs ...attribute.KeyValue)

RecordWorkflowsDeadLettered increments the workflows-dead-lettered counter.

func (*Metrics) RecordWorkflowsPurged 

func (m *Metrics) RecordWorkflowsPurged(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

RecordWorkflowsPurged adds to the workflows-purged counter: how many completed workflow_instances rows the completed-workflow retention policy has permanently deleted, so an operator can see retention working (or notice it silently doing nothing) rather than inferring it from table size.

func (*Metrics) ServeHTTP 

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

ServeHTTP returns an http.Handler that serves the current metrics in Prometheus exposition format (text/plain; version=0.0.4).

func (*Metrics) SetBackgroundLoopDuration 

func (m *Metrics) SetBackgroundLoopDuration(ctx context.Context, loopName string, seconds float64, extraAttrs ...attribute.KeyValue)

SetBackgroundLoopDuration sets the background loop duration gauge.

func (*Metrics) SetBackgroundLoopItemsProcessed 

func (m *Metrics) SetBackgroundLoopItemsProcessed(ctx context.Context, loopName string, count int64, extraAttrs ...attribute.KeyValue)

SetBackgroundLoopItemsProcessed sets the background loop items processed gauge.

func (*Metrics) SetBackgroundLoopLastRun 

func (m *Metrics) SetBackgroundLoopLastRun(ctx context.Context, loopName string, ts float64, extraAttrs ...attribute.KeyValue)

SetBackgroundLoopLastRun sets the background loop last run timestamp gauge.

func (*Metrics) SetConcurrencyKeysExpiringSoon 

func (m *Metrics) SetConcurrencyKeysExpiringSoon(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

SetConcurrencyKeysExpiringSoon sets the concurrency keys expiring soon gauge. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetConcurrencyKeysTotal 

func (m *Metrics) SetConcurrencyKeysTotal(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

SetConcurrencyKeysTotal sets the concurrency keys total gauge. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetEventHistoryRowCount 

func (m *Metrics) SetEventHistoryRowCount(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

SetEventHistoryRowCount sets the event history row count gauge. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetEventHistorySize 

func (m *Metrics) SetEventHistorySize(ctx context.Context, sizeBytes int64, workflowName string, extraAttrs ...attribute.KeyValue)

SetEventHistorySize sets the event history size gauge to the given value. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetFreshStepCount 

func (m *Metrics) SetFreshStepCount(ctx context.Context, val int64, extraAttrs ...attribute.KeyValue)

SetFreshStepCount sets the cumulative fresh step count gauge.

func (*Metrics) SetFreshThroughput 

func (m *Metrics) SetFreshThroughput(ctx context.Context, rate float64, extraAttrs ...attribute.KeyValue)

SetFreshThroughput sets the fresh step throughput rate gauge.

func (*Metrics) SetMemoryPressure 

func (m *Metrics) SetMemoryPressure(ctx context.Context, pressure float64, extraAttrs ...attribute.KeyValue)

SetMemoryPressure sets the memory pressure gauge.

func (*Metrics) SetMemoryPressureRatio 

func (m *Metrics) SetMemoryPressureRatio(ctx context.Context, ratio float64, extraAttrs ...attribute.KeyValue)

SetMemoryPressureRatio sets the memory pressure ratio gauge.

func (*Metrics) SetPluginConnectionsInUse 

func (m *Metrics) SetPluginConnectionsInUse(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

SetPluginConnectionsInUse sets the plugin connections in use gauge. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetPluginConnectionsMax 

func (m *Metrics) SetPluginConnectionsMax(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

SetPluginConnectionsMax sets the plugin connections max gauge. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetQueueDepth 

func (m *Metrics) SetQueueDepth(ctx context.Context, depth int64, extraAttrs ...attribute.KeyValue)

SetQueueDepth sets the queue depth gauge.

func (*Metrics) SetReplayStepCount 

func (m *Metrics) SetReplayStepCount(ctx context.Context, val int64, extraAttrs ...attribute.KeyValue)

SetReplayStepCount sets the cumulative replay step count gauge.

func (*Metrics) SetReplayThroughput 

func (m *Metrics) SetReplayThroughput(ctx context.Context, rate float64, extraAttrs ...attribute.KeyValue)

SetReplayThroughput sets the replay step throughput rate gauge.

func (*Metrics) SetRetentionLastRunTimestamp 

func (m *Metrics) SetRetentionLastRunTimestamp(ctx context.Context, ts int64, extraAttrs ...attribute.KeyValue)

SetRetentionLastRunTimestamp sets the retention last run timestamp gauge.

func (*Metrics) SetScalingPressure 

func (m *Metrics) SetScalingPressure(ctx context.Context, pressure float64, extraAttrs ...attribute.KeyValue)

SetScalingPressure sets the composite scaling pressure gauge.

func (*Metrics) SetWasmCacheBytes 

func (m *Metrics) SetWasmCacheBytes(ctx context.Context, bytes int64, extraAttrs ...attribute.KeyValue)

SetWasmCacheBytes sets the WASM cache bytes gauge. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetWasmCacheEntries 

func (m *Metrics) SetWasmCacheEntries(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

SetWasmCacheEntries sets the WASM cache entries gauge. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetWorkerCount 

func (m *Metrics) SetWorkerCount(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

SetWorkerCount sets the worker count gauge to the given value. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) SetWorkflowsStuck 

func (m *Metrics) SetWorkflowsStuck(ctx context.Context, count int64, extraAttrs ...attribute.KeyValue)

SetWorkflowsStuck sets the workflows stuck gauge. Uses delta tracking to convert absolute values to UpDownCounter deltas.

func (*Metrics) Shutdown 

func (m *Metrics) Shutdown(ctx context.Context) error

Shutdown flushes and shuts down the metrics pipeline. It should be called when the worker is shutting down.

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.