Retry batches answered with 429 or 5xx, and report rejected batches once

lib/logtail/log_devices/http.rb

@@ -1,6 +1,7 @@
 require "msgpack"
 require "net/https"
 require "set"
+require "time"
 require "zlib"
 
 require "logtail/config"
@@ -27,6 +28,10 @@ class HTTP
       MAX_UNTRACKED_DEPTH = 100 # nested hashes and arrays, see #encodable_value
       INITIAL_RECONNECT_WAIT = 1 # second
       MAX_RECONNECT_WAIT = 30 # seconds
+      MAX_RETRY_AFTER = 60 # seconds
+      # The HTTP statuses of rejected batches this process has warned about, see {#report_rejected_batch}.
+      REPORTED_REJECTIONS = []
+      REPORTED_REJECTIONS_LOCK = Mutex.new
       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}.
@@ -265,7 +270,7 @@ def reset_if_forked
         def dequeue_requests
           requests = []
           while @request_queue.size > 0 && (request_attempt = @request_queue.deq)
-            requests << request_attempt.request
+            requests << request_attempt
           end
           requests
         end
@@ -454,16 +459,18 @@ def flush_async
           req = build_request(msgs)
           if !req.nil?
             Logtail::Config.instance.debug { "New request placed on queue" }
-            request_attempt = RequestAttempt.new(req)
+            request_attempt = RequestAttempt.new(req, msgs.size)
             @request_queue.enq(request_attempt)
           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?
+        # Returns whether all of them were delivered. A request answered with 408, 429 or 5xx isn't
+        # retried, nothing would deliver the retry; one rejected with any other status that isn't
+        # 2xx is reported like in {#deliver_requests}. 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(request_attempts)
+          return true if request_attempts.empty?
 
           http = build_http
           http.open_timeout = http.read_timeout = SYNCHRONOUS_DELIVERY_TIMEOUT
@@ -477,8 +484,16 @@ def deliver_synchronously(requests)
             http.open_timeout = nil
             http.start
           end
-          requests.each { |request| @last_resp = http.request(request) }
-          true
+          delivered = true
+          request_attempts.each do |request_attempt|
+            resp = @last_resp = http.request(request_attempt.request)
+            next if resp.code.start_with?("2")
+
+            delivered = false
+            Logtail::Config.instance.debug { "Log delivery failed! status: #{resp.code}, body: #{resp.body}" }
+            report_rejected_batch(request_attempt, resp) unless resp.code == "408" || resp.code == "429" || resp.code.start_with?("5")
+          end
+          delivered
         rescue SignalException
           raise
         rescue Exception => e
@@ -528,7 +543,7 @@ def deliver_late_lines
           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)])
+            @late_delivery_failed = !deliver_synchronously([RequestAttempt.new(build_request(msgs), msgs.size)])
           end
           true
         end
@@ -577,9 +592,10 @@ def build_http
         end
 
         # Creates a loop that processes the `@request_queue` on an interval. After a failed
-        # connection it waits before reconnecting, twice as long after every consecutive
-        # failure up to {MAX_RECONNECT_WAIT}, so an unreachable host is not retried in a busy
-        # loop. A delivered request starts the wait over (see {#deliver_requests}).
+        # connection, or a 429 or 5xx response, it waits before reconnecting, twice as long after
+        # every consecutive failure up to {MAX_RECONNECT_WAIT}, so an unreachable host is not
+        # retried in a busy loop. A Retry-After header can make the wait longer. A delivered
+        # request starts the wait over (see {#deliver_requests}).
         def request_outlet
           loop do
             http = build_http
@@ -609,7 +625,8 @@ def request_outlet
         # Creates a loop that delivers requests over an open (kept alive) HTTP connection.
         # If the connection dies, the request is thrown back onto the queue and
         # the method returns. It is the responsibility of the caller to implement retries
-        # and establish a new connection.
+        # and establish a new connection. A 429 or 5xx response is handled the same way, and
+        # a request rejected with any other status is dropped (see {#report_rejected_batch}).
         def deliver_requests(conn)
           num_reqs = 0
 
@@ -632,16 +649,7 @@ def deliver_requests(conn)
                 resp = conn.request(request_attempt.request)
               rescue => e
                 Logtail::Config.instance.debug { "#deliver_requests error: #{e.message}" }
