Job metrics and subscribers

Author: stephenbinns

docs/README.md

@@ -798,6 +798,32 @@ Que.job_middleware.push(
 )
 ```
 
+#### Existing Middleware
+
+Que ships with middleware to expose job metrics using ActiveSupport notifications to subscribe to it you can implelent the following
+
+```ruby
+::ActiveSupport::Notifications.subscribe("que_job.worked") do |message, started, finished, labels|
+  # do something with notification.
+end
+```
+
+`started` and `finished` are numeric values representing a monotonic clock so can 
+be used for timing calculations without concerning ourselves with the system clock.
+
+`labels` is a hash containing the following keys
+
+* `job_class` - the class of the job.
+* `queue` - the queue this job was queued into.
+* `priority` - the priority of this job.
+* `latency` - the amount of time this job was waiting in the queue for.
+
+To use this middleware you will have to initialize it with Que
+
+```ruby
+Que.job_middleware.push(Que::ActiveSupport::JobMiddleware)
+```
+
 ### Defining Middleware For SQL statements
 
 SQL middleware wraps queries that Que executes, or which you might decide to execute via Que.execute(). You can use hook this into NewRelic or a similar service to instrument how long SQL queries take, for example.

lib/que/active_support/job_middleware.rb

@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+module Que
+  module ActiveSupport
+    module JobMiddleware
+      def self.call(job)
+        labels = {
+          job_class: job.que_attrs[:job_class],
+          priority: job.que_attrs[:priority],
+          queue: job.que_attrs[:queue],
+          latency: job.que_attrs[:latency],
+        }
+
+        started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        yield
+      ensure
+        ::ActiveSupport::Notifications.publish(
+          "que_job.worked",
+          started,
+          Process.clock_gettime(Process::CLOCK_MONOTONIC),
+          labels.merge(error: !!job.que_error.present?),
+        )
+      end
+    end
+  end
+end

spec/que/active_support/job_middleware_spec.rb

@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+if defined?(::ActiveSupport)
+  require 'que/active_support/job_middleware'
+
+  describe Que::ActiveSupport::JobMiddleware do
+    let(:job) do
+      Que::Job.new(
+        job_class: "Foo",
+        priority: 100,
+        queue: "foo_queue",
+        latency: 100,
+      )
+    end
+
+    let(:labels) do
+      {
+        job_class: "Foo",
+        priority: 100,
+        queue: "foo_queue",
+        latency: 100,
+      }
+    end
+
+    it "records metrics" do
+      called = false
+      ::ActiveSupport::Notifications.subscribe("que_job.worked") do |message, started, finished, metric_labels|
+        assert_equal "que_job.worked", message
+        assert started != nil
+        assert finished != nil
+        assert_equal labels.merge(error: false), metric_labels
+        called = true
+      end
+
+      Que::ActiveSupport::JobMiddleware.call(job) { }
+
+      assert_equal true, called
+    end
+  end
+end

spec/spec_helper.rb

@@ -8,6 +8,7 @@
 # in some spec runs.
 if ENV['USE_RAILS'] == 'true'
   require 'active_record'
+  require 'active_support'
   require 'active_job'
 
   ActiveJob::Base.queue_adapter = :que