Make gRPC sending and receiving optional (#4)

* in_opentelemetry: split the Handler class into proper files

Signed-off-by: Shizuo Fujita <[email protected]>

* out_opentelemetry: split the Handler class into proper files

Signed-off-by: Shizuo Fujita <[email protected]>

* gemspec: move grpc to development dependency

Signed-off-by: Shizuo Fujita <[email protected]>

* rubocop: update rubocop.yml

Signed-off-by: Shizuo Fujita <[email protected]>

* README: update

Signed-off-by: Shizuo Fujita <[email protected]>

* Add tests

Signed-off-by: Shizuo Fujita <[email protected]>

* gemspec: remove grpc dependency

Signed-off-by: Shizuo Fujita <[email protected]>

* CI: test separately when grpc gem is installed.

Signed-off-by: Shizuo Fujita <[email protected]>

* example: install grpc gem manually

Signed-off-by: Shizuo Fujita <[email protected]>

* Gemfile: add grpc

Signed-off-by: Shizuo Fujita <[email protected]>

* CI: use io-stream 0.7.0 to avoid error on Windows

Signed-off-by: Shizuo Fujita <[email protected]>

---------

Signed-off-by: Shizuo Fujita <[email protected]>

Author: Watson1978

.github/workflows/test.yml

@@ -35,8 +35,14 @@ jobs:
       matrix:
         os: ['ubuntu-latest', 'macos-latest', 'windows-latest']
         ruby-version: ['3.2', '3.3', '3.4']
+        gemfile:
+          - grpc
+          - http
 
-    name: Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }}
+    env:
+      BUNDLE_GEMFILE: gemfiles/${{ matrix.gemfile }}.gemfile
+
+    name: ${{ matrix.gemfile }} - Ruby ${{ matrix.ruby-version }} on ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
       - name: Set up Ruby

.gitignore

@@ -9,4 +9,5 @@
 
 /example/node_modules/
 
