in_opentelemetry: handle invalid gzip gracefully (#42)

Author: Watson1978

lib/fluent/plugin/in_opentelemetry.rb

@@ -58,7 +58,7 @@ def start
       super
 
       if @http_config
-        http_handler = Opentelemetry::HttpInputHandler.new
+        http_handler = Opentelemetry::HttpInputHandler.new(@http_config, log)
         http_server_create_http_server(:in_opentelemetry_http_server, addr: @http_config.bind, port: @http_config.port, logger: log) do |serv|
           serv.post("/v1/logs") do |req|
             http_handler.logs(req) { |record| router.emit(tag_for(Opentelemetry::RECORD_TYPE_LOGS), Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_LOGS, "message" => record }) }

lib/fluent/plugin/opentelemetry/http_input_handler.rb

@@ -32,6 +32,11 @@ def close
 class Fluent::Plugin::Opentelemetry::HttpInputHandler
   using Fluent::PluginHelper::HttpServer::Extension
 
+  def initialize(http_config, logger)
+    @http_config = http_config
+    @logger = logger
+  end
+
   def logs(req, &block)
     common(req, Fluent::Plugin::Opentelemetry::Request::Logs, Fluent::Plugin::Opentelemetry::Response::Logs, &block)
   end
@@ -53,12 +58,20 @@ def common(req, request_class, response_class)
     return response_unsupported_media_type unless valid_content_type?(content_type)
     return response_bad_request(content_type) unless valid_content_encoding?(content_encoding)
 
-    body = Zlib::GzipReader.new(StringIO.new(body)).read if content_encoding == Fluent::Plugin::Opentelemetry::CONTENT_ENCODING_GZIP
+    if content_encoding == Fluent::Plugin::Opentelemetry::CONTENT_ENCODING_GZIP
+      begin
+        body = Zlib::GzipReader.new(StringIO.new(body)).read
+      rescue Zlib::Error => e
+        @logger.warn { "Failed to decompress gzip payload: #{e.message}" }
+        return response_bad_request(content_type)
+      end
+    end
 
     begin
       record = request_class.new(body).record
-    rescue Google::Protobuf::ParseError
+    rescue Google::Protobuf::ParseError => e
       # The format in request body does not comply with the OpenTelemetry protocol.
+      @logger.warn { "Failed to parse OpenTelemetry payload: #{e.message}" }
       return response_bad_request(content_type)
     end
 

test/fluent/plugin/test_in_opentelemetry_http.rb

@@ -113,15 +113,6 @@ def test_receive_compressed_json
       assert_equal(expected_events, d.events)
     end
 
-    def test_invalid_json
-      d = create_driver
-      res = d.run(expect_records: 0) do
-        post_json("/v1/logs", TestData::JSON::INVALID)
-      end
-
-      assert_equal(400, res.status)
-    end
-
     data("metrics" => {
            request_path: "/v1/metrics",
            request_data: TestData::ProtocolBuffers::METRICS,
@@ -162,6 +153,25 @@ def test_receive_compressed_protocol_buffers
       assert_equal(expected_events, d.events)
     end
 
+    def test_invalid_json
+      d = create_driver
+      res = d.run(expect_records: 0) do
+        post_json("/v1/logs", TestData::JSON::INVALID)
+      end
+
+      assert_equal(400, res.status)
+    end
+
+    def test_invalid_gzip_payload
+      d = create_driver
+      res = d.run(expect_records: 0) do
+        # Post plain JSON payload as gzip
+        post_json("/v1/logs", TestData::JSON::LOGS, headers: { "Content-Encoding" => "gzip" })
+      end
+
+      assert_equal(400, res.status)
+    end
+
     def test_invalid_protocol_buffers
       d = create_driver
       res = d.run(expect_records: 0) do