Log each Sidekiq line once, with the job's ID, and write logged Hashes at the top level

docs/lib/integration-helpers.ts

@@ -81,7 +81,7 @@ export function getLogTypeInfo(logType: LogType): {
       return {
         title: 'Sidekiq',
         description:
-          'The Sidekiq integration configures structured JSON logging for Sidekiq worker and client logs, maintaining consistent format with other logs.',
+          'The Sidekiq integration configures structured JSON logging for Sidekiq worker and client logs, maintaining consistent format with other logs. Every line logged while a job runs includes the job ID and class in ctx.jid and ctx.class.',
       };
 
     case LogType.GOODJOB:

lib/log_struct/integrations/sidekiq.rb

@@ -6,7 +6,10 @@
 rescue LoadError
   # Sidekiq gem is not available, integration will be skipped
 end
-require_relative "sidekiq/logger" if defined?(::Sidekiq)
+if defined?(::Sidekiq)
+  require_relative "sidekiq/logger"
+  require_relative "sidekiq/job_context_middleware"
+end
 
 module LogStruct
   module Integrations
@@ -25,6 +28,13 @@ def self.setup(config)
         # Configure Sidekiq server (worker) to use our logger
         ::Sidekiq.configure_server do |sidekiq_config|
           sidekiq_config.logger = LogStruct::Integrations::Sidekiq::Logger.new("Sidekiq-Server")
+
+          # First in the chain, so lines logged by other middleware include the job too
+          sidekiq_config.server_middleware do |chain|
+            chain.prepend LogStruct::Integrations::Sidekiq::JobContextMiddleware
+          end
+
+          replace_default_error_handler(sidekiq_config)
         end
 
         # Configure Sidekiq client (Rails app) to use our logger
@@ -34,6 +44,55 @@ def self.setup(config)
 
         true
       end
+
+      # Sidekiq's default error handler logs three lines for each failed job: the job (with its
+      # arguments) as a JSON string, the error, and the backtrace. Log one structured line instead,
+      # without the arguments, which LogStruct can't filter inside a JSON string.
+      sig { params(sidekiq_config: T.untyped).void }
+      def self.replace_default_error_handler(sidekiq_config)
+        return unless defined?(::Sidekiq::Config::ERROR_HANDLER)
+
+        handlers = sidekiq_config.error_handlers
+        return unless handlers.delete(::Sidekiq::Config::ERROR_HANDLER)
+
+        handlers.unshift(JOB_ERROR_HANDLER)
+      end
+
+      sig { params(error: Exception, context: T::Hash[T.untyped, T.untyped]).void }
+      def self.log_job_error(error, context)
+        job = context[:job]
+        job = nil unless job.is_a?(Hash)
+
+        # LogStruct's ActiveJob integration already logs the errors of ActiveJob jobs
+        return if job&.key?("wrapped") && LogStruct.config.integrations.enable_activejob
+
+        details = {
+          context: context[:context],
+          jid: job&.dig("jid"),
+          job_class: job && (job["wrapped"] || job["class"]),
+          queue: job&.dig("queue"),
+          retry_count: job&.dig("retry_count")
+        }.compact
+
+        if error.is_a?(StandardError)
+          ::Rails.logger.warn(Log.from_exception(Source::Sidekiq, error, details))
+        else
+          # Log::Error only takes a StandardError; Sidekiq also reports others (e.g. NoMemoryError)
+          ::Rails.logger.warn(Log::Sidekiq.new(
+            level: Level::Warn,
+            event: Event::Error,
+            message: "#{error.class.name}: #{error.message}",
+            context: details
+          ))
+        end
+      end
+
+      JOB_ERROR_HANDLER = T.let(
+        ->(error, context, _config = nil) { log_job_error(error, context) },
+        T.proc.params(error: Exception, context: T::Hash[T.untyped, T.untyped], config: T.untyped).void
+      )
+
+      private_class_method :replace_default_error_handler, :log_job_error
     end
   end
 end

lib/log_struct/integrations/sidekiq/job_context_middleware.rb

@@ -0,0 +1,32 @@
+# typed: strict
+# frozen_string_literal: true
+
+module LogStruct
+  module Integrations
+    module Sidekiq
+      # Sidekiq server middleware that adds the running job's ID and class to every line
+      # logged while it runs, in the same `ctx` field as Sidekiq's own log lines. A query
+      # on ctx.jid then returns every line for one job run.
+      class JobContextMiddleware
+        extend T::Sig
+
+        sig do
+          params(
+            job_instance: T.untyped,
+            job_hash: T::Hash[String, T.untyped],
+            queue: String,
+            block: T.proc.returns(T.untyped)
+          ).returns(T.untyped)
+        end
+        def call(job_instance, job_hash, queue, &block)
+          context = {
+            jid: job_hash["jid"],
+            # Same precedence as Sidekiq's own context, so ActiveJob jobs show their class
+            class: job_hash["display_class"] || job_hash["wrapped"] || job_hash["class"]
+          }
+          ::SemanticLogger.named_tagged(sidekiq_context: context, &block)
+        end
+      end
+    end
+  end
+end

lib/log_struct/integrations/sidekiq/logger.rb

@@ -37,7 +37,9 @@ class Logger < LogStruct::SemanticLogger::Logger
               message: msg,
               process_id: ::Process.pid,
               thread_id: Sidekiq.tid,
-              context: ::Sidekiq::Context.current || {}
+              # Log a copy: SemanticLogger may still be serializing the struct on its appender
+              # thread when Sidekiq adds :elapsed to the live context after the job finishes.
+              context: (::Sidekiq::Context.current || {}).dup
             )
 
             # Pass the struct to SemanticLogger

