Exit quickly when Better Stack can't be reached, and deliver lines logged during shutdown

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)

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.

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

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

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