-
-                # Throw the request back on the queue for a retry if it has been attempted less
-                # than 3 times
-                if request_attempt.attempts < 3
-                  Logtail::Config.instance.debug { "Request is being retried, #{request_attempt.attempts} previous attempts" }
-                  @request_queue.enq(request_attempt)
-                else
-                  Logtail::Config.instance.debug { "Request is being dropped, #{request_attempt.attempts} previous attempts" }
-                end
-
+                retry_or_drop(request_attempt)
                 return false
               ensure
                 @requests_in_flight -= 1
@@ -650,21 +658,70 @@ def deliver_requests(conn)
               num_reqs += 1
 
               @last_resp = resp
-              @reconnect_wait = INITIAL_RECONNECT_WAIT
+              delivered = resp.code.start_with?("2")
 
               Logtail::Config.instance.debug do
-                if resp.code == "202"
+                if delivered
                   "Logs successfully sent! View your logs at https://telemetry.betterstack.com"
                 else
                   "Log delivery failed! status: #{resp.code}, body: #{resp.body}"
                 end
               end
+
+              # A request the server didn't read (408, which Better Stack also sends for a new
+              # connection that stayed unused too long), too many requests or a server error: retry
+              # the request like after a failed connection, without starting the wait over, and
+              # wait as long as Retry-After asks.
+              if resp.code == "408" || resp.code == "429" || resp.code.start_with?("5")
+                retry_or_drop(request_attempt)
+                @reconnect_wait = [@reconnect_wait, retry_after(resp)].max
+                return false
+              end
+
+              report_rejected_batch(request_attempt, resp) unless delivered
+              @reconnect_wait = INITIAL_RECONNECT_WAIT
             end
           end
 
           true
         end
 
+        # Throws the request back on the queue for a retry if it has been attempted less
+        # than 3 times
+        def retry_or_drop(request_attempt)
+          if request_attempt.attempts < 3
+            Logtail::Config.instance.debug { "Request is being retried, #{request_attempt.attempts} previous attempts" }
+            @request_queue.enq(request_attempt)
+          else
+            Logtail::Config.instance.debug { "Request is being dropped, #{request_attempt.attempts} previous attempts" }
+          end
+        end
+
+        # The seconds to wait before a retry that the Retry-After header asks for, given in
+        # seconds or as an HTTP date, at most {MAX_RETRY_AFTER}. 0 without a valid header.
+        def retry_after(resp)
+          value = resp["Retry-After"].to_s.strip
+          seconds = value.match?(/\A\d+\z/) ? value.to_i : (Time.httpdate(value) - Time.now).ceil
+          seconds.clamp(0, MAX_RETRY_AFTER)
+        rescue ArgumentError
+          0
+        end
+
+        # Warns about a batch Better Stack rejected, once per HTTP status in this process. It
+        # goes to stderr, never to a Logtail logger, whose lines would be rejected the same way.
+        def report_rejected_batch(request_attempt, resp)
+          first_rejection = REPORTED_REJECTIONS_LOCK.synchronize do
+            !REPORTED_REJECTIONS.include?(resp.code) && REPORTED_REJECTIONS.push(resp.code)
+          end
+          return unless first_rejection
+
+          lines = request_attempt.line_count
+          status = "HTTP #{resp.code} #{resp.message}".strip
+          hint = " - check your source token" if resp.code == "401" || resp.code == "403"
+          warn("Logtail: Better Stack rejected #{lines || "some"} log #{lines == 1 ? "line" : "lines"} " \
+            "with #{status}#{hint}. Further rejections with this status won't be reported.")
+        end
+
         # Builds the `Authorization` header value for HTTP delivery to the Logtail API.
         def authorization_payload
           "Bearer #{@source_token}"

lib/logtail/log_devices/http/request_attempt.rb

@@ -4,11 +4,12 @@ class HTTP
       # Represents an attempt to deliver a request. Requests can be retried, hence
       # why we keep track of the number of attempts.
       class RequestAttempt
-        attr_reader :attempts, :request
+        attr_reader :attempts, :request, :line_count
 
