Metrics

River emits metrics from internal operations that application code can't otherwise instrument directly. Starting in River 0.41, clients can receive these metrics through HookMetricEmit or export them to an OpenTelemetry-compatible provider with otelriver.

These metrics include the duration of River's core job-fetch query, which can provide an early warning that a queue is slowing down even when individual jobs continue to run at their usual speed.

Building a custom integration

Custom integrations can implement HookMetricEmit directly or use river.HookMetricEmitFunc. Use a type switch with a default case so that the integration continues to work when a future River version adds another metric type. The current payload types are described in the next section.

metricChan := make(chan rivertype.Metric, 100)

metricPlugin := river.HookMetricEmitFunc(func(
    _ context.Context,
    params *rivertype.HookMetricEmitParams,
) {
    select {
    case metricChan <- params.Metric:
    default:
        // Drop the metric rather than block River's job-fetch loop.
    }
})

go func() {
    for metric := range metricChan {
        switch metric := metric.(type) {
        case *rivertype.JobGetAvailableDurationMetric:
            recordFetchDuration(metric.Queue, metric.Duration)

        case *rivertype.JobGetAvailableCountMetric:
            recordFetchCount(metric.Queue, metric.Count)

        default:
            // Ignore metrics added by future River versions.
        }
    }
}()

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Plugins: []rivertype.Plugin{metricPlugin},
    // ...
})

Metric hooks run synchronously in hot paths, including while River is fetching jobs. They must not perform network access, persistence, or expensive processing inline. Use a metrics library that records asynchronously, or hand metrics to a bounded channel as shown above. If the channel is full, dropping a metric is preferable to slowing down the job-fetch loop.

Available metrics

The example above handles the two metric payloads River currently emits after a successful job-fetch operation:

• JobGetAvailableDurationMetric: How long it took to lock a batch of available jobs. Its fields contain duration and queue name. The timing of this metric growing out of hand is an early sign of database degradation (e.g. from a long-running query) that may lead to queue degradation.
• JobGetAvailableCountMetric: How many jobs were locked in the batch. Its fields contain the count and queue name.

Each payload implements rivertype.Metric and has a corresponding name, MetricNameJobGetAvailableDuration or MetricNameJobGetAvailableCount. More metric types may be added in future River versions.

Plug and play OpenTelemetry with otelriver

For a ready-made integration that handles these payloads, install otelriver:

go get -u github.com/riverqueue/rivercontrib/otelriver

Add it to Config.Plugins:

import (
    "github.com/riverqueue/river"
    "github.com/riverqueue/river/riverdriver/riverpgxv5"
    "github.com/riverqueue/river/rivertype"
    "github.com/riverqueue/rivercontrib/otelriver"
)

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Plugins: []rivertype.Plugin{
        otelriver.NewMiddleware(nil),
    },
    // ...
})
if err != nil {
    panic(err)
}

Despite its constructor's name, otelriver.NewMiddleware implements both River middleware and HookMetricEmit. Registering it under Config.Plugins activates both interfaces, enabling its existing insertion and work instrumentation along with the internal River metrics. An existing configuration under Config.Middleware should be moved to Config.Plugins to receive the internal metrics.

otelriver exports the job-fetch payloads as:

• river.job_get_available_duration: A histogram of successful job-fetch durations, tagged by queue.
• river.job_get_available_count: A histogram of the number of jobs locked by successful fetches, tagged by queue.

See OpenTelemetry for provider configuration and the other traces and metrics available through otelriver.