add

Author: Watson1978

lib/fluent/plugin/out_otlp.rb

@@ -68,15 +68,13 @@ def write(chunk)
       response = connection.post
 
       if response.status != 200
-        message = "got error response #{response.status} from '#{uri}'"
-
         if @http_config.retryable_response_codes&.include?(response.status)
-          raise RetryableResponse, message
+          raise RetryableResponse, "got retryable error response from '#{uri}', response code is #{response.status}"
         end
         if @http_config.error_response_as_unrecoverable
-          raise Fluent::UnrecoverableError, message
+          raise Fluent::UnrecoverableError, "got unrecoverable error response from '#{uri}', response code is #{response.status}"
         else
-          log.error message
+          log.error "got error response from '#{uri}', response code is #{response.status}"
         end
       end
     end

test/fluent/plugin/test_out_otlp.rb

@@ -140,17 +140,65 @@ def test_send_compressed_message
     assert_equal(TestData::ProtocolBuffers::LOGS, decompress(server_request.body).force_encoding(Encoding::ASCII_8BIT))
   end
 
-  def test_illigal
+  def test_unrecoverable_error
     set_server_response_code(500)
     event = { "type" => "otlp_logs", "message" => TestData::JSON::LOGS }
 
     d = create_driver
-    d.run(default_tag: "otlp.test") do
+    d.run(default_tag: "otlp.test", shutdown: false) do
       d.feed(event)
     end
+
+    assert_match(%r{got unrecoverable error response from 'http://127.0.0.1:4318/v1/logs', response code is 500},
+                 d.instance.log.out.logs.join)
+
+    d.instance_shutdown
   end
 
+  def test_error_with_disabled_unrecoverable
+    set_server_response_code(500)
+    event = { "type" => "otlp_logs", "message" => TestData::JSON::LOGS }
+
+    d = create_driver(<<~CONFIG)
+      <http>
+        endpoint "http://127.0.0.1:4318"
+        error_response_as_unrecoverable false
+      </http>
+    CONFIG
+    d.run(default_tag: "otlp.test", shutdown: false) do
+      d.feed(event)
+    end
+
+    assert_match(%r{got error response from 'http://127.0.0.1:4318/v1/logs', response code is 500},
+                 d.instance.log.out.logs.join)
 
+    d.instance_shutdown
+  end
+
+  def test_write_with_retryable_response
+    old_report_on_exception = Thread.report_on_exception
+    Thread.report_on_exception = false # thread finished as invalid state since RetryableResponse raises.
+
+    set_server_response_code(503)
+    event = { "type" => "otlp_logs", "message" => TestData::JSON::LOGS }
+
+    d = create_driver(<<~CONFIG)
+      <http>
+        endpoint "http://127.0.0.1:4318"
+        retryable_response_codes [503]
+      </http>
+    CONFIG
+
+    assert_raise(Fluent::Plugin::OtlpOutput::RetryableResponse) do
+      d.run(default_tag: "otlp.test", shutdown: false) do
+        d.feed(event)
+      end
+    end
+
+    d.instance_shutdown(log: $log)
+  ensure
+    Thread.report_on_exception = old_report_on_exception
+  end
 
   def decompress(data)
     Zlib::GzipReader.new(StringIO.new(data)).read