River metrics have arrived

We've been publishing otelriver for some time now, giving River users an easy way to send telemetry to any provider that supports OpenTelemetry (OTel). It's implemented as middleware that wraps workers and provides key information like how many jobs are worked, what kinds they are, and how long each run takes.

Those stats are undoubtedly important, but there are other equally important areas where River has traditionally provided little insight. Years ago I wrote about how database-backed job queues can fall apart in the presence of long-running queries that cause table bloat. Under these degenerate conditions, the time taken to work an individual job actually stays the same — the problem is that the rate at which jobs are worked drops by a factor of about a thousand.

To monitor for an unhealthy queue, the timing you want to keep an eye on is how long it takes a client to lock a batch of jobs for work — that is, how long it takes to run the core SELECT ... FOR UPDATE ... SKIP LOCKED query that iterates over the job table to find candidates to lock. If the query needs to iterate over thousands of dead rows before it finds live ones, you'll see this number careen into the stratosphere.

Until now, this query's been difficult to instrument by virtue of living deep in River's codebase. Go as a general rule has very little in the way of dynamic programming available, making the query difficult for user code to reach without first-party support. With River v0.41, that first-party support has arrived with the introduction of HookMetricEmit, a River hook that attaches a metric receiver:

type HookMetricEmit interface {
    Hook

    MetricEmit(ctx context.Context, params *HookMetricEmitParams)
}

In its first iteration, the metrics subsystem sends one of two metric types to MetricEmit:

• MetricNameJobGetAvailableDuration (time.Duration): Time taken to lock a batch of jobs waiting to run.

• MetricNameJobGetAvailableCount (int): Number of jobs in a locked batch (when possible, each client locks multiple jobs for efficiency).

Staying future-compatible

River is intentionally starting with a small set of metrics, but future releases will expand the set considerably. Ideally, River will provide visibility into all its main database operations, including leader election, job cleaning, job rescue, job scheduling, and reindexing.

To stay forward-compatible, custom HookMetricEmit implementations should use a type switch on params.Metric and safely discard unrecognized metric types using a default case to protect against a future River version accidentally breaking the integration:

func (p *metricsPlugin) MetricEmit(ctx context.Context, params *rivertype.HookMetricEmitParams) {
    switch metric := params.Metric.(type) {
    case *rivertype.JobGetAvailableDurationMetric:
        p.recordFetchDuration(ctx, metric.Queue, metric.Duration)

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

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

Keeping metrics paths non-blocking

Metric emission occurs in some of River's hottest paths, such as when a client is locking jobs for work. HookMetricEmit handlers should always be non-blocking to make sure they don't slow anything down:

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

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

go func() {
    for metric := range metricChan {
        exportMetric(metric)
    }
}()

Any persistence, network access, or processing-heavy code should run out of band by passing values over a channel or the like.

We could've designed the metrics subsystem to push metrics through channels internally, but many metric packages already decouple recording from network export. Leaving that responsibility to integrations avoids an unnecessary layer of indirection in what's a relatively common case.

Plug and play with otelriver

HookMetricEmit is explicitly designed as low-level infrastructure. It's meant to let users receive a metric and perform any action — you could send it to your metrics provider, but you could also log it or store it in Redis. It's there to provide flexibility and keep River maximally extensible.

For a more batteries-included approach, otelriver has been updated to implement HookMetricEmit and emit job-fetch metrics alongside the work statistics it already provides. Just add it to Config.Plugins and you're ready to go:

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

Notably, we'd previously recommended registering otelriver under Config.Middleware. To receive the new metrics, move it to Config.Plugins, which detects both hook and middleware implementations on configured objects.

Recap + try metrics today

In short:

• River v0.41 introduces HookMetricEmit, a low-level listener for metrics that enables arbitrary integrations.

• Two metrics for (1) time to lock a batch of jobs (MetricNameJobGetAvailableDuration) and (2) number of jobs locked in a batch (MetricNameJobGetAvailableCount) are available immediately. This set will be expanded in the future.

• otelriver has been updated to implement the new hook and will send the new metrics to any OTel-compatible provider, such as Datadog.

Upgrade River today and try its new metrics subsystem for yourself. As usual, we'd love to hear from you about how it went and what you think we should work on next.