Merge pull request reidmorrison#224 from reidmorrison/sidekiq-refinements

Refine Sidekiq metrics that are generated

Author: reidmorrison

.gitignore

@@ -16,3 +16,4 @@ Gemfile.lock
 TODO.md
 
 .tool-versions
+test/dummy/db/test.sqlite3

README.md

@@ -37,26 +37,30 @@ make_col event:parse_json(log)
 
 make_col
    time:parse_isotime(event.timestamp),
+   application:string(event.application),
+   environment:string(event.environment),
    duration:duration_ms(event.duration_ms),
    level:string(event.level),
    name:string(event.name),
    message:string(event.message),
-   metric:string(event.metric),
    named_tags:event.named_tags,
    payload:event.payload,
+   metric:string(event.metric),
+   metric_amount:float64(event.metric_amount),
    tags:array(event.tags),
+   exception:event.exception,
    host:string(event.host),
-   application:string(event.application),
-   environment:string(event.environment),
-   level_index:int64(event.level_index),
    pid:int64(event.pid),
    thread:string(event.thread),
    file:string(event.file),
    line:int64(event.line),
-   backtrace:array(event.backtrace),
-   metric_amount:int64(event.metric_amount),
    dimensions:event.dimensions,
-   exception:event.exception
+   backtrace:array(event.backtrace),
+   level_index:int64(event.level_index)
+
+set_valid_from(time)
+drop_col timestamp, log, event, stream
+rename_col timestamp:time
 ~~~
 
 Now queries can be built to drill down into each of these fields, including `payload` which is a nested object.
@@ -65,15 +69,24 @@ For example to find all failed Sidekiq job calls where the causing exception cla
 ~~~ruby
 filter environment = "uat2"
 filter level = "error"
-filter metric ~ /Sidekiq/
+filter metric = "sidekiq.job.perform"
 filter (string(exception.cause.name) = "NoMethodError")
 ~~~
 
-Or, to create a dashboard showing the duration of all successful Sidekiq jobs:
+Example: create a dashboard showing the duration of all successful Sidekiq jobs:
 ~~~ruby
 filter environment = "production"
 filter level = "info"
-filter metric ~ /Sidekiq.*/
+filter metric = "sidekiq.job.perform"
+timechart duration:avg(duration), group_by(name)
+~~~
+
+Example: create a dashboard showing the queue latency of all Sidekiq jobs. 
+The queue latency is the time between when the job was enqueued and when it was started:
+~~~ruby
+filter environment = "production"
+filter level = "info"
+filter metric = "sidekiq.queue.latency"
 timechart duration:avg(duration), group_by(name)
 ~~~
 
@@ -83,7 +96,14 @@ timechart duration:avg(duration), group_by(name)
 
 For complete documentation see: https://logger.rocketjob.io/rails
 
-## Upgrading to Semantic Logger v4.15 - Sidekiq Support
+## Upgrading to Semantic Logger V4.16 - Sidekiq Metrics Support
+
+Rails Semantic Logger now supports Sidekiq metrics. 
+Below are the metrics that are now available when the JSON logging format is used:
+- `sidekiq.job.perform` - The duration of each Sidekiq job.
+- `sidekiq.queue.latency` - The time between when a Sidekiq job was enqueued and when it was started.
+
+## Upgrading to Semantic Logger v4.15 & V4.16 - Sidekiq Support
 
 Rails Semantic Logger introduces direct support for Sidekiq v4, v5, v6, and v7. 
 Please remove any previous custom patches or configurations to make Sidekiq work with Semantic Logger.

lib/rails_semantic_logger.rb

@@ -69,7 +69,7 @@ def self.subscriber_patterns(subscriber)
 require("rails_semantic_logger/extensions/active_support/log_subscriber") if defined?(ActiveSupport::LogSubscriber)
 
 begin
-  require 'rackup'
+  require "rackup"
 rescue LoadError
   # No need to do anything, will fall back to Rack
 end

lib/rails_semantic_logger/extensions/sidekiq/sidekiq.rb

@@ -33,18 +33,26 @@ module Sidekiq
   if defined?(::Sidekiq::JobLogger)
     # Let Semantic Logger handle duration logging
     class JobLogger
-      def call(item, queue)
+      def call(item, queue, &block)
         klass  = item["wrapped"] || item["class"]
-        metric = "Sidekiq/#{klass}/perform" if klass
         logger = klass ? SemanticLogger[klass] : Sidekiq.logger
-        logger.info("Start #perform")
-        logger.measure_info(
-          "Completed #perform",
-          on_exception_level: :error,
-          log_exception:      :full,
-          metric:             metric
-        ) do
-          yield
+
+        SemanticLogger.tagged(queue: queue) do
+          # Latency is the time between when the job was enqueued and when it started executing.
+          logger.info(
+            "Start #perform",
+            metric:        "sidekiq.queue.latency",
+            metric_amount: job_latency_ms(item)
+          )
+
+          # Measure the duration of running the job
+          logger.measure_info(
+            "Completed #perform",
+            on_exception_level: :error,
+            log_exception:      :full,
+            metric:             "sidekiq.job.perform",
+            &block
+          )
         end
       end
 