-        def initialize(req)
+        def initialize(req, line_count = nil)
           @attempts = 0
           @request = req
+          @line_count = line_count
         end
 
         def attempted!

spec/logtail/log_devices/http_spec.rb

@@ -94,6 +94,26 @@
       http.close
       expect(Process.clock_gettime(Process::CLOCK_MONOTONIC) - closing).to be < 1
     end
+
+    it "drops the lines written after it once Better Stack answered one of them with 429 or 5xx" do
+      stub = stub_request(:post, "https://in.logs.betterstack.com/").to_return({status: 503}, {status: 202})
+      http.close
+
+      2.times { http.write(Logtail::LogEntry.new("INFO", Time.now, nil, "test log message", nil, nil)) }
+
+      expect(stub).to have_been_requested.once
+    end
+
+    it "reports a line written after it that Better Stack rejects" do
+      # A status is reported once per process, so none has been reported yet.
+      stub_const("Logtail::LogDevices::HTTP::REPORTED_REJECTIONS", [])
+      stub_request(:post, "https://in.logs.betterstack.com/").to_return(status: [403, "Forbidden"])
+      http.close
+      expect(http).to receive(:warn).once.with("Logtail: Better Stack rejected 1 log line with HTTP 403 " \
+        "Forbidden - check your source token. Further rejections with this status won't be reported.")
+
+      http.write(Logtail::LogEntry.new("INFO", Time.now, nil, "test log message", nil, nil))
+    end
   end
 
   # Testing a private method because it helps break down our tests
@@ -168,6 +188,41 @@
       http.instance_variable_get(:@request_outlet_thread).kill.join
     end
 
+    context "when Better Stack answers the delivery in the calling thread with an error status" do
+      let(:http) { described_class.new("MYKEY", flush_continuously: false) }
+
+      # A status is reported once per process, so none has been reported yet in each example.
+      before { stub_const("Logtail::LogDevices::HTTP::REPORTED_REJECTIONS", []) }
+
+      it "drops batches rejected with 401 and warns only once" do
+        stub = stub_request(:post, "https://in.logs.betterstack.com/").to_return(status: [401, "Unauthorized"])
+        expect(http).to receive(:warn).once.with("Logtail: Better Stack rejected 2 log lines with HTTP 401 " \
+          "Unauthorized - check your source token. Further rejections with this status won't be reported.")
+
+        2.times do
+          2.times { |i| http.write(Logtail::LogEntry.new("INFO", time, nil, "line #{i}", nil, nil)) }
+          http.flush
+        end
+
+        expect(stub).to have_been_requested.twice
+        http.close
+      end
+
+      it "neither retries nor reports a batch answered with 408, 429 or 5xx" do
+        stub = stub_request(:post, "https://in.logs.betterstack.com/").
+          to_return({status: 503}, {status: 429}, {status: [408, "Request Time-out"]})
+        expect(http).not_to receive(:warn)
+
+        3.times do
+          http.write(Logtail::LogEntry.new("INFO", time, nil, "test log message", nil, nil))
+          http.flush
+        end
+
+        expect(stub).to have_been_requested.times(3)
+        http.close
+      end
+    end
+
     it "waits up to 5 seconds for a slow host when it delivers in the calling thread" do
       # Takes 3 seconds to answer each request
       slow_ingest = Class.new(LocalIngestServer) do
@@ -522,6 +577,106 @@ def delivered_entry
       end
     end
 
