Add support for metrics for time to lock jobs and locked count

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Add `otelriver` hook support for River producer metrics, including `river.job_get_available_duration` and `river.job_get_available_count`. [PR #64](https://github.com/riverqueue/rivercontrib/pull/64).
+
 ## [0.11.0] - 2026-07-02
 
 ### Added

otelriver/example_middleware_test.go

@@ -14,9 +14,7 @@ import (
 func ExampleMiddleware() {
 	_, err := river.NewClient(riverpgxv5.New(nil), &river.Config{
 		Logger: slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelWarn, ReplaceAttr: slogutil.NoLevelTime})),
-		Middleware: []rivertype.Middleware{
-			// Install the OpenTelemetry middleware to run for all jobs inserted
-			// or worked by this River client.
+		Plugins: []rivertype.Plugin{
 			otelriver.NewMiddleware(nil),
 		},
 		TestOnly: true, // suitable only for use in tests; remove for live environments

otelriver/go.mod

@@ -3,10 +3,10 @@ module github.com/riverqueue/rivercontrib/otelriver
 go 1.25.0
 
 require (
-	github.com/riverqueue/river v0.29.0
-	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.29.0
-	github.com/riverqueue/river/rivershared v0.29.0
-	github.com/riverqueue/river/rivertype v0.29.0
+	github.com/riverqueue/river v0.41.0
+	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.0
+	github.com/riverqueue/river/rivershared v0.41.0
+	github.com/riverqueue/river/rivertype v0.41.0
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/otel v1.43.0
 	go.opentelemetry.io/otel/metric v1.43.0
@@ -23,19 +23,19 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
-	github.com/jackc/pgx/v5 v5.9.2 // indirect
+	github.com/jackc/pgx/v5 v5.10.0 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/riverqueue/river/riverdriver v0.29.0 // indirect
-	github.com/tidwall/gjson v1.18.0 // indirect
-	github.com/tidwall/match v1.1.1 // indirect
+	github.com/riverqueue/river/riverdriver v0.41.0 // indirect
+	github.com/tidwall/gjson v1.19.0 // indirect
+	github.com/tidwall/match v1.2.0 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect
 	github.com/tidwall/sjson v1.2.5 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.uber.org/goleak v1.3.0 // indirect
-	golang.org/x/sync v0.19.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
-	golang.org/x/text v0.32.0 // indirect
+	golang.org/x/text v0.40.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 

otelriver/go.sum

@@ -18,8 +18,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw=
-github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
+github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
@@ -28,16 +28,16 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/riverqueue/river v0.29.0 h1:PMO4k6n7HcIjjgrbnG2UG04Exh8aLmQksOddOoYDASA=
-github.com/riverqueue/river v0.29.0/go.mod h1:S8BbQbxCrJLYygmnrnraltHhWlGzZzwjqcRbY3wdq7w=
-github.com/riverqueue/river/riverdriver v0.29.0 h1:o7mV07RPXrGJdwXUKxVTOyvG1/cDmJIMI3V4Le4/LBo=
-github.com/riverqueue/river/riverdriver v0.29.0/go.mod h1:bmkdn74EG4Ogsv44JkC1CBxFZ3JHfYsN+e0K8Dq0otU=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.29.0 h1:l3D17JWq/00QEt0bcawyDMxZYmM1YAk11Y/nRRVk5C8=
-github.com/riverqueue/river/riverdriver/riverpgxv5 v0.29.0/go.mod h1:mpncN3m7DR7VpD78LV5CczbSpwkWcLeJ5j1kkJiOt9s=
-github.com/riverqueue/river/rivershared v0.29.0 h1:Niwbmp/CQAKPZ+zT3teCgEmPhksyW0f2cx4X03FurEk=
-github.com/riverqueue/river/rivershared v0.29.0/go.mod h1:74WjXTYKV4nTfLemIPloPqiA3Tjqe5BFvnALrNbS62k=
-github.com/riverqueue/river/rivertype v0.29.0 h1:26hpzbd44piqJZ+1zO4RO6GRKpmZVX3Ncx+Ki+w2gtg=
-github.com/riverqueue/river/rivertype v0.29.0/go.mod h1:rWpgI59doOWS6zlVocROcwc00fZ1RbzRwsRTU8CDguw=
+github.com/riverqueue/river v0.41.0 h1:E7Yfyhn74IgVaCBKsBpTHoC0LxWJvoG0jKt7Tb+lJZg=
+github.com/riverqueue/river v0.41.0/go.mod h1:WXiAF1/2gfUPj3H+WqQG3Q5NW5w9JmRXRTe7U9zybNc=
+github.com/riverqueue/river/riverdriver v0.41.0 h1:A5g80n6fCGu9Bp4hSq7gys7mvfOiuWkbQHuFx3iPp7w=
+github.com/riverqueue/river/riverdriver v0.41.0/go.mod h1:OeGOhZYoH7Ac0CrVanNAAwD9C0WAXy4LzUEl65Ex0B4=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.0 h1:uHGluToMMWvCnpNkAbEWEmP4YAjdseuriEXGu1dwOp4=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.41.0/go.mod h1:0ueUi+3eW5fIsibS5VkD2HzQmwLKTAiM/uiiVMX/SME=
+github.com/riverqueue/river/rivershared v0.41.0 h1:ax3uz5KiqfmcvRQ3kKB6iYs9Nt8J5L1RNfOKoVMWkqw=
+github.com/riverqueue/river/rivershared v0.41.0/go.mod h1:FaZ7bxC2DORhyFDVyHKRiZzgQPf2b/IELwPBXnHZVLA=
+github.com/riverqueue/river/rivertype v0.41.0 h1:dfscvt1asf1PpeeHTMFxQdV1ZfMoReiiMNFCPPfR0as=
+github.com/riverqueue/river/rivertype v0.41.0/go.mod h1:D1Ad+EaZiaXbQbJcJcfeicXJMBKno0n6UcfKI5Q7DIQ=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
@@ -48,10 +48,11 @@ github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
-github.com/tidwall/gjson v1.18.0 h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY=
-github.com/tidwall/gjson v1.18.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
-github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
+github.com/tidwall/gjson v1.19.0 h1:xwxm7n691Uf3u5OFjzngavjGTh55KX5q/9w9xHW88JU=
+github.com/tidwall/gjson v1.19.0/go.mod h1:V37/opeE/JbLUOfH0QTXiNez2l0RUjYUhpT4szFQAfc=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/match v1.2.0 h1:0pt8FlkOwjN2fPt4bIl4BoNxb98gGHN2ObFEDkrfZnM=
+github.com/tidwall/match v1.2.0/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
 github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
@@ -71,12 +72,12 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09
 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
-golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
 golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU=
-golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY=
+golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
+golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

otelriver/middleware.go

@@ -66,10 +66,10 @@ type MiddlewareConfig struct {
 	TracerProvider trace.TracerProvider
 }
 
-// Middleware is a River middleware that emits OpenTelemetry metrics when jobs
-// are inserted or worked.
+// Middleware is a River middleware and hook that emits OpenTelemetry traces and
+// metrics.
 type Middleware struct {
-	river.MiddlewareDefaults
+	river.PluginDefaults
 
 	config  *MiddlewareConfig
 	meter   metric.Meter
@@ -83,6 +83,8 @@ type middlewareMetrics struct {
 	insertManyCount                  metric.Int64Counter
 	insertManyDuration               metric.Float64Gauge
 	insertManyDurationHistogram      metric.Float64Histogram
+	jobGetAvailableDuration          metric.Float64Histogram
+	jobGetAvailableCount             metric.Int64Histogram
 	messagingClientConsumedMessages  metric.Int64Counter
 	messagingClientOperationDuration metric.Float64Histogram
 	messagingClientSentMessages      metric.Int64Counter
@@ -125,6 +127,8 @@ func NewMiddleware(config *MiddlewareConfig) *Middleware {
 		insertManyCount:             mustInt64Counter(meter, prefix+"insert_many_count", metric.WithDescription("Number of job batches inserted (all jobs are inserted in a batch, but batches may be one job)"), metric.WithUnit("{job_batch}")),
 		insertManyDuration:          mustFloat64Gauge(meter, prefix+"insert_many_duration", metric.WithDescription("Duration of job batch insertion"), metric.WithUnit(durationUnit)),
 		insertManyDurationHistogram: mustFloat64Histogram(meter, prefix+"insert_many_duration_histogram", metric.WithDescription("Duration of job batch insertion (histogram)"), metric.WithUnit(durationUnit)),
+		jobGetAvailableDuration:     mustFloat64Histogram(meter, prefix+"job_get_available_duration", metric.WithDescription("Duration of successful JobGetAvailable calls"), metric.WithUnit(durationUnit)),
+		jobGetAvailableCount:        mustInt64Histogram(meter, prefix+"job_get_available_count", metric.WithDescription("Number of jobs locked by successful JobGetAvailable calls"), metric.WithUnit("{job}")),
 		workCount:                   mustInt64Counter(meter, prefix+"work_count", metric.WithDescription("Number of jobs worked"), metric.WithUnit("{job}")),
 		workDuration:                mustFloat64Gauge(meter, prefix+"work_duration", metric.WithDescription("Duration of job being worked"), metric.WithUnit(durationUnit)),
 		workDurationHistogram:       mustFloat64Histogram(meter, prefix+"work_duration_histogram", metric.WithDescription("Duration of job being worked (histogram)"), metric.WithUnit(durationUnit)),
@@ -227,6 +231,21 @@ func (m *Middleware) InsertMany(ctx context.Context, manyParams []*rivertype.Job
 	return insertRes, err
 }
 
+func (m *Middleware) MetricEmit(ctx context.Context, params *rivertype.HookMetricEmitParams) {
+	if params == nil {
+		return
+	}
+
+	switch riverMetric := params.Metric.(type) {
+	case *rivertype.JobGetAvailableDurationMetric:
+		m.metrics.jobGetAvailableDuration.Record(ctx, m.durationInPreferredUnit(riverMetric.Duration),
+			metric.WithAttributes(attribute.String("queue", riverMetric.Queue)))
+	case *rivertype.JobGetAvailableCountMetric:
+		m.metrics.jobGetAvailableCount.Record(ctx, int64(riverMetric.Count),
+			metric.WithAttributes(attribute.String("queue", riverMetric.Queue)))
+	}
+}
+
 func (m *Middleware) Work(ctx context.Context, job *rivertype.JobRow, doInner func(context.Context) error) error {
 	spanName := prefix + "work"
 	if m.config.EnableWorkSpanJobKindSuffix {
@@ -352,6 +371,14 @@ func mustFloat64Histogram(meter metric.Meter, name string, options ...metric.Flo
 	return metric
 }
 
+func mustInt64Histogram(meter metric.Meter, name string, options ...metric.Int64HistogramOption) metric.Int64Histogram {
+	metric, err := meter.Int64Histogram(name, options...)
+	if err != nil {
+		panic(err)
+	}
+	return metric
+}
+
 func mustInt64Counter(meter metric.Meter, name string, options ...metric.Int64CounterOption) metric.Int64Counter {
 	metric, err := meter.Int64Counter(name, options...)
 	if err != nil {

otelriver/middleware_test.go

@@ -24,6 +24,7 @@ import (
 // Verify interface compliance.
 var (
 	_ rivertype.JobInsertMiddleware = &Middleware{}
+	_ rivertype.HookMetricEmit      = &Middleware{}
 	_ rivertype.WorkerMiddleware    = &Middleware{}
 )
 
@@ -60,6 +61,68 @@ func TestMiddleware(t *testing.T) {
 		return setupConfig(t, &MiddlewareConfig{})
 	}
 
+	t.Run("MetricJobGetAvailableDuration", func(t *testing.T) {
+		t.Parallel()
+
+		middleware, bundle := setup(t)
+
+		middleware.MetricEmit(ctx, &rivertype.HookMetricEmitParams{
+			Metric: &rivertype.JobGetAvailableDurationMetric{
+				Duration: 2500 * time.Millisecond,
+				Queue:    "critical",
+			},
+		})
+
+		var metrics metricdata.ResourceMetrics
+		require.NoError(t, bundle.metricReader.Collect(ctx, &metrics))
+		metric, metricData := requireHistogramCount(t, metrics, "river.job_get_available_duration", 1,
+			attribute.String("queue", "critical"))
+		require.Equal(t, "s", metric.Unit)
+		require.InDelta(t, 2.5, metricData.DataPoints[0].Sum, 0.001)
+	})
+
+	t.Run("MetricJobGetAvailableDurationUnitMS", func(t *testing.T) {
+		t.Parallel()
+
+		middleware, bundle := setupConfig(t, &MiddlewareConfig{
+			DurationUnit: "ms",
+		})
+
+		middleware.MetricEmit(ctx, &rivertype.HookMetricEmitParams{
+			Metric: &rivertype.JobGetAvailableDurationMetric{
+				Duration: 2500 * time.Millisecond,
+				Queue:    "critical",
+			},
+		})
+
+		var metrics metricdata.ResourceMetrics
+		require.NoError(t, bundle.metricReader.Collect(ctx, &metrics))
+		metric, metricData := requireHistogramCount(t, metrics, "river.job_get_available_duration", 1,
+			attribute.String("queue", "critical"))
+		require.Equal(t, "ms", metric.Unit)
+		require.InDelta(t, 2500.0, metricData.DataPoints[0].Sum, 0.001)
+	})
+
+	t.Run("MetricJobGetAvailableCount", func(t *testing.T) {
+		t.Parallel()
+
+		middleware, bundle := setup(t)
+
+		middleware.MetricEmit(ctx, &rivertype.HookMetricEmitParams{
+			Metric: &rivertype.JobGetAvailableCountMetric{
+				Count: 42,
+				Queue: "critical",
+			},
+		})
+
+		var metrics metricdata.ResourceMetrics
+		require.NoError(t, bundle.metricReader.Collect(ctx, &metrics))
+		metric, metricData := requireInt64HistogramCount(t, metrics, "river.job_get_available_count", 1,
+			attribute.String("queue", "critical"))
+		require.Equal(t, "{job}", metric.Unit)
+		require.EqualValues(t, 42, metricData.DataPoints[0].Sum)
+	})
+
 	t.Run("InsertManySuccess", func(t *testing.T) {
 		t.Parallel()
 
@@ -902,6 +965,15 @@ func requireHistogramCount(t *testing.T, metrics metricdata.ResourceMetrics, nam
 	return metric, metricData
 }
 
+func requireInt64HistogramCount(t *testing.T, metrics metricdata.ResourceMetrics, name string, count uint64, attrs ...attribute.KeyValue) (metricdata.Metrics, metricdata.Histogram[int64]) { //nolint:unparam
+	t.Helper()
+
+	metric, metricData := requireMetric[metricdata.Histogram[int64]](t, metrics, name)
+	require.Equal(t, count, metricData.DataPoints[0].Count)
+	metricdatatest.AssertHasAttributes(t, metric, attrs...)
+	return metric, metricData
+}
+
 func requireNoMetric(t *testing.T, metrics metricdata.ResourceMetrics, name string) {
 	t.Helper()