diff --git a/CHANGELOG.md b/CHANGELOG.md index 58e0d43..227d69a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,5 @@ ## Unreleased -- Switch to GitHub Actions Trusted Publishing for gem releases (replaces manual API key publishing) [#89](https://github.com/DataDog/fluent-plugin-datadog/pull/89) +- Fix silent log loss on transient network errors: HTTP timeouts/resets (e.g. `Net::ReadTimeout`, `EOFError`, `Errno::ECONNRESET`) are now retried, and failed flushes are re-raised so Fluentd's buffer retry engages instead of dropping the chunk ## 0.15.0 - Provide a configuration option to delete kubernetes and docker attributes from the log after the relevant information has been extracted into tags [#78](https://github.com/DataDog/fluent-plugin-datadog/pull/78) by [@sambart19]. diff --git a/lib/fluent/plugin/out_datadog.rb b/lib/fluent/plugin/out_datadog.rb index 8df25e4..f15f084 100644 --- a/lib/fluent/plugin/out_datadog.rb +++ b/lib/fluent/plugin/out_datadog.rb @@ -175,8 +175,9 @@ def write(chunk) process_tcp_event(record[0], @max_retries, @max_backoff, DD_MAX_BATCH_SIZE) end end - rescue Exception => e - log.error("Uncaught processing exception in datadog forwarder #{e.message}") + rescue StandardError => e + log.error("Processing exception in datadog forwarder #{e.class}: #{e.message}") + raise end end @@ -328,6 +329,10 @@ def send_retries(payload, max_retries, max_backoff) retries += 1 retry end + # Bounded retries exhausted: re-raise so the caller (and, ultimately, + # Fluentd core's buffer retry) is signalled instead of the failure being + # swallowed and the chunk silently dropped. + raise end end @@ -345,6 +350,28 @@ class DatadogHTTPClient < DatadogClient require 'net/http' require 'net/http/persistent' + # Transient network exceptions that warrant a retry. This mirrors the set + # Ruby's Net::HTTP retries for idempotent requests (see + # Net::HTTP#max_retries=), which notably does NOT include POST, plus the + # connection-establishment errors net-http-persistent wraps in its own + # Error. Because our log POSTs are non-idempotent, Net::HTTP will not retry + # them for us, so we classify these ourselves and route them through + # send_retries (and, on exhaustion, up to Fluentd core). + RETRYABLE_NETWORK_EXCEPTIONS = [ + Net::OpenTimeout, + Net::ReadTimeout, + EOFError, + IOError, + SocketError, + Errno::ECONNRESET, + Errno::ECONNREFUSED, + Errno::ECONNABORTED, + Errno::EPIPE, + Errno::ETIMEDOUT, + OpenSSL::SSL::SSLError, + Net::HTTP::Persistent::Error, + ].freeze + def initialize(logger, use_ssl, no_ssl_validation, host, ssl_port, port, http_proxy, custom_headers, use_compression, api_key, force_v1_routes = false) @logger = logger protocol = use_ssl ? "https" : "http" @@ -384,7 +411,13 @@ def initialize(logger, use_ssl, no_ssl_validation, host, ssl_port, port, http_pr def send(payload) request = Net::HTTP::Post.new @uri.request_uri request.body = payload - response = @client.request @uri, request + begin + response = @client.request @uri, request + rescue *RETRYABLE_NETWORK_EXCEPTIONS => e + # Transient network failure before we ever saw a response. Net::HTTP + # won't retry a POST for us, so surface it as retryable. + raise RetryableError.new "Unable to send payload, transient network error: #{e.class}: #{e.message}" + end res_code = response.code.to_i # on a backend error or on an http 429, retry with backoff if res_code >= 500 || res_code == 429 diff --git a/test/plugin/test_out_datadog.rb b/test/plugin/test_out_datadog.rb index 9996574..be12839 100644 --- a/test/plugin/test_out_datadog.rb +++ b/test/plugin/test_out_datadog.rb @@ -465,6 +465,99 @@ def create_valid_subject end end + # v1 routes + sub_test_case "http transient network errors (v1 routes)" do + test "should raise RetryableError on Net::ReadTimeout" do + api_key = 'XXX' + stub_dd_request_with_error(api_key, Net::ReadTimeout) + payload = '{}' + client = Fluent::DatadogOutput::DatadogHTTPClient.new Logger.new(STDOUT), false, false, "datadog.com", 443, 80, nil, {}, false, api_key, true + assert_raise(Fluent::DatadogOutput::RetryableError) do + client.send(payload) + end + end + + test "should raise RetryableError on EOFError (end of file reached)" do + api_key = 'XXX' + stub_dd_request_with_error(api_key, EOFError) + payload = '{}' + client = Fluent::DatadogOutput::DatadogHTTPClient.new Logger.new(STDOUT), false, false, "datadog.com", 443, 80, nil, {}, false, api_key, true + assert_raise(Fluent::DatadogOutput::RetryableError) do + client.send(payload) + end + end + + test "should raise RetryableError on Errno::ECONNRESET" do + api_key = 'XXX' + stub_dd_request_with_error(api_key, Errno::ECONNRESET) + payload = '{}' + client = Fluent::DatadogOutput::DatadogHTTPClient.new Logger.new(STDOUT), false, false, "datadog.com", 443, 80, nil, {}, false, api_key, true + assert_raise(Fluent::DatadogOutput::RetryableError) do + client.send(payload) + end + end + end + + sub_test_case "send_retries" do + # Logger stub that accepts Fluentd's warn(msg, hash) signature, which the + # stdlib Logger does not. + class NullLogger + def warn(*args, **kwargs); end + def error(*args, **kwargs); end + def info(*args, **kwargs); end + def debug(*args, **kwargs); end + end + + def build_failing_client(error_to_raise) + Class.new(Fluent::DatadogOutput::DatadogClient) do + attr_reader :calls + define_method(:initialize) do + @logger = NullLogger.new + @calls = 0 + @error_to_raise = error_to_raise + end + define_method(:send) do |_payload| + @calls += 1 + raise @error_to_raise + end + end.new + end + + test "re-raises RetryableError after exhausting bounded retries" do + client = build_failing_client(Fluent::DatadogOutput::RetryableError.new("boom")) + assert_raise(Fluent::DatadogOutput::RetryableError) do + client.send_retries("payload", 1, 1) + end + # initial attempt + 1 retry + assert_equal 2, client.calls + end + + test "does not swallow non-RetryableError exceptions" do + client = build_failing_client(ArgumentError.new("bad")) + assert_raise(ArgumentError) do + client.send_retries("payload", 1, 1) + end + # non-retryable: attempted once, no retries + assert_equal 1, client.calls + end + end + + sub_test_case "write error propagation" do + test "write re-raises instead of swallowing errors" do + plugin = create_valid_subject + def plugin.process_http_events(*) + raise Fluent::DatadogOutput::RetryableError.new("boom") + end + fake_chunk = Object.new + def fake_chunk.msgpack_each + yield ["{}"] + end + assert_raise(Fluent::DatadogOutput::RetryableError) do + plugin.write(fake_chunk) + end + end + end + def stub_dd_request_with_return_code(api_key, return_code, v2_routes = false) if v2_routes stub_dd_request_v2_routes(api_key).