@@ -60,14 +68,18 @@ def prepare(job_hash, &block)
       end
 
       def job_hash_context(job_hash)
-        h        = {
-          class: job_hash["display_class"] || job_hash["wrapped"] || job_hash["class"],
-          jid:   job_hash["jid"]
-        }
-        h[:bid]  = job_hash["bid"] if job_hash["bid"]
-        h[:tags] = job_hash["tags"] if job_hash["tags"]
+        h         = {jid: job_hash["jid"]}
+        h[:bid]   = job_hash["bid"] if job_hash["bid"]
+        h[:tags]  = job_hash["tags"] if job_hash["tags"]
+        h[:queue] = job_hash["queue"] if job_hash["queue"]
         h
       end
+
+      def job_latency_ms(job)
+        return unless job && job["enqueued_at"]
+
+        (Time.now.to_f - job["enqueued_at"].to_f) * 1000
+      end
     end
   end
 
@@ -80,48 +92,47 @@ def self.with_context(msg, &block)
       end
 
       def self.job_hash_context(job_hash)
-        klass       = job_hash["wrapped"] || job_hash["class"]
-        event       = { class: klass, jid: job_hash["jid"] }
-        event[:bid] = job_hash["bid"] if job_hash["bid"]
-        event
+        h         = {jid: job_hash["jid"]}
+        h[:bid]   = job_hash["bid"] if job_hash["bid"]
+        h[:queue] = job_hash["queue"] if job_hash["queue"]
+        h
       end
     end
   end
 
   # Exception is already logged by Semantic Logger during the perform call
-  # Sidekiq <= v6.5
   if defined?(::Sidekiq::ExceptionHandler)
+    # Sidekiq <= v6.5
     module ExceptionHandler
       class Logger
-        def call(ex, ctx)
-          unless ctx.empty?
-            job_hash = ctx[:job] || {}
-            klass = job_hash["display_class"] || job_hash["wrapped"] || job_hash["class"]
-            logger = klass ? SemanticLogger[klass] : Sidekiq.logger
-            ctx[:context] ? logger.warn(ctx[:context], ctx) : logger.warn(ctx)
-          end
+        def call(_exception, ctx)
+          return if ctx.empty?
+
+          job_hash = ctx[:job] || {}
+          klass    = job_hash["display_class"] || job_hash["wrapped"] || job_hash["class"]
+          logger   = klass ? SemanticLogger[klass] : Sidekiq.logger
+          ctx[:context] ? logger.warn(ctx[:context], ctx) : logger.warn(ctx)
         end
       end
     end
-  # Sidekiq >= v7
   elsif defined?(::Sidekiq::Config)
+    # Sidekiq >= v7
     class Config
       remove_const :ERROR_HANDLER
 
-      ERROR_HANDLER = ->(ex, ctx, cfg = Sidekiq.default_configuration) {
+      ERROR_HANDLER = ->(ex, ctx, cfg = Sidekiq.default_configuration) do
         unless ctx.empty?
           job_hash = ctx[:job] || {}
-          klass = job_hash["display_class"] || job_hash["wrapped"] || job_hash["class"]
-          logger = klass ? SemanticLogger[klass] : Sidekiq.logger
+          klass    = job_hash["display_class"] || job_hash["wrapped"] || job_hash["class"]
+          logger   = klass ? SemanticLogger[klass] : Sidekiq.logger
           ctx[:context] ? logger.warn(ctx[:context], ctx) : logger.warn(ctx)
         end
-      }
+      end
     end
   else
     # Sidekiq >= 6.5
-    # TODO: Not taking effect. See test/sidekiq_test.rb
-    def self.default_error_handler(ex, ctx)
-      binding.irb
+    Sidekiq.error_handlers.delete(Sidekiq::DEFAULT_ERROR_HANDLER)
+    Sidekiq.error_handlers << ->(ex, ctx) do
       unless ctx.empty?
         job_hash = ctx[:job] || {}
         klass = job_hash["display_class"] || job_hash["wrapped"] || job_hash["class"]
@@ -132,10 +143,13 @@ def self.default_error_handler(ex, ctx)
   end
 
   # Logging within each worker should use its own logger
-  if Sidekiq::VERSION.to_i == 4
+  case Sidekiq::VERSION.to_i
+  when 4
     module Worker
       def self.included(base)
