Support job-specific log levels

Changes.md

@@ -5,6 +5,11 @@
 HEAD
 ---------
 
+- Support job-specific log levels. This will allow you to turn on debugging for
+  problematic workers. [fatkodima, #4287]
+```ruby
+MyWorker.set(log_level: :debug).perform_async(...)
+```
 - Support ad-hoc job tags. You can tag your jobs with, e.g, subdomain, tenant, country,
   locale, application, version, user/client, "alpha/beta/pro/ent", types of jobs,
   teams/people responsible for jobs, additional metadata, etc.

lib/sidekiq.rb

@@ -205,9 +205,8 @@ module Sidekiq
       return self.logger
     end
 
-    unless logger.is_a?(Sidekiq::LogContext)
-      logger.extend(Sidekiq::LogContext)
-    end
+    logger.extend(Sidekiq::LogContext)
+    logger.extend(Sidekiq::LoggerThreadSafeLevel)
 
     @logger = logger
   end

lib/sidekiq/job_logger.rb

@@ -23,8 +23,15 @@ module Sidekiq
       raise
     end
 
-    def with_job_hash_context(job_hash, &block)
-      @logger.with_context(job_hash_context(job_hash), &block)
+    def with_job_hash_context_and_log_level(job_hash, &block)
+      level = job_hash["log_level"]
+      if level
+        @logger.log_at(level) do
+          @logger.with_context(job_hash_context(job_hash), &block)
+        end
+      else
+        @logger.with_context(job_hash_context(job_hash), &block)
+      end
     end
 
     def job_hash_context(job_hash)

lib/sidekiq/logger.rb

@@ -17,8 +17,74 @@ module Sidekiq
     end
   end
 
+  module LoggerThreadSafeLevel
+    LOG_LEVEL_MAP = Hash[Logger::Severity.constants.map { |c| [c.to_s, Logger::Severity.const_get(c)] }]
+    LOG_LEVEL_MAP.default_proc = proc do |_, level|
+      Sidekiq.logger.info("Invalid log level: #{level.inspect}")
+      nil
+    end
+
+    Logger::Severity.constants.each do |severity|
+      define_method("#{severity.downcase}?") do
+        Logger.const_get(severity) >= level
+      end
+    end
+
+    def local_level
+      Thread.current[:sidekiq_log_level]
+    end
+
+    def local_level=(level)
+      case level
+      when Integer
+        Thread.current[:sidekiq_log_level] = level
+      when Symbol, String
+        Thread.current[:sidekiq_log_level] = LOG_LEVEL_MAP[level.to_s.upcase]
+      when nil
+        Thread.current[:sidekiq_log_level] = nil
+      else
+        raise ArgumentError, "Invalid log level: #{level.inspect}"
+      end
+    end
+
+    def level
+      local_level || super
+    end
+
+    # Change the thread-local level for the duration of the given block.
+    def log_at(level)
+      old_local_level = local_level
+      self.local_level = level
+      yield
+    ensure
+      self.local_level = old_local_level
+    end
+
+    # Redefined to check severity against #level, and thus the thread-local level, rather than +@level+.
+    # FIXME: Remove when the minimum Ruby version supports overriding Logger#level.
+    def add(severity, message = nil, progname = nil, &block)
+      severity ||= UNKNOWN
+      progname ||= @progname
+
+      return true if @logdev.nil? || severity < level
+
+      if message.nil?
+        if block_given?
+          message = yield
+        else
+          message = progname
+          progname = @progname
+        end
+      end
+
+      @logdev.write \
+        format_message(format_severity(severity), Time.now, progname, message)
+    end
+  end
+
   class Logger < ::Logger
     include LogContext
+    include LoggerThreadSafeLevel
 
     def initialize(*args)
       super

lib/sidekiq/processor.rb

@@ -117,7 +117,7 @@ module Sidekiq
       # job structure to the Web UI
       pristine = cloned(job_hash)
 
-      @job_logger.with_job_hash_context(job_hash) do
+      @job_logger.with_job_hash_context_and_log_level(job_hash) do
         @retrier.global(pristine, queue) do
           @job_logger.call(job_hash, queue) do
             stats(pristine, queue) do

test/test_job_logger.rb

@@ -7,7 +7,8 @@ class TestJobLogger < Minitest::Test
   def setup
     @old = Sidekiq.logger
     @output = StringIO.new
-    @logger = Sidekiq::Logger.new(@output)
+    @logger = Sidekiq::Logger.new(@output, level: :info)
+    Sidekiq.logger = @logger
 
     Thread.current[:sidekiq_context] = nil
     Thread.current[:sidekiq_tid] = nil
@@ -16,9 +17,9 @@ class TestJobLogger < Minitest::Test
   def teardown
     Thread.current[:sidekiq_context] = nil
     Thread.current[:sidekiq_tid] = nil
+    Sidekiq.logger = @old
   end
 
-
   def test_pretty_output
     jl = Sidekiq::JobLogger.new(@logger)
 
@@ -26,7 +27,7 @@ class TestJobLogger < Minitest::Test
     p = @logger.formatter = Sidekiq::Logger::Formatters::Pretty.new
     job = {"jid"=>"1234abc", "wrapped"=>"FooWorker", "class"=>"Wrapper", "tags" => ["bar", "baz"]}
     # this mocks what Processor does
-    jl.with_job_hash_context(job) do
+    jl.with_job_hash_context_and_log_level(job) do
       jl.call(job, 'queue') {}
     end
 
@@ -46,7 +47,7 @@ class TestJobLogger < Minitest::Test
     jl = Sidekiq::JobLogger.new(@logger)
     job = {"jid"=>"1234abc", "wrapped"=>"Wrapper", "class"=>"FooWorker", "bid"=>"b-xyz", "tags" => ["bar", "baz"]}
     # this mocks what Processor does
-    jl.with_job_hash_context(job) do
+    jl.with_job_hash_context_and_log_level(job) do
       jl.call(job, 'queue') {}
     end
     a, b = @output.string.lines
@@ -59,6 +60,40 @@ class TestJobLogger < Minitest::Test
     assert_equal(["bid", "class", "jid", "tags"], keys)
   end
 
+  def test_custom_log_level
+    jl = Sidekiq::JobLogger.new(@logger)
+    job = {"class"=>"FooWorker", "log_level"=>"debug"}
+
+    assert @logger.info?
+    jl.with_job_hash_context_and_log_level(job) do
+      jl.call(job, "queue") do
+        assert @logger.debug?
+        @logger.debug("debug message")
+      end
+    end
+    assert @logger.info?
+
+    a, b, c = @output.string.lines
+    assert_match(/INFO: start/, a)
+    assert_match(/DEBUG: debug message/, b)
+    assert_match(/INFO: done/, c)
+  end
+
+  def test_custom_log_level_uses_default_log_level_for_invalid_value
+    jl = Sidekiq::JobLogger.new(@logger)
+    job = {"class"=>"FooWorker", "log_level"=>"non_existent"}
+
+    assert @logger.info?
+    jl.with_job_hash_context_and_log_level(job) do
+      jl.call(job, "queue") do
+        assert @logger.info?
+      end
+    end
+    assert @logger.info?
+    log_level_warning = @output.string.lines[0]
+    assert_match(/INFO: Invalid log level/, log_level_warning)
+  end
+
   def reset(io)
     io.truncate(0)
     io.rewind