lib/log_struct/semantic_logger/concerns/log_methods.rb

@@ -13,75 +13,75 @@ module LogMethods
         sig { params(message: T.untyped, payload: T.untyped, block: T.nilable(T.proc.returns(String))).returns(T::Boolean) }
         def debug(message = nil, payload = nil, &block)
           instrument_log(message, :debug)
-          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct) || message.is_a?(Hash)
+          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct)
             super(nil, payload: message, &block)
+          elsif message.is_a?(Hash)
+            # As the payload itself, so its keys are written at the top level of the line
+            super(nil, message, &block)
           else
             super
           end
-          broadcasts.each do |logger|
-            next unless logger.respond_to?(:debug)
-            message.is_a?(String) ? logger.debug(message) : (logger.debug(&block) if block)
-          end
+          broadcast_message(:debug, message, block)
           result
         end
 
         sig { params(message: T.untyped, payload: T.untyped, block: T.nilable(T.proc.returns(String))).returns(T::Boolean) }
         def info(message = nil, payload = nil, &block)
           instrument_log(message, :info)
-          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct) || message.is_a?(Hash)
+          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct)
             super(nil, payload: message, &block)
+          elsif message.is_a?(Hash)
+            # As the payload itself, so its keys are written at the top level of the line
+            super(nil, message, &block)
           else
             super
           end
-          broadcasts.each do |logger|
-            next unless logger.respond_to?(:info)
-            message.is_a?(String) ? logger.info(message) : (logger.info(&block) if block)
-          end
+          broadcast_message(:info, message, block)
           result
         end
 
         sig { params(message: T.untyped, payload: T.untyped, block: T.nilable(T.proc.returns(String))).returns(T::Boolean) }
         def warn(message = nil, payload = nil, &block)
           instrument_log(message, :warn)
-          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct) || message.is_a?(Hash)
+          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct)
             super(nil, payload: message, &block)
+          elsif message.is_a?(Hash)
+            # As the payload itself, so its keys are written at the top level of the line
+            super(nil, message, &block)
           else
             super
           end
-          broadcasts.each do |logger|
-            next unless logger.respond_to?(:warn)
-            message.is_a?(String) ? logger.warn(message) : (logger.warn(&block) if block)
-          end
+          broadcast_message(:warn, message, block)
           result
         end
 
         sig { params(message: T.untyped, payload: T.untyped, block: T.nilable(T.proc.returns(String))).returns(T::Boolean) }
         def error(message = nil, payload = nil, &block)
           instrument_log(message, :error)
-          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct) || message.is_a?(Hash)
+          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct)
             super(nil, payload: message, &block)
+          elsif message.is_a?(Hash)
+            # As the payload itself, so its keys are written at the top level of the line
+            super(nil, message, &block)
           else
             super
           end
-          broadcasts.each do |logger|
-            next unless logger.respond_to?(:error)
-            message.is_a?(String) ? logger.error(message) : (logger.error(&block) if block)
-          end
+          broadcast_message(:error, message, block)
           result
         end
 
         sig { params(message: T.untyped, payload: T.untyped, block: T.nilable(T.proc.returns(String))).returns(T::Boolean) }
         def fatal(message = nil, payload = nil, &block)
           instrument_log(message, :fatal)
-          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct) || message.is_a?(Hash)
+          result = if message.is_a?(LogStruct::Log::Interfaces::CommonFields) || message.is_a?(T::Struct)
             super(nil, payload: message, &block)