-        raise ArgumentError, "You cannot include Sidekiq::Worker in an ActiveJob: #{base.name}" if base.ancestors.any? { |c| c.name == "ActiveJob::Base" }
+        if base.ancestors.any? { |c| c.name == "ActiveJob::Base" }
+          raise ArgumentError, "You cannot include Sidekiq::Worker in an ActiveJob: #{base.name}"
+        end
 
         base.extend(ClassMethods)
         base.include(SemanticLogger::Loggable)
@@ -144,10 +158,12 @@ def self.included(base)
         base.class_attribute :sidekiq_retries_exhausted_block
       end
     end
-  elsif Sidekiq::VERSION.to_i == 5
+  when 5
     module Worker
       def self.included(base)
-        raise ArgumentError, "You cannot include Sidekiq::Worker in an ActiveJob: #{base.name}" if base.ancestors.any? { |c| c.name == "ActiveJob::Base" }
+        if base.ancestors.any? { |c| c.name == "ActiveJob::Base" }
+          raise ArgumentError, "You cannot include Sidekiq::Worker in an ActiveJob: #{base.name}"
+        end
 
         base.extend(ClassMethods)
         base.include(SemanticLogger::Loggable)
@@ -156,10 +172,12 @@ def self.included(base)
         base.sidekiq_class_attribute :sidekiq_retries_exhausted_block
       end
     end
-  elsif Sidekiq::VERSION.to_i == 6
+  when 6
     module Worker
       def self.included(base)
-        raise ArgumentError, "Sidekiq::Worker cannot be included in an ActiveJob: #{base.name}" if base.ancestors.any? { |c| c.name == "ActiveJob::Base" }
+        if base.ancestors.any? { |c| c.name == "ActiveJob::Base" }
+          raise ArgumentError, "Sidekiq::Worker cannot be included in an ActiveJob: #{base.name}"
+        end
 
         base.include(Options)
         base.extend(ClassMethods)
@@ -169,7 +187,9 @@ def self.included(base)
   else
     module Job
       def self.included(base)
-        raise ArgumentError, "Sidekiq::Job cannot be included in an ActiveJob: #{base.name}" if base.ancestors.any? { |c| c.name == "ActiveJob::Base" }
+        if base.ancestors.any? { |c| c.name == "ActiveJob::Base" }
+          raise ArgumentError, "Sidekiq::Job cannot be included in an ActiveJob: #{base.name}"
+        end
 
         base.include(Options)
         base.extend(ClassMethods)
@@ -178,14 +198,15 @@ def self.included(base)
     end
   end
 
-  if Sidekiq::VERSION.to_i == 4
+  if defined?(::Sidekiq::Middleware::Server::Logging)
+    # Sidekiq v4
     # Convert string to machine readable format
     class Processor
       def log_context(job_hash)
-        klass       = job_hash["wrapped"] || job_hash["class"]
-        event       = { class: klass, jid: job_hash["jid"] }
-        event[:bid] = job_hash["bid"] if job_hash["bid"]
-        event
+        h         = {jid: job_hash["jid"]}
+        h[:bid]   = job_hash["bid"] if job_hash["bid"]
+        h[:queue] = job_hash["queue"] if job_hash["queue"]
+        h
       end
     end
 
@@ -194,16 +215,26 @@ module Middleware
       module Server
         class Logging
           def call(worker, item, queue)
-            worker.logger.info("Start #perform")
-            worker.logger.measure_info(
-              "Completed #perform",
-              on_exception_level: :error,
-              log_exception:      :full,
-              metric:             "Sidekiq/#{worker.class.name}/perform"
-            ) do
-              yield
+            SemanticLogger.tagged(queue: queue) do
+              worker.logger.info(
+                "Start #perform",
+                metric:        "sidekiq.queue.latency",
+                metric_amount: job_latency_ms(item)
+              )
+              worker.logger.measure_info(
+                "Completed #perform",
+                on_exception_level: :error,
+                log_exception:      :full,
+                metric:             "sidekiq.job.perform"
+              ) { yield }
             end
           end
+
+          def job_latency_ms(job)
+            return unless job && job["enqueued_at"]
+
+            (Time.now.to_f - job["enqueued_at"].to_f) * 1000
+          end
         end
       end
     end

lib/rails_semantic_logger/version.rb

@@ -1,3 +1,3 @@
 module RailsSemanticLogger
-  VERSION = "4.15.0".freeze
+  VERSION = "4.16.0".freeze
 end

test/dummy/config/application.rb

@@ -6,9 +6,7 @@
 
 module Dummy
   class Application < Rails::Application
-    if config.respond_to?(:load_defaults)
-      config.load_defaults "#{Rails::VERSION::MAJOR}.#{Rails::VERSION::MINOR}"
-    end
+    config.load_defaults "#{Rails::VERSION::MAJOR}.#{Rails::VERSION::MINOR}" if config.respond_to?(:load_defaults)
 
     # Configure sensitive parameters which will be filtered from the log file.
     config.filter_parameters += [:password]