diff --git a/lib/logtail/log_devices/http.rb b/lib/logtail/log_devices/http.rb index eeb6c7a..c266d7f 100755 --- a/lib/logtail/log_devices/http.rb +++ b/lib/logtail/log_devices/http.rb @@ -24,6 +24,7 @@ class HTTP USER_AGENT = "Logtail Ruby/#{Logtail::VERSION} (HTTP)".freeze INITIAL_RECONNECT_WAIT = 1 # second MAX_RECONNECT_WAIT = 30 # seconds + SYNCHRONOUS_DELIVERY_TIMEOUT = 5 # seconds, to connect and to read the response # Instantiates a new HTTP log device that can be passed to {Logtail::Logger#initialize}. # @@ -85,7 +86,14 @@ def initialize(source_token, options = {}) @request_queue = options[:request_queue] || FlushableDroppingSizedQueue.new(25) @successive_error_count = 0 @requests_in_flight = 0 + @last_resp = nil @reconnect_wait = INITIAL_RECONNECT_WAIT + @closed = false + @late_delivery_failed = false + + # Delivers what is still buffered when the process exits. One hook per device, however + # many loggers write to it. + at_exit { close } end # Write a new log line message to the buffer, and flush asynchronously if the @@ -96,12 +104,21 @@ def write(msg) return unless Logtail.config.send_to_better_stack?(msg) @msg_queue.enq(msg) + # No thread delivers what is written after #close, e.g. by an at_exit hook that runs + # after the device's own. + return deliver_late_lines if @closed # Lazily start flush threads to ensure threads are alive after forking processes. # If the threads are started during instantiation they will not be copied when # the current process is forked. This is the case with various web servers, # such as phusion passenger. - ensure_flush_threads_are_started + begin + ensure_flush_threads_are_started + rescue ThreadError + # Ruby refuses new threads while it shuts down, e.g. to a thread that logs in an + # `ensure` block as it is killed at exit. + return deliver_late_lines + end if @msg_queue.full? Logtail::Config.instance.debug { "Flushing HTTP buffer via write" } @@ -119,8 +136,12 @@ def flush true end - # Closes the log device, cleans up, and attempts one last delivery. + # Closes the log device, cleans up, and attempts one last delivery. Closing it again does + # nothing; lines written after it are delivered right away (see {#write}). def close + return if @closed + @closed = true + # Kill the flush thread immediately since we are about to flush again. @flush_thread.kill.join if @flush_thread @@ -241,6 +262,35 @@ def flush_async end end + # Sends the requests in the calling thread, for when no outlet thread delivers them. + # Returns whether all of them were sent. Errors only go to the debug log, also those that + # aren't StandardErrors (WebMock refuses to connect with one); signals are raised as usual. + def deliver_synchronously(requests) + return true if requests.empty? + + http = build_http + http.open_timeout = http.read_timeout = SYNCHRONOUS_DELIVERY_TIMEOUT + begin + http.start + rescue ThreadError + # While Ruby shuts down it refuses new threads, and Net::HTTP (before Ruby 4.0) needs + # one to time out connecting. Then it connects without a timeout, but only to a host + # that has answered before. + raise if @last_resp.nil? + http.open_timeout = nil + http.start + end + requests.each { |request| @last_resp = http.request(request) } + true + rescue SignalException + raise + rescue Exception => e + Logtail::Config.instance.debug { "Synchronous delivery failed: #{e.message}" } + false + ensure + http.finish if http && http.started? + end + # Waits on the request queue. This is used in {#flush} to ensure # the log data has been delivered before returning. def wait_on_request_queue @@ -250,6 +300,10 @@ def wait_on_request_queue Logtail::Config.instance.debug { "Request queue is empty and no requests are in flight, finish waiting" } return true end + if outlet_stalled? + Logtail::Config.instance.debug { "The HTTP outlet can't deliver the requests, finish waiting" } + return false + end Logtail::Config.instance.debug do "Request size #{@request_queue.size}, reqs in-flight #{@requests_in_flight}, " \ "continue waiting (iteration #{i + 1})" @@ -258,6 +312,30 @@ def wait_on_request_queue end end + # Whether the outlet thread can't deliver anything while {#close} waits for it: the thread + # is dead, or the host has never answered and the outlet has already waited to reconnect + # after a failed connection (@reconnect_wait grows after each wait until a response). + def outlet_stalled? + return true unless @request_outlet_thread && @request_outlet_thread.alive? + + @last_resp.nil? && @reconnect_wait > INITIAL_RECONNECT_WAIT + end + + # Delivers the buffered lines in the calling thread, for {#write} when no thread can. After + # a delivery fails, e.g. to an unreachable host, later lines are dropped so they can't + # hold up the exit one by one. + def deliver_late_lines + msgs = @msg_queue.flush + return true if msgs.empty? + + if @late_delivery_failed + Logtail::Config.instance.debug { "Dropping #{msgs.size} log lines, an earlier synchronous delivery failed" } + else + @late_delivery_failed = !deliver_synchronously([build_request(msgs)]) + end + true + end + # Flushes the message queue on an interval. You will notice that {#write} also # flushes the buffer if it is full. This method takes note of this via the # `@last_async_flush` variable as to not flush immediately after a write flush. @@ -343,13 +421,15 @@ def deliver_requests(conn) Logtail::Config.instance.debug { "Waiting on next request, threads waiting: #{@request_queue.size}" } end + # Counted as in flight before it leaves the queue, so close never sees neither + @requests_in_flight += 1 request_attempt = @request_queue.deq if request_attempt.nil? + @requests_in_flight -= 1 sleep(1) else request_attempt.attempted! - @requests_in_flight += 1 begin resp = conn.request(request_attempt.request) diff --git a/lib/logtail/logger.rb b/lib/logtail/logger.rb index c2dc67f..f21bc21 100755 --- a/lib/logtail/logger.rb +++ b/lib/logtail/logger.rb @@ -192,8 +192,6 @@ def initialize(*io_devices_and_loggers) Logtail::Config.instance.debug { "Logtail::Logger instantiated, level: #{level}, formatter: #{formatter.class}" } @initialized = true - - at_exit { self.close } end # Sets a new formatted on the logger. diff --git a/spec/logtail/log_devices/http_exit_spec.rb b/spec/logtail/log_devices/http_exit_spec.rb new file mode 100644 index 0000000..625d607 --- /dev/null +++ b/spec/logtail/log_devices/http_exit_spec.rb @@ -0,0 +1,142 @@ +require "spec_helper" +require "delegate" + +# A local ingesting server that takes 3 seconds to answer each request +class SlowAnsweringIngestServer < LocalIngestServer + class SlowSocket < SimpleDelegator + def write(*args) + sleep 3 + super + end + end + + private + + def serve(socket) + super(SlowSocket.new(socket)) + end +end + +# What the HTTP device does while the process that logs with it exits. These run logtail in a +# real Ruby process, and count what reaches a local ingesting server. +describe Logtail::LogDevices::HTTP, "when the process exits" do + let(:ingest) { LocalIngestServer.new } + + after { ingest.stop } + + # A local port that nothing listens on, so connecting to it is refused. + def unused_port + server = TCPServer.new("127.0.0.1", 0) + server.addr[1] + ensure + server.close + end + + it "delivers a line that a thread logs while Ruby kills it at exit, without printing an error" do + result = run_ruby(<<-RUBY) + require "logtail" + logger = Logtail::Logger.new(Logtail::LogDevices::HTTP.new("token", #{ingest.device_options})) + logger.info("logged before exit") + started = Queue.new + Thread.new do + begin + started << true + sleep + ensure + logger.info("logged by a thread that Ruby kills at exit") + end + end + started.pop + RUBY + + expect(result.stderr).not_to include("log writing failed") + expect(ingest.messages).to contain_exactly("logged before exit", "logged by a thread that Ruby kills at exit") + end + + it "doesn't warn about the instance variables it uses to close, with warnings on" do + # Ruby 2.7 and older warn about an instance variable that is read before it's set + result = run_ruby(<<-RUBY) + $VERBOSE = true + require "logtail" + logger = Logtail::Logger.new(Logtail::LogDevices::HTTP.new("token", #{ingest.device_options})) + logger.info("logged before close") + logger.close + logger.info("logged after close") + RUBY + + expect(result.status).to be_success, result.stderr + expect(result.stderr).not_to match(/@(closed|late_delivery_failed|last_resp) not initialized/) + expect(ingest.messages).to contain_exactly("logged before close", "logged after close") + end + + it "delivers a line that an at_exit hook logs after the device closed" do + result = run_ruby(<<-RUBY) + require "logtail" + logger = nil + at_exit { logger.info("logged by an at_exit hook registered first, which runs last") } + logger = Logtail::Logger.new(Logtail::LogDevices::HTTP.new("token", #{ingest.device_options})) + logger.info("logged before exit") + RUBY + + expect(result.status).to be_success, result.stderr + expect(ingest.messages).to contain_exactly("logged before exit", "logged by an at_exit hook registered first, which runs last") + end + + it "waits up to 5 seconds for a slow host when it delivers lines logged after closing" do + slow_ingest = SlowAnsweringIngestServer.new + result = run_ruby(<<-RUBY) + require "logtail" + logger = Logtail::Logger.new(Logtail::LogDevices::HTTP.new("token", #{slow_ingest.device_options})) + logger.close + logger.info("first line after close") + logger.info("second line after close") + RUBY + + # A delivery that times out would drop the second line + expect(result.status).to be_success, result.stderr + expect(slow_ingest.messages).to contain_exactly("first line after close", "second line after close") + ensure + slow_ingest.stop if slow_ingest + end + + it "delivers a batch that the outlet takes off the queue while close checks whether it's done" do + result = run_ruby(<<-RUBY) + require "logtail" + # Widens the moment between the outlet taking a batch off the queue and counting it as in flight + Logtail::LogDevices::HTTP::RequestAttempt.prepend(Module.new { define_method(:attempted!) { sleep 1.5; super() } }) + logger = Logtail::Logger.new(Logtail::LogDevices::HTTP.new("token", #{ingest.device_options})) + logger.info("logged right before exit") + RUBY + + expect(result.status).to be_success, result.stderr + expect(ingest.messages).to eq(["logged right before exit"]) + end + + it "closes a device once at exit, however many loggers write to it" do + result = run_ruby(<<-RUBY) + require "logtail" + closes = 0 + at_exit { puts "closes: " + closes.to_s } + Logtail::LogDevices::HTTP.prepend(Module.new { define_method(:close) { closes += 1; super() } }) + device = Logtail::LogDevices::HTTP.new("token", #{ingest.device_options}) + 3.times { |n| Logtail::Logger.new(device).info("line " + n.to_s) } + RUBY + + expect(result.stdout).to eq("closes: 1\n"), result.stderr + expect(ingest.messages).to contain_exactly("line 0", "line 1", "line 2") + end + + it "exits within 5 seconds when the ingesting host can't be reached, however many loggers there are" do + result = run_ruby(<<-RUBY) + exiting = nil + at_exit { puts Process.clock_gettime(Process::CLOCK_MONOTONIC) - exiting } + require "logtail" + device = Logtail::LogDevices::HTTP.new("token", ingesting_host: "127.0.0.1", ingesting_port: #{unused_port}, ingesting_scheme: "http") + 2.times { |n| Logtail::Logger.new(device).info("line " + n.to_s) } + exiting = Process.clock_gettime(Process::CLOCK_MONOTONIC) + RUBY + + expect(result.status).to be_success, result.stderr + expect(result.stdout.to_f).to be < 5 + end +end diff --git a/spec/logtail/log_devices/http_spec.rb b/spec/logtail/log_devices/http_spec.rb index 2f6ac4e..fa807da 100755 --- a/spec/logtail/log_devices/http_spec.rb +++ b/spec/logtail/log_devices/http_spec.rb @@ -68,6 +68,29 @@ expect(http).to receive(:flush).exactly(1).times http.close end + + it "does nothing when called again, e.g. by the device's at_exit hook" do + http.close + expect(http).not_to receive(:flush) + http.close + end + + it "doesn't let a line written after it raise when delivering it fails, whatever the error" do + http.close + + # WebMock refuses to connect with an error that isn't a StandardError + expect { http.write(Logtail::LogEntry.new("INFO", Time.now, nil, "test log message", nil, nil)) }.not_to raise_error + end + + it "stops waiting for the outlet thread once it has died" do + stub_request(:post, "https://in.logs.betterstack.com/") + http.write(Logtail::LogEntry.new("INFO", Time.now, nil, "test log message", nil, nil)) + http.instance_variable_get(:@request_outlet_thread).kill.join + + closing = Process.clock_gettime(Process::CLOCK_MONOTONIC) + http.close + expect(Process.clock_gettime(Process::CLOCK_MONOTONIC) - closing).to be < 1 + end end # Testing a private method because it helps break down our tests diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index a16418e..36a96a7 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -12,6 +12,7 @@ require File.join(File.dirname(__FILE__), 'support', 'timecop') require File.join(File.dirname(__FILE__), 'support', 'webmock') require File.join(File.dirname(__FILE__), 'support', 'logtail') +require File.join(File.dirname(__FILE__), 'support', 'processes') RSpec::Support::ObjectFormatter.default_instance.max_formatted_output_length = 5_000 diff --git a/spec/support/processes.rb b/spec/support/processes.rb new file mode 100644 index 0000000..15332fe --- /dev/null +++ b/spec/support/processes.rb @@ -0,0 +1,82 @@ +require "msgpack" +require "rbconfig" +require "socket" +require "tmpdir" +require "zlib" + +# A local stand-in for the Better Stack ingesting host, for tests that run logtail in a separate +# process: it answers every request with 202 and records the message of every line it receives. +class LocalIngestServer + def initialize + @server = TCPServer.new("127.0.0.1", 0) + @messages = [] + @lock = Mutex.new + @connections = [] + @thread = Thread.new do + loop { @connections << Thread.new(@server.accept) { |socket| serve(socket) } } + end + end + + # The options for Logtail::LogDevices::HTTP.new that deliver to this server, as Ruby code. + def device_options + %(ingesting_host: "127.0.0.1", ingesting_port: #{@server.addr[1]}, ingesting_scheme: "http") + end + + def messages + @lock.synchronize { @messages.dup } + end + + def stop + @thread.kill.join + @connections.each { |connection| connection.kill.join } + @server.close + end + + private + + def serve(socket) + while socket.gets + headers = {} + while (header = socket.gets) && header != "\r\n" + name, value = header.split(":", 2) + headers[name.downcase] = value.strip + end + lines = MessagePack.unpack(Zlib::Inflate.inflate(socket.read(headers["content-length"].to_i))) + @lock.synchronize { @messages.concat(lines.map { |line| line["message"] }) } + socket.write("HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\n\r\n") + end + rescue IOError, SystemCallError + ensure + socket.close + end +end + +module ProcessHelpers + ProcessResult = Struct.new(:stdout, :stderr, :status) + + # Runs the Ruby script in a new process with logtail from lib/ and waits for it to exit, at + # most `timeout` seconds (generous: TruffleRuby starts slowly on CI). + def run_ruby(script, timeout: 120) + Dir.mktmpdir do |dir| + File.write(File.join(dir, "script.rb"), script) + lib = File.expand_path("../../lib", __dir__) + pid = Process.spawn(RbConfig.ruby, "-I", lib, File.join(dir, "script.rb"), + out: File.join(dir, "stdout"), err: File.join(dir, "stderr")) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + status = nil + until (status = Process.waitpid2(pid, Process::WNOHANG)) + if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + Process.kill("KILL", pid) + Process.wait(pid) + raise "The Ruby process didn't exit within #{timeout} seconds:\n#{File.read(File.join(dir, "stderr"))}" + end + sleep 0.05 + end + ProcessResult.new(File.read(File.join(dir, "stdout")), File.read(File.join(dir, "stderr")), status[1]) + end + end +end + +RSpec.configure do |config| + config.include ProcessHelpers +end