[AGENT-16861] Fix silent log loss on transient network errors

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].

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

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).