+    context "when Better Stack answers with an error status" do
+      # Raised to leave the outlet's endless loop, see above.
+      let(:stop_outlet) { Class.new(Exception) }
+      let(:http_device) { described_class.new("MYKEY", flush_continuously: false, requests_per_conn: 1) }
+      let(:request_queue) { http_device.instance_variable_get(:@request_queue) }
+      let(:waits) { [] }
+
+      before do
+        allow(http_device).to receive(:sleep) { |seconds| waits << seconds }
+        # Leave the loop once there is nothing left to deliver.
+        allow(request_queue).to receive(:deq).and_wrap_original { |deq| deq.call || raise(stop_outlet) }
+        # A status is reported once per process, so none has been reported yet in each example.
+        stub_const("Logtail::LogDevices::HTTP::REPORTED_REJECTIONS", [])
+      end
+
+      def queue_batch(size)
+        size.times { |i| http_device.write(Logtail::LogEntry.new("INFO", Time.now, nil, "line #{i}", nil, nil)) }
+        http_device.send(:flush_async)
+      end
+
+      it "retries a batch answered with 500 and delivers it once" do
+        stub = stub_request(:post, "https://in.logs.betterstack.com/").to_return({status: 500}, {status: 202})
+        queue_batch(1)
+
+        expect { http_device.send(:request_outlet) }.to raise_error(stop_outlet)
+        expect(stub).to have_been_requested.twice
+        expect(waits).to eq([1])
+      end
+
+      it "retries a batch answered with 408 without a warning, and delivers it once" do
+        # Better Stack answers 408 when a new connection stays unused for more than about 10 seconds
+        stub = stub_request(:post, "https://in.logs.betterstack.com/").
+          to_return({status: [408, "Request Time-out"], headers: {"Connection" => "close"}}, {status: 202})
+        queue_batch(1)
+        expect(http_device).not_to receive(:warn)
+
+        expect { http_device.send(:request_outlet) }.to raise_error(stop_outlet)
+        expect(stub).to have_been_requested.twice
+        expect(waits).to eq([1])
+      end
+
+      it "waits before retrying a batch answered with 429 at least as long as Retry-After says" do
+        stub = stub_request(:post, "https://in.logs.betterstack.com/").
+          to_return({status: 429, headers: {"Retry-After" => "3"}}, {status: 202})
+        queue_batch(1)
+
+        expect { http_device.send(:request_outlet) }.to raise_error(stop_outlet)
+        expect(stub).to have_been_requested.twice
+        expect(waits).to eq([3])
+      end
+
+      it "reads Retry-After as a date too, and waits 60 seconds at most" do
+        Timecop.freeze(Time.utc(2026, 10, 1, 12, 0, 0)) do
+          stub = stub_request(:post, "https://in.logs.betterstack.com/").to_return(
+            {status: 503, headers: {"Retry-After" => (Time.now + 5).httpdate}},
+            {status: 503, headers: {"Retry-After" => "3600"}},
+            {status: 202}
+          )
+          queue_batch(1)
+
+          expect { http_device.send(:request_outlet) }.to raise_error(stop_outlet)
+          expect(stub).to have_been_requested.times(3)
+          expect(waits).to eq([5, 60])
+        end
+      end
+
+      it "drops a batch answered with 503 three times, backing off in between" do
+        stub = stub_request(:post, "https://in.logs.betterstack.com/").to_return(status: 503)
+        queue_batch(1)
+        expect(http_device).not_to receive(:warn)
+
+        expect { http_device.send(:request_outlet) }.to raise_error(stop_outlet)
+        expect(stub).to have_been_requested.times(3)
+        expect(waits).to eq([1, 2, 4])
+      end
+
+      it "drops batches answered with 401 and warns only once" do
+        stub = stub_request(:post, "https://in.logs.betterstack.com/").to_return(status: [401, "Unauthorized"])
+        queue_batch(2)
+        queue_batch(2)
+        expect(http_device).to receive(:warn).once.with("Logtail: Better Stack rejected 2 log lines with HTTP 401 " \
+          "Unauthorized - check your source token. Further rejections with this status won't be reported.")
+
+        expect { http_device.send(:request_outlet) }.to raise_error(stop_outlet)
+        expect(stub).to have_been_requested.twice
+        expect(waits).to eq([])
+      end
+
+      it "drops a batch answered with 413 and warns" do
+        stub = stub_request(:post, "https://in.logs.betterstack.com/").to_return(status: [413, "Payload Too Large"])
+        queue_batch(1)
+
+        expect do
+          expect { http_device.send(:request_outlet) }.to raise_error(stop_outlet)
+        end.to output(/^Logtail: Better Stack rejected 1 log line with HTTP 413 Payload Too Large\. Further rejections with this status won't be reported\.$/).to_stderr
+        expect(stub).to have_been_requested.once
+        expect(waits).to eq([])
+      end
+    end
+
     it "lets close stop the outlet while it waits to reconnect" do
       connection_attempts = 0
       allow_any_instance_of(Net::HTTP).to receive(:start) do