diff --git a/docs/lib/integration-helpers.ts b/docs/lib/integration-helpers.ts index 5c44f181..cf0955a6 100644 --- a/docs/lib/integration-helpers.ts +++ b/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: diff --git a/lib/log_struct/integrations/sidekiq.rb b/lib/log_struct/integrations/sidekiq.rb index 2a98b814..fbc72bb6 100644 --- a/lib/log_struct/integrations/sidekiq.rb +++ b/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 diff --git a/lib/log_struct/integrations/sidekiq/job_context_middleware.rb b/lib/log_struct/integrations/sidekiq/job_context_middleware.rb new file mode 100644 index 00000000..3cb2dc5d --- /dev/null +++ b/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 diff --git a/lib/log_struct/integrations/sidekiq/logger.rb b/lib/log_struct/integrations/sidekiq/logger.rb index 6aef29ad..ad7531b6 100644 --- a/lib/log_struct/integrations/sidekiq/logger.rb +++ b/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 diff --git a/lib/log_struct/semantic_logger/concerns/log_methods.rb b/lib/log_struct/semantic_logger/concerns/log_methods.rb index 14f3d2c6..18538be7 100644 --- a/lib/log_struct/semantic_logger/concerns/log_methods.rb +++ b/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 diff --git a/lib/log_struct/semantic_logger/formatter.rb b/lib/log_struct/semantic_logger/formatter.rb index 03b592b4..a75357a8 100644 --- a/lib/log_struct/semantic_logger/formatter.rb +++ b/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( diff --git a/lib/log_struct/semantic_logger/logger.rb b/lib/log_struct/semantic_logger/logger.rb index 90fe3e45..d1518882 100644 --- a/lib/log_struct/semantic_logger/logger.rb +++ b/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 diff --git a/test/log_struct/integrations/sidekiq_test.rb b/test/log_struct/integrations/sidekiq_test.rb new file mode 100644 index 00000000..8119e4eb --- /dev/null +++ b/test/log_struct/integrations/sidekiq_test.rb @@ -0,0 +1,119 @@ +# typed: true +# frozen_string_literal: true + +require "test_helper" + +module LogStruct + module Integrations + class SidekiqTest < ActiveSupport::TestCase + setup do + @io = setup_isolated_logger + end + + teardown do + teardown_isolated_logger + end + + test "logs a copy of the Sidekiq context" do + logged = T.let([], T::Array[T.untyped]) + subscriber = ActiveSupport::Notifications.subscribe("log.logstruct") do |*_args, payload| + logged << payload[:log] + end + + ::Sidekiq::Context.with(jid: "abc123", class: "HardJob") do + Sidekiq::Logger.new("Sidekiq-Server").info("start") + ::Sidekiq::Context.add(:elapsed, "0.001") + end + + assert_equal({jid: "abc123", class: "HardJob"}, logged.last.context) + ensure + ActiveSupport::Notifications.unsubscribe(subscriber) + end + + test "lines logged while a job runs include the job's ID and class" do + job_hash = {"jid" => "abc123", "class" => "Sidekiq::ActiveJob::Wrapper", "wrapped" => "HardJob"} + + Sidekiq::JobContextMiddleware.new.call(nil, job_hash, "default") do + LogStruct::SemanticLogger::Logger.new("Rails").info("Rendering PDF") + end + LogStruct::SemanticLogger::Logger.new("Rails").info("After the job") + ::SemanticLogger.flush + + lines = @io.string.lines.map { |line| JSON.parse(line) } + inside = lines.find { |line| line["msg"] == "Rendering PDF" } + after = lines.find { |line| line["msg"] == "After the job" } + + assert_equal({"jid" => "abc123", "class" => "HardJob"}, T.must(inside)["ctx"]) + assert_nil T.must(after)["ctx"] + end + + test "a logged Hash's own ctx keeps its keys and gains the job's" do + job_hash = {"jid" => "abc123", "class" => "HardJob"} + + Sidekiq::JobContextMiddleware.new.call(nil, job_hash, "default") do + LogStruct::SemanticLogger::Logger.new("Rails").info({msg: "Rendering PDF", ctx: {template_id: "tpl_1"}}) + end + ::SemanticLogger.flush + + line = JSON.parse(T.must(@io.string.lines.last)) + + assert_equal({"jid" => "abc123", "class" => "HardJob", "template_id" => "tpl_1"}, line["ctx"]) + end + + test "a failed job is logged once, as a structured line without its arguments" do + sidekiq_config = ::Sidekiq::Config.new + sidekiq_config.logger = Sidekiq::Logger.new("Sidekiq-Server") + Sidekiq.send(:replace_default_error_handler, sidekiq_config) + error = ArgumentError.new("Template not found") + error.set_backtrace(["app/jobs/hard_job.rb:12:in 'perform'"]) + job = {"jid" => "abc123", + "class" => "HardJob", + "queue" => "default", + "retry_count" => 2, + "args" => ["secret@example.com"]} + + sidekiq_config.error_handlers.each { |handler| handler.call(error, {context: "Job raised exception", job: job}, sidekiq_config) } + ::SemanticLogger.flush + + lines = sidekiq_lines + + assert_equal 1, lines.size + line = T.must(lines.first) + + assert_equal ["error", "warn"], [line["evt"], line["lvl"]] + assert_equal ["ArgumentError", "Template not found"], [line["error_class"], line["msg"]] + assert_equal ["abc123", "HardJob", "default", 2], [line["jid"], line["job_class"], line["queue"], line["retry_count"]] + refute_includes @io.string, "secret@example.com" + end + + test "ActiveJob jobs' errors are left to the ActiveJob integration" do + sidekiq_config = ::Sidekiq::Config.new + Sidekiq.send(:replace_default_error_handler, sidekiq_config) + job = {"jid" => "abc123", "class" => "Sidekiq::ActiveJob::Wrapper", "wrapped" => "HardJob"} + + sidekiq_config.error_handlers.each { |handler| handler.call(ArgumentError.new("boom"), {job: job}, sidekiq_config) } + ::SemanticLogger.flush + + assert_empty sidekiq_lines + end + + test "an app's own error handlers stay" do + sidekiq_config = ::Sidekiq::Config.new + app_handler = ->(_error, _context, _config) {} + sidekiq_config.error_handlers << app_handler + + Sidekiq.send(:replace_default_error_handler, sidekiq_config) + + refute_includes sidekiq_config.error_handlers, ::Sidekiq::Config::ERROR_HANDLER + assert_includes sidekiq_config.error_handlers, app_handler + assert_equal 2, sidekiq_config.error_handlers.size + end + + private + + def sidekiq_lines + @io.string.lines.map { |line| JSON.parse(line) }.select { |line| line["src"] == "sidekiq" } + end + end + end +end diff --git a/test/log_struct/semantic_logger_test.rb b/test/log_struct/semantic_logger_test.rb index 343dfbcf..2abbc0f4 100644 --- a/test/log_struct/semantic_logger_test.rb +++ b/test/log_struct/semantic_logger_test.rb @@ -75,9 +75,9 @@ class SemanticLoggerTest < ActiveSupport::TestCase # Parse the first line of JSON output (last line might be empty) log = JSON.parse(output.lines.first.strip) - # Check the payload or direct fields - data = log["payload"] || log + data = log + assert_nil log["payload"] assert_equal "User login", data["message"] assert_equal "test", data["user"] @@ -86,6 +86,40 @@ class SemanticLoggerTest < ActiveSupport::TestCase assert data["password"]["_filtered"], "Password should include _filtered summary" end + test "writes a Hash's own src and evt at the top level" do + clear_log_buffer(@io) + + %i[debug info warn error fatal].each do |level| + @logger.public_send(level, {src: "rails", evt: "rate_limit", rule: "logins/#{level}"}) + end + ::SemanticLogger.flush + + lines = @io.string.lines.map { |line| JSON.parse(line) } + + assert_equal %w[logins/debug logins/info logins/warn logins/error logins/fatal], lines.pluck("rule") + lines.each do |line| + assert_equal "rails", line["src"] + assert_equal "rate_limit", line["evt"] + assert_nil line["payload"] + end + end + + test "keeps the line's timestamp and level when a logged Hash sets them" do + clear_log_buffer(@io) + reported = [] + + LogStruct.stub(:handle_exception, ->(error, **) { reported << error.message }) do + @logger.info({msg: "Posted to Slack", ts: "1700000000.000100", lvl: "debug"}) + ::SemanticLogger.flush + end + + line = JSON.parse(T.must(@io.string.lines.last)) + + assert_equal "info", line["lvl"] + refute_equal "1700000000.000100", line["ts"] + assert_equal ["A logged Hash attempted to set reserved key: ts", "A logged Hash attempted to set reserved key: lvl"], reported.uniq + end + test "supports tagged logging" do @logger.tagged("request-123") do @logger.info("Processing request") @@ -143,6 +177,17 @@ class SemanticLoggerTest < ActiveSupport::TestCase assert_includes secondary_output, "Test broadcast message" end + test "broadcast_to ignores SemanticLogger loggers, which share its appenders" do + sidekiq_logger = LogStruct::Integrations::Sidekiq::Logger.new("Sidekiq-Server") + + @logger.broadcast_to(sidekiq_logger) + @logger.info("Written once") + ::SemanticLogger.flush + + assert_empty @logger.broadcasts + assert_equal 1, @io.string.scan("Written once").size + end + test "stop_broadcasting_to removes logger from broadcasts" do secondary_io = StringIO.new secondary_logger = Logger.new(secondary_io)