+          elsif message.is_a?(Hash)
+            # As the payload itself, so its keys are written at the top level of the line
+            super(nil, message, &block)
           else
             super
           end
-          broadcasts.each do |logger|
-            next unless logger.respond_to?(:fatal)
-            message.is_a?(String) ? logger.fatal(message) : (logger.fatal(&block) if block)
-          end
+          broadcast_message(:fatal, message, block)
           result
         end
 
@@ -94,6 +94,22 @@ def instrument_log(message, level)
 
           ::ActiveSupport::Notifications.instrument("log.logstruct", log: message, level: level)
         end
+
+        private
+
+        # Other loggers get String messages, and the block to call for the rest
+        sig { params(level: Symbol, message: T.untyped, block: T.nilable(T.proc.returns(String))).void }
+        def broadcast_message(level, message, block)
+          broadcasts.each do |logger|
+            next unless logger.respond_to?(level)
+
+            if message.is_a?(String)
+              logger.public_send(level, message)
+            elsif block
+              logger.public_send(level, &block)
+            end
+          end
+        end
       end
     end
   end

lib/log_struct/semantic_logger/formatter.rb

@@ -67,8 +67,10 @@ def call(log, logger)
         json = if logstruct
           # Use our formatter to process LogStruct types directly
           @logstruct_formatter.call(log.level, log.time, log.name, logstruct)
-        elsif log.payload.is_a?(Hash) || log.payload.is_a?(T::Struct)
-          # Process hashes and T::Structs through our formatter
+        elsif log.payload.is_a?(Hash)
+          # Process hashes through our formatter
+          @logstruct_formatter.call(log.level, log.time, log.name, without_reserved_keys(log.payload))
+        elsif log.payload.is_a?(T::Struct)
           @logstruct_formatter.call(log.level, log.time, log.name, log.payload)
         else
           # For plain messages, create a Plain log entry
@@ -80,11 +82,17 @@ def call(log, logger)
           @logstruct_formatter.call(log.level, log.time, log.name, plain_log)
         end
 
-        # Add request_id from named_tags if present
+        # Add the request ID and the running Sidekiq job's context from named_tags
         request_id = log.named_tags[:request_id]
-        if request_id
+        sidekiq_context = log.named_tags[:sidekiq_context]
+        if request_id || sidekiq_context
           data = JSON.parse(json)
-          data["req_id"] = request_id
+          data["req_id"] = request_id if request_id
+          if sidekiq_context
+            ctx = data["ctx"]
+            # A logged Hash's own ctx keeps its keys and gains the job's
+            data["ctx"] = ctx.is_a?(Hash) ? sidekiq_context.transform_keys(&:to_s).merge(ctx) : (ctx || sidekiq_context)
+          end
           json = data.to_json
         end
 
@@ -94,6 +102,26 @@ def call(log, logger)
 
       private
 
+      # A logged Hash may set its own src and evt, but the line's timestamp and level come from the logger
+      HASH_RESERVED_KEYS = T.let(%w[ts lvl].freeze, T::Array[String])
+
+      sig { params(hash: T::Hash[T.untyped, T.untyped]).returns(T::Hash[T.untyped, T.untyped]) }
+      def without_reserved_keys(hash)
+        reserved = hash.keys.select { |key| HASH_RESERVED_KEYS.include?(key.to_s) }
+        return hash if reserved.empty?
+
+        # A copy, since other appenders format the same Hash
+        copy = hash.dup
+        reserved.each do |key|
+          LogStruct.handle_exception(
+            ArgumentError.new("A logged Hash attempted to set reserved key: #{key}"),
+            source: Source::Internal
+          )
+          copy.delete(key)
+        end
+        copy
+      end
+
       # Extract a LogStruct from the various places it might be stored in a SemanticLogger::Log
       sig do
         params(log: ::SemanticLogger::Log).returns(

lib/log_struct/semantic_logger/logger.rb

@@ -90,7 +90,10 @@ def initialize(name = "Application", level: nil, filter: nil)
       # ::Logger, ActiveSupport::Logger, test doubles, etc.
       sig { params(logger: T.untyped).returns(T.untyped) }
       def broadcast_to(logger)
-        @broadcasts << logger
+        # Every SemanticLogger logger writes to the same appenders, so broadcasting to one
+        # would write each message twice. Sidekiq's Rails integration broadcasts Rails.logger
+        # to Sidekiq's logger, for example.
+        @broadcasts << logger unless logger.is_a?(::SemanticLogger::Logger)
         logger
       end