+/gemfiles/*.lock
 Gemfile.lock

.rubocop.yml

@@ -6,12 +6,18 @@ AllCops:
   SuggestExtensions: false
   TargetRubyVersion: 3.2
 
+Gemspec/DevelopmentDependencies:
+  Enabled: false
+
 Layout/LineLength:
   Enabled: false
 
 Lint/UnusedMethodArgument:
   Enabled: false
 
+Lint/SuppressedException:
+  Enabled: false
+
 Metrics:
   Enabled: false
 

Gemfile

@@ -5,6 +5,7 @@ source "https://rubygems.org"
 # Specify your gem's dependencies in fluent-plugin-opentelemetry.gemspec
 gemspec
 
+gem "grpc", "~> 1.73"
 gem "grpc-tools"
 gem "irb"
 gem "rake", "~> 13.0"

README.md

@@ -46,6 +46,7 @@ This requires to receive data via HTTP/HTTPS.
 #### `<grpc>` section
 
 This requires to receive data via gRPC.
+It needs to install `grpc` gem manually to use this feature.
 
 > [!WARNING]
 > Now, gRPC feature status is experimental.
@@ -103,6 +104,7 @@ This requires to send data via HTTP/HTTPS.
 #### `<grpc>` section
 
 This requires to send data via gRPC.
+It needs to install `grpc` gem manually to use this feature.
 
 > [!WARNING]
 > Now, gRPC feature status is experimental.

example/fluentd/Dockerfile

@@ -7,7 +7,4 @@ RUN apt-get update && \
     make \
     gcc \
     git
-RUN git clone https://github.com/fluent-plugins-nursery/fluent-plugin-opentelemetry.git && \
-    cd fluent-plugin-opentelemetry && \
-    gem build fluent-plugin-opentelemetry.gemspec && \
-    gem install fluent-plugin-opentelemetry-*.gem
+RUN gem install fluent-plugin-opentelemetry grpc

fluent-plugin-opentelemetry.gemspec

@@ -22,7 +22,7 @@ Gem::Specification.new do |spec|
   spec.files = IO.popen(%w[git ls-files -z], chdir: __dir__, err: IO::NULL) do |ls|
     ls.readlines("\x0", chomp: true).reject do |f|
       (f == gemspec) ||
-        f.start_with?(*%w[test/ .git .github Gemfile example])
+        f.start_with?(*%w[test/ .git .github gemfiles Gemfile example])
     end
   end
   spec.require_paths = ["lib"]
@@ -31,5 +31,4 @@ Gem::Specification.new do |spec|
   spec.add_dependency("excon", "~> 1.2")
   spec.add_dependency("fluentd", "~> 1.18")
   spec.add_dependency("google-protobuf", "~> 4.30")
-  spec.add_dependency("grpc", "~> 1.71")
 end

gemfiles/grpc.gemfile

@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+source "https://rubygems.org"
+
+gemspec path: "../"
+
+gem "grpc", "~> 1.73"
+gem "io-stream", "= 0.7.0"
+gem "rake", "~> 13.0"
+gem "test-unit", "~> 3.0"
+gem "timecop"

gemfiles/http.gemfile

@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+source "https://rubygems.org"
+
+gemspec path: "../"
+
+# gem "grpc", "~> 1.73"
+gem "io-stream", "= 0.7.0"
+gem "rake", "~> 13.0"
+gem "test-unit", "~> 3.0"
+gem "timecop"

lib/fluent/plugin/in_opentelemetry.rb

@@ -2,29 +2,15 @@
 
 require "fluent/plugin/input"
 require "fluent/plugin/opentelemetry/constant"
-require "fluent/plugin/opentelemetry/request"
-require "fluent/plugin/opentelemetry/response"
-require "fluent/plugin/opentelemetry/service_handler"
+require "fluent/plugin/opentelemetry/http_input_handler"
 require "fluent/plugin_helper/http_server"
 require "fluent/plugin_helper/thread"
 
-require "zlib"
+begin
+  require "grpc"
 
-unless Fluent::PluginHelper::HttpServer::Request.method_defined?(:headers)
-  # This API was introduced at fluentd v1.19.0.
-  # Ref. https://github.com/fluent/fluentd/pull/4903
-  # If we have supported v1.19.0+ only, we can remove this patch.
-  module Fluent::PluginHelper::HttpServer
-    module Extension
-      refine Request do
-        def headers
-          @request.headers
-        end
-      end
-    end
-  end
-
-  using Fluent::PluginHelper::HttpServer::Extension
+  require "fluent/plugin/opentelemetry/grpc_input_handler"
+rescue LoadError
 end
 
 module Fluent::Plugin
@@ -57,6 +43,10 @@ class OpentelemetryInput < Input
     def configure(conf)
       super
 
+      if @grpc_config && !defined?(GRPC)
+        raise Fluent::ConfigError, "To use gRPC feature, please install grpc gem such as 'fluent-gem install grpc'."
+      end
+
       unless [@http_config, @grpc_config].any?
         raise Fluent::ConfigError, "Please configure either <http> or <grpc> section, or both."
       end
@@ -66,7 +56,7 @@ def start
       super
 
       if @http_config
-        http_handler = HttpHandler.new
+        http_handler = Opentelemetry::HttpInputHandler.new
         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, Fluent::EventTime.now, { type: Opentelemetry::RECORD_TYPE_LOGS, message: record }) }
@@ -82,7 +72,7 @@ def start
 
       if @grpc_config
         thread_create(:in_opentelemetry_grpc_server) do
-          grpc_handler = GrpcHandler.new(@grpc_config, log)
+          grpc_handler = Opentelemetry::GrpcInputHandler.new(@grpc_config, log)
           grpc_handler.run(
             logs: lambda { |record|
               router.emit(@tag, Fluent::EventTime.now, { type: Opentelemetry::RECORD_TYPE_LOGS, message: record })
@@ -97,115 +87,5 @@ def start
         end
       end
     end
-
-    class HttpHandler
-      def logs(req, &block)
-        common(req, Opentelemetry::Request::Logs, Opentelemetry::Response::Logs, &block)
-      end
-
-      def metrics(req, &block)
-        common(req, Opentelemetry::Request::Metrics, Opentelemetry::Response::Metrics, &block)
-      end
-
-      def traces(req, &block)
-        common(req, Opentelemetry::Request::Traces, Opentelemetry::Response::Traces, &block)
-      end
-
-      private
-
-      def common(req, request_class, response_class)
-        content_type = req.headers["content-type"]
-        content_encoding = req.headers["content-encoding"]&.first
-        return response_unsupported_media_type unless valid_content_type?(content_type)
-        return response_bad_request(content_type) unless valid_content_encoding?(content_encoding)
-
-        body = req.body
-        body = Zlib::GzipReader.new(StringIO.new(body)).read if content_encoding == Opentelemetry::CONTENT_ENCODING_GZIP
-
-        begin
-          record = request_class.new(body).record
-        rescue Google::Protobuf::ParseError
-          # The format in request body does not comply with the OpenTelemetry protocol.
-          return response_bad_request(content_type)
-        end
-
-        yield record
-
-        res = response_class.new
-        response(200, content_type, res.body(type: Opentelemetry::Response.type(content_type)))
-      end
-
-      def valid_content_type?(content_type)
-        case content_type
-        when Opentelemetry::CONTENT_TYPE_PROTOBUF, Opentelemetry::CONTENT_TYPE_JSON
-          true
-        else
-          false
-        end
-      end
-
-      def valid_content_encoding?(content_encoding)
-        return true if content_encoding.nil?
-
-        content_encoding == Opentelemetry::CONTENT_ENCODING_GZIP
-      end
-
-      def response(code, content_type, body)
-        [code, { Opentelemetry::CONTENT_TYPE => content_type }, body]
-      end
-
-      def response_unsupported_media_type
-        response(415, Opentelemetry::CONTENT_TYPE_PAIN, "415 unsupported media type, supported: [application/json, application/x-protobuf]")
-      end
-
-      def response_bad_request(content_type)
-        response(400, content_type, "") # TODO: fix body message
-      end
-    end
-
-    class GrpcHandler
-      class ExceptionInterceptor < GRPC::ServerInterceptor
-        def request_response(request:, call:, method:)
-          # call actual service
-          yield
-        rescue StandardError => e
-          puts "[#{method}] Error: #{e.message}"
-          raise
-        end
-      end
-
-      def initialize(grpc_config, logger)
-        @grpc_config = grpc_config
-        @logger = logger
-      end
-
-      def run(logs:, metrics:, traces:)
-        server = GRPC::RpcServer.new(interceptors: [ExceptionInterceptor.new])
-        server.add_http2_port("#{@grpc_config.bind}:#{@grpc_config.port}", :this_port_is_insecure)
-
-        logs_handler = Opentelemetry::ServiceHandler::Logs.new
-        logs_handler.callback = lambda { |request|
-          logs.call(request.to_json)
-          Opentelemetry::Response::Logs.build
-        }
-        server.handle(logs_handler)
-
-        metrics_handler = Opentelemetry::ServiceHandler::Metrics.new
-        metrics_handler.callback = lambda { |request|
-          metrics.call(request.to_json)
-          Opentelemetry::Response::Metrics.build
-        }
-        server.handle(metrics_handler)
-
-        traces_handler = Opentelemetry::ServiceHandler::Traces.new
-        traces_handler.callback = lambda { |request|
-          traces.call(request.to_json)
-          Opentelemetry::Response::Traces.build
-        }
-        server.handle(traces_handler)
-
-        server.run_till_terminated
-      end
-    end
   end
 end