diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ced190a..b90dfcf9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added `rivertype.HookMetricEmit` for receiving metrics emitted by River. Initial metrics report the duration of successful job fetches with `JobGetAvailableDurationMetric` and the number of jobs fetched with `JobGetAvailableCountMetric`. [PR #1285](https://github.com/riverqueue/river/pull/1285). + ### Changed - Both hooks and middleware should now be registered under a general `Config.Plugins` instead, a change that simpifies things somewhat, and better handles cases where a hook or middleware might be _both_ a hook and a middleware as opposed to only one or the other. `Config.Hooks` and `Config.Middleware` are still available for use, but considered deprecated in favor of the more general `Config.Plugins`. [PR #1284](https://github.com/riverqueue/river/pull/1284). diff --git a/client.go b/client.go index ef4b36e0..071638b4 100644 --- a/client.go +++ b/client.go @@ -226,11 +226,11 @@ type Config struct { // lifecycle (see rivertype.Hook), installed globally. // // The effect of hooks in this list will depend on the specific hook - // interfaces they implement, so for example implementing - // rivertype.HookInsertBegin will cause the hook to be invoked before a job - // is inserted, or implementing rivertype.HookWorkBegin will cause it to be - // invoked before a job is worked. Hook structs may implement multiple hook - // interfaces. + // interfaces they implement. rivertype.HookInsertBegin will cause the hook + // to be invoked before a job is inserted. rivertype.HookMetricEmit will + // cause the hook to be invoked when River emits a metric. Implementing + // rivertype.HookWorkBegin will cause it to be invoked before a job is + // worked. Hook structs may implement multiple hook interfaces. // // Order in this list is significant. A hook that appears first will be // entered before a hook that appears later. For any particular phase, order diff --git a/hook_defaults_funcs.go b/hook_defaults_funcs.go index c768b6b8..58c1d2cf 100644 --- a/hook_defaults_funcs.go +++ b/hook_defaults_funcs.go @@ -27,6 +27,18 @@ func (f HookInsertBeginFunc) IsHook() bool { return true } func (f HookInsertBeginFunc) IsPlugin() bool { return true } +// HookMetricEmitFunc is a convenience helper for implementing +// rivertype.HookMetricEmit using a simple function instead of a struct. +type HookMetricEmitFunc func(ctx context.Context, params *rivertype.HookMetricEmitParams) + +func (f HookMetricEmitFunc) IsHook() bool { return true } + +func (f HookMetricEmitFunc) IsPlugin() bool { return true } + +func (f HookMetricEmitFunc) MetricEmit(ctx context.Context, params *rivertype.HookMetricEmitParams) { + f(ctx, params) +} + // HookPeriodicJobsStartFunc is a convenience helper for implementing // rivertype.HookPeriodicJobsStart using a simple function instead of a struct. type HookPeriodicJobsStartFunc func(ctx context.Context, params *rivertype.HookPeriodicJobsStartParams) error diff --git a/hook_defaults_funcs_test.go b/hook_defaults_funcs_test.go index 52007e2f..c95c246d 100644 --- a/hook_defaults_funcs_test.go +++ b/hook_defaults_funcs_test.go @@ -12,6 +12,10 @@ var ( _ rivertype.HookInsertBegin = HookInsertBeginFunc(func(ctx context.Context, params *rivertype.JobInsertParams) error { return nil }) _ rivertype.Plugin = HookInsertBeginFunc(func(ctx context.Context, params *rivertype.JobInsertParams) error { return nil }) + _ rivertype.Hook = HookMetricEmitFunc(func(ctx context.Context, params *rivertype.HookMetricEmitParams) {}) + _ rivertype.HookMetricEmit = HookMetricEmitFunc(func(ctx context.Context, params *rivertype.HookMetricEmitParams) {}) + _ rivertype.Plugin = HookMetricEmitFunc(func(ctx context.Context, params *rivertype.HookMetricEmitParams) {}) + _ rivertype.Hook = HookPeriodicJobsStartFunc(func(ctx context.Context, params *rivertype.HookPeriodicJobsStartParams) error { return nil }) _ rivertype.HookPeriodicJobsStart = HookPeriodicJobsStartFunc(func(ctx context.Context, params *rivertype.HookPeriodicJobsStartParams) error { return nil }) _ rivertype.Plugin = HookPeriodicJobsStartFunc(func(ctx context.Context, params *rivertype.HookPeriodicJobsStartParams) error { return nil }) diff --git a/internal/pluginlookup/plugin_lookup.go b/internal/pluginlookup/plugin_lookup.go index c2445329..36d83286 100644 --- a/internal/pluginlookup/plugin_lookup.go +++ b/internal/pluginlookup/plugin_lookup.go @@ -15,6 +15,7 @@ type PluginKind string const ( PluginKindHookInsertBegin PluginKind = "hook_insert_begin" + PluginKindHookMetricEmit PluginKind = "hook_metric_emit" PluginKindHookPeriodicJobsStart PluginKind = "hook_periodic_jobs_start" PluginKindHookWorkBegin PluginKind = "hook_work_begin" PluginKindHookWorkEnd PluginKind = "hook_work_end" @@ -85,6 +86,9 @@ func NewPluginLookup(plugins []any) PluginLookupInterface { if _, ok := plugin.(rivertype.HookInsertBegin); ok { pluginsByKind[PluginKindHookInsertBegin] = append(pluginsByKind[PluginKindHookInsertBegin], plugin) } + if _, ok := plugin.(rivertype.HookMetricEmit); ok { + pluginsByKind[PluginKindHookMetricEmit] = append(pluginsByKind[PluginKindHookMetricEmit], plugin) + } if _, ok := plugin.(rivertype.HookPeriodicJobsStart); ok { pluginsByKind[PluginKindHookPeriodicJobsStart] = append(pluginsByKind[PluginKindHookPeriodicJobsStart], plugin) } diff --git a/internal/pluginlookup/plugin_lookup_test.go b/internal/pluginlookup/plugin_lookup_test.go index 2da383c3..3d9484bb 100644 --- a/internal/pluginlookup/plugin_lookup_test.go +++ b/internal/pluginlookup/plugin_lookup_test.go @@ -30,6 +30,7 @@ func TestEmptyPluginLookup(t *testing.T) { pluginLookup, _ := setup(t) require.Nil(t, pluginLookup.ByKind(PluginKindHookInsertBegin)) + require.Nil(t, pluginLookup.ByKind(PluginKindHookMetricEmit)) require.Nil(t, pluginLookup.ByKind(PluginKindHookWorkBegin)) require.Nil(t, pluginLookup.ByKind(PluginKindMiddlewareJobInsert)) require.Nil(t, pluginLookup.ByKind(PluginKindMiddlewareWorker)) @@ -199,6 +200,7 @@ func TestPluginLookup(t *testing.T) { lookup, isPluginLookup := NewPluginLookup([]any{ &testHookInsertAndWorkBegin{}, &testHookInsertBegin{}, + &testHookMetricEmit{}, &testHookWorkBegin{}, &testHookWorkEnd{}, &testMiddlewareJobInsertAndWorker{}, @@ -219,6 +221,9 @@ func TestPluginLookup(t *testing.T) { &testHookInsertAndWorkBegin{}, &testHookInsertBegin{}, }, pluginLookup.ByKind(PluginKindHookInsertBegin)) + require.Equal(t, []any{ + &testHookMetricEmit{}, + }, pluginLookup.ByKind(PluginKindHookMetricEmit)) require.Equal(t, []any{ &testHookInsertAndWorkBegin{}, &testHookWorkBegin{}, @@ -236,13 +241,16 @@ func TestPluginLookup(t *testing.T) { &testMiddlewareWorker{}, }, pluginLookup.ByKind(PluginKindMiddlewareWorker)) - require.Len(t, pluginLookup.pluginsByKind, 5) + require.Len(t, pluginLookup.pluginsByKind, 6) // Repeat lookups to make sure we get the same result. require.Equal(t, []any{ &testHookInsertAndWorkBegin{}, &testHookInsertBegin{}, }, pluginLookup.ByKind(PluginKindHookInsertBegin)) + require.Equal(t, []any{ + &testHookMetricEmit{}, + }, pluginLookup.ByKind(PluginKindHookMetricEmit)) require.Equal(t, []any{ &testHookInsertAndWorkBegin{}, &testHookWorkBegin{}, @@ -276,6 +284,7 @@ func TestPluginLookup(t *testing.T) { } parallelLookupLoop(PluginKindHookInsertBegin) + parallelLookupLoop(PluginKindHookMetricEmit) parallelLookupLoop(PluginKindHookWorkBegin) parallelLookupLoop(PluginKindHookInsertBegin) parallelLookupLoop(PluginKindHookWorkBegin) @@ -387,6 +396,19 @@ func (t *testHookInsertBegin) InsertBegin(ctx context.Context, params *rivertype return nil } +// +// testHookMetricEmit +// + +var _ rivertype.HookMetricEmit = &testHookMetricEmit{} + +type testHookMetricEmit struct{ rivertype.Hook } + +func (t *testHookMetricEmit) IsPlugin() bool { return true } + +func (t *testHookMetricEmit) MetricEmit(ctx context.Context, params *rivertype.HookMetricEmitParams) { +} + // // testHookMiddlewarePlugin // diff --git a/producer.go b/producer.go index b6b0267b..88e30420 100644 --- a/producer.go +++ b/producer.go @@ -191,15 +191,16 @@ type producer struct { // Jobs which are currently being worked. Only used by main goroutine. activeJobs map[int64]*jobexecutor.JobExecutor - completer jobcompleter.JobCompleter - config *producerConfig - id atomic.Int64 // atomic because it's written at startup and read during shutdown - exec riverdriver.Executor - errorHandler jobexecutor.ErrorHandler - fetchLimiter *chanutil.DebouncedChan - state riverpilot.ProducerState - pilot riverpilot.Pilot - workers *Workers + completer jobcompleter.JobCompleter + config *producerConfig + id atomic.Int64 // atomic because it's written at startup and read during shutdown + exec riverdriver.Executor + errorHandler jobexecutor.ErrorHandler + fetchLimiter *chanutil.DebouncedChan + metricEmitHooks []rivertype.HookMetricEmit // memoized hooks of type HookMetricEmit for reuse in dispatchWork + state riverpilot.ProducerState + pilot riverpilot.Pilot + workers *Workers // Receives job IDs to cancel. Written by notifier goroutine, only read from // main goroutine. @@ -243,7 +244,7 @@ func newProducer(archetype *baseservice.Archetype, exec riverdriver.Executor, pi errorHandler = &errorHandlerAdapter{config.ErrorHandler} } - return baseservice.Init(archetype, &producer{ + producer := baseservice.Init(archetype, &producer{ activeJobs: make(map[int64]*jobexecutor.JobExecutor), cancelCh: make(chan int64, 1000), completer: config.Completer, @@ -257,6 +258,10 @@ func newProducer(archetype *baseservice.Archetype, exec riverdriver.Executor, pi retryPolicy: config.RetryPolicy, workers: config.Workers, }) + + producer.metricEmitHooks = producer.metricEmitHooksFromLookup() + + return producer } // Start starts the producer. It backgrounds a goroutine which is stopped when @@ -785,7 +790,33 @@ func (p *producer) maybeCancelJob(ctx context.Context, id int64) { executor.Cancel(ctx) } +func (p *producer) metricEmitHooksFromLookup() []rivertype.HookMetricEmit { + pluginLookup := p.config.PluginLookupGlobal + if pluginLookup == nil { + return nil + } + + plugins := pluginLookup.ByKind(pluginlookup.PluginKindHookMetricEmit) + if len(plugins) < 1 { + return nil + } + + metricEmitHooks := make([]rivertype.HookMetricEmit, len(plugins)) + for i, plugin := range plugins { + metricEmitHooks[i] = plugin.(rivertype.HookMetricEmit) //nolint:forcetypeassert + } + + return metricEmitHooks +} + func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultCh chan<- producerFetchResult) { + // When a queue is paused, innerFetchLoop dispatches with count zero so it can + // continue servicing state changes without attempting to lock jobs or emit metrics. + if count <= 0 { + fetchResultCh <- producerFetchResult{} + return + } + // This intentionally removes any deadlines or cancellation from the parent // context because we don't want it to get cancelled if the producer is asked // to shut down. In that situation, we want to finish fetching any jobs we are @@ -799,6 +830,11 @@ func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultC // rarely hit, but exists to protect against degenerate cases. const maxAttemptedBy = 100 + var startedAt time.Time + if len(p.metricEmitHooks) > 0 { + startedAt = time.Now() + } + jobs, err := p.pilot.JobGetAvailable(ctx, p.exec, p.state, &riverdriver.JobGetAvailableParams{ ClientID: p.config.ClientID, MaxAttemptedBy: maxAttemptedBy, @@ -813,9 +849,30 @@ func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultC return } + if len(p.metricEmitHooks) > 0 { + p.emitMetric(ctx, &rivertype.HookMetricEmitParams{ + Metric: &rivertype.JobGetAvailableDurationMetric{ + Duration: time.Since(startedAt), + Queue: p.config.Queue, + }, + }) + p.emitMetric(ctx, &rivertype.HookMetricEmitParams{ + Metric: &rivertype.JobGetAvailableCountMetric{ + Count: len(jobs), + Queue: p.config.Queue, + }, + }) + } + fetchResultCh <- producerFetchResult{jobs: jobs} } +func (p *producer) emitMetric(ctx context.Context, params *rivertype.HookMetricEmitParams) { + for _, hook := range p.metricEmitHooks { + hook.MetricEmit(ctx, params) + } +} + // Periodically logs an informational log line giving some insight into the // current state of the producer. func (p *producer) heartbeatLogLoop(ctx context.Context, wg *sync.WaitGroup) { diff --git a/producer_test.go b/producer_test.go index 69a2f8bf..e20e6f0a 100644 --- a/producer_test.go +++ b/producer_test.go @@ -33,6 +33,17 @@ import ( const testClientID = "test-client-id" +type countingPluginLookup struct { + pluginlookup.PluginLookupInterface + + count int +} + +func (l *countingPluginLookup) ByKind(kind pluginlookup.PluginKind) []any { + l.count++ + return l.PluginLookupInterface.ByKind(kind) +} + func Test_Producer_CanSafelyCompleteJobsWhileFetchingNewOnes(t *testing.T) { // We have encountered previous data races with the list of active jobs on // Producer because we need to know the count of active jobs in order to @@ -159,6 +170,143 @@ func Test_Producer_CanSafelyCompleteJobsWhileFetchingNewOnes(t *testing.T) { } } +func TestProducer_MetricEmitHook(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + type testBundle struct { + archetype *baseservice.Archetype + config *Config + exec riverdriver.Executor + metrics chan *rivertype.HookMetricEmitParams + pluginLookup *countingPluginLookup + producer *producer + queue string + schema string + } + + setup := func(t *testing.T) *testBundle { + t.Helper() + + var ( + archetype = riversharedtest.BaseServiceArchetype(t) + driver = riverpgxv5.New(riversharedtest.DBPool(ctx, t)) + exec = driver.GetExecutor() + jobUpdates = make(chan []jobcompleter.CompleterJobUpdated, 10) + metrics = make(chan *rivertype.HookMetricEmitParams, 10) + pilot = &riverpilot.StandardPilot{} + queueName = "test_producer_metric_hook" + schema = riverdbtest.TestSchema(ctx, t, driver, nil) + ) + + t.Cleanup(riverinternaltest.DiscardContinuously(jobUpdates)) + + completer := jobcompleter.NewInlineCompleter(archetype, schema, exec, pilot, jobUpdates) + t.Cleanup(completer.Stop) + + metricHook := HookMetricEmitFunc(func(ctx context.Context, params *rivertype.HookMetricEmitParams) { + paramsCopy := *params + metrics <- ¶msCopy + }) + pluginLookup := &countingPluginLookup{ + PluginLookupInterface: pluginlookup.NewPluginLookup([]any{metricHook}), + } + + producer := newProducer(archetype, exec, pilot, &producerConfig{ + ClientID: testClientID, + Completer: completer, + ErrorHandler: newTestErrorHandler(), + FetchCooldown: FetchCooldownDefault, + FetchPollInterval: 50 * time.Millisecond, + JobTimeout: JobTimeoutDefault, + MaxWorkers: 1_000, + PluginLookupByJob: pluginlookup.NewJobPluginLookup(), + PluginLookupGlobal: pluginLookup, + Queue: queueName, + QueuePollInterval: queuePollIntervalDefault, + QueueReportInterval: queueReportIntervalDefault, + RetryPolicy: &DefaultClientRetryPolicy{}, + SchedulerInterval: riverinternaltest.SchedulerShortInterval, + Schema: schema, + StaleProducerRetentionPeriod: time.Minute, + Workers: NewWorkers(), + }) + + return &testBundle{ + archetype: archetype, + config: newTestConfig(t, schema), + exec: exec, + metrics: metrics, + pluginLookup: pluginLookup, + producer: producer, + queue: queueName, + schema: schema, + } + } + + t.Run("EmitsMetricsForFetch", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + + scheduledAt := time.Now().UTC().Add(-time.Second) + insertParams := make([]*riverdriver.JobInsertFastParams, 2) + for i := range insertParams { + params, err := insertParamsFromConfigArgsAndOptions(bundle.archetype, bundle.config, noOpArgs{}, &InsertOpts{ + Queue: bundle.queue, + }) + require.NoError(t, err) + params.ScheduledAt = &scheduledAt + insertParams[i] = (*riverdriver.JobInsertFastParams)(params) + } + + _, err := bundle.exec.JobInsertFastMany(ctx, &riverdriver.JobInsertFastManyParams{ + Jobs: insertParams, + Schema: bundle.schema, + }) + require.NoError(t, err) + + fetchResultCh := make(chan producerFetchResult, 1) + bundle.producer.dispatchWork(ctx, 2, fetchResultCh) + + fetchResult := riversharedtest.WaitOrTimeout(t, fetchResultCh) + require.NoError(t, fetchResult.err) + require.Len(t, fetchResult.jobs, 2) + require.Equal(t, 1, bundle.pluginLookup.count) + + metricsByName := make(map[rivertype.MetricName]rivertype.Metric) + for _, metric := range riversharedtest.WaitOrTimeoutN(t, bundle.metrics, 2) { + metricsByName[metric.Metric.Name()] = metric.Metric + } + + durationMetric, durationMetricFound := metricsByName[rivertype.MetricNameJobGetAvailableDuration].(*rivertype.JobGetAvailableDurationMetric) + require.True(t, durationMetricFound) + require.Equal(t, bundle.queue, durationMetric.Queue) + require.GreaterOrEqual(t, durationMetric.Duration, time.Duration(0)) + + countMetric, countMetricFound := metricsByName[rivertype.MetricNameJobGetAvailableCount].(*rivertype.JobGetAvailableCountMetric) + require.True(t, countMetricFound) + require.Equal(t, bundle.queue, countMetric.Queue) + require.Equal(t, 2, countMetric.Count) + }) + + t.Run("SkipsMetricsWhenNoFetchAttempted", func(t *testing.T) { + t.Parallel() + + bundle := setup(t) + + fetchResultCh := make(chan producerFetchResult, 1) + bundle.producer.dispatchWork(ctx, 0, fetchResultCh) + + fetchResult := riversharedtest.WaitOrTimeout(t, fetchResultCh) + require.NoError(t, fetchResult.err) + require.Empty(t, fetchResult.jobs) + require.Equal(t, 1, bundle.pluginLookup.count) + require.Empty(t, bundle.metrics) + }) +} + func TestProducer_PollOnly(t *testing.T) { t.Parallel() diff --git a/rivertype/river_type.go b/rivertype/river_type.go index ab06a794..475c75ed 100644 --- a/rivertype/river_type.go +++ b/rivertype/river_type.go @@ -235,6 +235,30 @@ func JobStates() []JobState { } } +// MetricName identifies a metric emitted through HookMetricEmit. +type MetricName string + +const ( + // MetricNameJobGetAvailableDuration is the duration of a successful + // JobGetAvailable call. + MetricNameJobGetAvailableDuration MetricName = "job_get_available_duration" + + // MetricNameJobGetAvailableCount is the number of jobs locked by a + // successful JobGetAvailable call. + MetricNameJobGetAvailableCount MetricName = "job_get_available_count" +) + +// Metric is a strongly typed metric payload emitted through HookMetricEmit. +// +// River provides all Metric implementations. New metric types may be added in +// future versions without changing HookMetricEmit's method signature. +type Metric interface { + // Name identifies the emitted metric. + Name() MetricName + + isMetric() +} + // AttemptError is an error from a single job attempt that failed due to an // error or a panic. type AttemptError struct { @@ -303,6 +327,8 @@ type JobInsertParams struct { // // List of hook interfaces that may be implemented: // - HookInsertBegin +// - HookMetricEmit +// - HookPeriodicJobsStart // - HookWorkBegin // - HookWorkEnd // @@ -323,6 +349,56 @@ type HookInsertBegin interface { InsertBegin(ctx context.Context, params *JobInsertParams) error } +// HookMetricEmit is an interface to a hook that receives metrics emitted by +// River. +type HookMetricEmit interface { + Hook + + // MetricEmit is invoked each time River emits a metric. It should not block + // on network I/O, and should usually pass metrics through to an asynchronous + // instrumentation package like OpenTelemetry. + MetricEmit(ctx context.Context, params *HookMetricEmitParams) +} + +// HookMetricEmitParams are parameters for HookMetricEmit. +type HookMetricEmitParams struct { + // Metric is the emitted metric payload. Use a type switch to access + // metric-specific fields. + Metric Metric +} + +// JobGetAvailableDurationMetric is emitted after a successful JobGetAvailable +// call with the call's duration. +type JobGetAvailableDurationMetric struct { + // Duration is how long the JobGetAvailable call took. + Duration time.Duration + + // Queue is the queue that jobs were locked from. + Queue string +} + +func (m *JobGetAvailableDurationMetric) Name() MetricName { + return MetricNameJobGetAvailableDuration +} + +func (m *JobGetAvailableDurationMetric) isMetric() {} + +// JobGetAvailableCountMetric is emitted after a successful JobGetAvailable +// call with the number of jobs locked. +type JobGetAvailableCountMetric struct { + // Count is the number of jobs locked. + Count int + + // Queue is the queue that jobs were locked from. + Queue string +} + +func (m *JobGetAvailableCountMetric) Name() MetricName { + return MetricNameJobGetAvailableCount +} + +func (m *JobGetAvailableCountMetric) isMetric() {} + // HookPeriodicJobsStart is an interface to a hook that runs when the periodic // job enqueuer starts on a newly elected leader. type HookPeriodicJobsStart interface {