diff --git a/lib/logtail/log_devices/http.rb b/lib/logtail/log_devices/http.rb index 926ab1e..7449a61 100755 --- a/lib/logtail/log_devices/http.rb +++ b/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}" diff --git a/lib/logtail/log_devices/http/request_attempt.rb b/lib/logtail/log_devices/http/request_attempt.rb index 803e434..7bea258 100755 --- a/lib/logtail/log_devices/http/request_attempt.rb +++ b/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! diff --git a/spec/logtail/log_devices/http_spec.rb b/spec/logtail/log_devices/http_spec.rb index 9adcb17..b41f27a 100755 --- a/spec/logtail/log_devices/http_spec.rb +++ b/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