FEATURE: Adding Shoryuken metrics (#104)

* Adding Shoryuken metrics

* fixed build check

* added namespace resolution operator for shoryuken shutdown

* adding unit tests for shoryuken metrics

* changed false case in shoryuken unit test

* included custom lables

* adding  Errno::ECONNRESET

README.md

@@ -13,6 +13,7 @@ To learn more see [Instrumenting Rails with Prometheus](https://samsaffron.com/a
   * [Rails integration](#rails-integration)
     * [Per-process stats](#per-process-stats)
     * [Sidekiq metrics](#sidekiq-metrics)
+    * [Shoryuken metrics](#shoryuken-metrics)
     * [ActiveRecord Connection Pool Metrics](#activerecord-connection-pool-metrics)
     * [Delayed Job plugin](#delayed-job-plugin)
     * [Hutch metrics](#hutch-message-processing-tracer)
@@ -297,6 +298,19 @@ Sometimes the Sidekiq server shuts down before it can send metrics, that were ge
   end
 ```
 
+#### Shoryuken metrics
+
+For Shoryuken metrics (how many jobs ran? how many failed? how long did they take? how many were restarted?)
+
+```ruby
+Shoryuken.configure_server do |config|
+  config.server_middleware do |chain|
+    require 'prometheus_exporter/instrumentation'
+    chain.add PrometheusExporter::Instrumentation::Shoryuken
+  end
+end
+```
+
 #### Delayed Job plugin
 
 In an initializer:

lib/prometheus_exporter/instrumentation.rb

@@ -9,3 +9,4 @@
 require_relative "instrumentation/hutch"
 require_relative "instrumentation/unicorn"
 require_relative "instrumentation/active_record"
+require_relative "instrumentation/shoryuken"

lib/prometheus_exporter/instrumentation/shoryuken.rb

@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+module PrometheusExporter::Instrumentation
+  class Shoryuken
+
+    def initialize(client: nil)
+      @client = client || PrometheusExporter::Client.default
+    end
+
+    def call(worker, queue, msg, body)
+      success = false
+      start = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
+      result = yield
+      success = true
+      result
+    rescue ::Shoryuken::Shutdown => e
+      shutdown = true
+      raise e
+    ensure
+      duration = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) - start
+      @client.send_json(
+          type: "shoryuken",
+          queue: queue,
+          name: worker.class.name,
+          success: success,
+          shutdown: shutdown,
+          duration: duration
+      )
+    end
+  end
+end

lib/prometheus_exporter/server.rb

@@ -14,3 +14,4 @@
 require_relative "server/hutch_collector"
 require_relative "server/unicorn_collector"
 require_relative "server/active_record_collector"
+require_relative "server/shoryuken_collector"

lib/prometheus_exporter/server/collector.rb

@@ -18,6 +18,7 @@ def initialize(json_serializer: nil)
       register_collector(HutchCollector.new)
       register_collector(UnicornCollector.new)
       register_collector(ActiveRecordCollector.new)
+      register_collector(ShoryukenCollector.new)
     end
 
     def register_collector(collector)

lib/prometheus_exporter/server/shoryuken_collector.rb

@@ -0,0 +1,59 @@
+# frozen_string_literal: true
+
+module PrometheusExporter::Server
+  class ShoryukenCollector < TypeCollector
+
+    def type
+      "shoryuken"
+    end
+
+    def collect(obj)
+      default_labels = { job_name: obj['name'] , queue_name: obj['queue'] }
+      custom_labels = obj['custom_labels']
+      labels = custom_labels.nil? ? default_labels : default_labels.merge(custom_labels)
+
+      ensure_shoryuken_metrics
+      @shoryuken_job_duration_seconds.observe(obj["duration"], labels)
+      @shoryuken_jobs_total.observe(1, labels)
+      @shoryuken_restarted_jobs_total.observe(1, labels) if obj["shutdown"]
+      @shoryuken_failed_jobs_total.observe(1, labels) if !obj["success"] && !obj["shutdown"]
+    end
+
+    def metrics
+      if @shoryuken_jobs_total
+        [
+            @shoryuken_job_duration_seconds,
+            @shoryuken_jobs_total,
+            @shoryuken_restarted_jobs_total,
+            @shoryuken_failed_jobs_total,
+        ]
+      else
+        []
+      end
+    end
+
+    protected
+
+    def ensure_shoryuken_metrics
+      if !@shoryuken_jobs_total
+
+        @shoryuken_job_duration_seconds =
+            PrometheusExporter::Metric::Counter.new(
+                "shoryuken_job_duration_seconds", "Total time spent in shoryuken jobs.")
+
+        @shoryuken_jobs_total =
+            PrometheusExporter::Metric::Counter.new(
+                "shoryuken_jobs_total", "Total number of shoryuken jobs executed.")
+
+        @shoryuken_restarted_jobs_total =
+            PrometheusExporter::Metric::Counter.new(
+                "shoryuken_restarted_jobs_total", "Total number of shoryuken jobs that we restarted because of a shoryuken shutdown.")
+
+        @shoryuken_failed_jobs_total =
+            PrometheusExporter::Metric::Counter.new(
+                "shoryuken_failed_jobs_total", "Total number of failed shoryuken jobs.")
+
+      end
+    end
+  end
+end

test/server/collector_test.rb

@@ -294,6 +294,28 @@ def test_it_can_collect_sidekiq_metrics_with_custom_labels
     assert(result.include?('sidekiq_jobs_total{job_name="WrappedClass",service="service1"} 1'), "has sidekiq working job from ActiveJob")
   end
 
+  def test_it_can_collect_shoryuken_metrics_with_custom_lables
+    collector = PrometheusExporter::Server::Collector.new
+    client = PipedClient.new(collector, custom_labels: { service: 'service1' })
+
+    instrument = PrometheusExporter::Instrumentation::Shoryuken.new(client: client)
+
+    instrument.call("hello", nil, "default", "body") do
+    end
+    begin
+      instrument.call(false, nil, "default", "body") do
+        boom
+      end
+    rescue
+    end
+
+    result = collector.prometheus_metrics_text
+
+    assert(result.include?("shoryuken_failed_jobs_total{job_name=\"FalseClass\",queue_name=\"\",service=\"service1\"} 1"), "has failed job")
+    assert(result.include?("shoryuken_jobs_total{job_name=\"String\",queue_name=\"\",service=\"service1\"} 1"), "has working job")
+    assert(result.include?("shoryuken_job_duration_seconds{job_name=\"String\",queue_name=\"\",service=\"service1\"} "), "has duration")
+  end
+
   def test_it_merges_custom_labels_for_generic_metrics
     name = 'test_name'
     help = 'test_help'

test/server/web_server_test.rb

@@ -36,7 +36,7 @@ def find_free_port
       begin
         TCPSocket.new("localhost", port).close
         port += 1
-      rescue Errno::ECONNREFUSED
+      rescue Errno::ECONNREFUSED, Errno::ECONNRESET
         break
       end
     end