add grpc support

Author: Watson1978

.rubocop_todo.yml

@@ -1,18 +1,34 @@
 # This configuration was generated by
 # `rubocop --auto-gen-config`
-# on 2025-04-10 06:21:04 UTC using RuboCop version 1.75.2.
+# on 2025-04-15 08:39:58 UTC using RuboCop version 1.75.2.
 # The point is for the user to remove these configuration records
 # one by one as the offenses are removed from the code base.
 # Note that changes in the inspected code, or installation of new
 # versions of RuboCop, may require this file to be generated again.
 
+# Offense count: 2
+# This cop supports safe autocorrection (--autocorrect).
+# Configuration parameters: Max, AllowHeredoc, AllowURI, URISchemes, IgnoreCopDirectives, AllowedPatterns, SplitStrings.
+# URISchemes: http, https
+Layout/LineLength:
+  Exclude:
+    - 'lib/fluent/plugin/in_otlp.rb'
+
 # Offense count: 1
 # This cop supports safe autocorrection (--autocorrect).
-Lint/RedundantStringCoercion:
+# Configuration parameters: AutoCorrect, ContextCreatingMethods, MethodCreatingMethods.
+Lint/UselessAccessModifier:
   Exclude:
     - 'lib/fluent/plugin/out_otlp.rb'
 
-# Offense count: 11
+# Offense count: 3
+# Configuration parameters: MinNameLength, AllowNamesEndingInNumbers, AllowedNames, ForbiddenNames.
+# AllowedNames: as, at, by, cc, db, id, if, in, io, ip, of, on, os, pp, to
+Naming/MethodParameterName:
+  Exclude:
+    - 'lib/fluent/plugin/otlp/service_stub.rb'
+
+# Offense count: 21
 # Configuration parameters: AllowedConstants.
 Style/Documentation:
   Exclude:
@@ -21,9 +37,11 @@ Style/Documentation:
     - 'lib/fluent/plugin/in_otlp.rb'
     - 'lib/fluent/plugin/otlp/request.rb'
     - 'lib/fluent/plugin/otlp/response.rb'
+    - 'lib/fluent/plugin/otlp/service_handler.rb'
+    - 'lib/fluent/plugin/otlp/service_stub.rb'
     - 'lib/fluent/plugin/out_otlp.rb'
 
-# Offense count: 3
+# Offense count: 4
 # This cop supports safe autocorrection (--autocorrect).
 # Configuration parameters: MinBodyLength, AllowConsecutiveConditionals.
 Style/GuardClause:
@@ -40,3 +58,25 @@ Style/IfUnlessModifier:
 Style/MixinUsage:
   Exclude:
     - 'test/helper.rb'
+
+# Offense count: 1
+# This cop supports safe autocorrection (--autocorrect).
+# Configuration parameters: Strict, AllowedNumbers, AllowedPatterns.
+Style/NumericLiterals:
+  MinDigits: 6
+
+# Offense count: 1
+# This cop supports unsafe autocorrection (--autocorrect-all).
+# Configuration parameters: ConvertCodeThatCanStartToReturnNil, AllowedMethods, MaxChainLength.
+# AllowedMethods: present?, blank?, presence, try, try!
+Style/SafeNavigation:
+  Exclude:
+    - 'lib/fluent/plugin/out_otlp.rb'
+
+# Offense count: 3
+# This cop supports safe autocorrection (--autocorrect).
+# Configuration parameters: ExactNameMatch, AllowPredicates, AllowDSLWriters, IgnoreClassMethods, AllowedMethods.
+# AllowedMethods: to_ary, to_a, to_c, to_enum, to_h, to_hash, to_i, to_int, to_io, to_open, to_path, to_proc, to_r, to_regexp, to_str, to_s, to_sym
+Style/TrivialAccessors:
+  Exclude:
+    - 'lib/fluent/plugin/otlp/service_handler.rb'

lib/fluent/plugin/in_otlp.rb

@@ -4,6 +4,7 @@
 require "fluent/plugin/otlp/constant"
 require "fluent/plugin/otlp/request"
 require "fluent/plugin/otlp/response"
+require "fluent/plugin/otlp/service_handler"
 require "fluent/plugin_helper/http_server"
 
 require "zlib"
@@ -107,21 +108,76 @@ def response_bad_request(content_type)
       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 run(logs:, metrics:, traces:)
+        Thread.new do
+          server = GRPC::RpcServer.new(interceptors: [ExceptionInterceptor.new])
+          server.add_http2_port("0.0.0.0:4317", :this_port_is_insecure)
+
+          logs_handler = Otlp::ServiceHandler::Logs.new
+          logs_handler.callback = lambda { |request|
+            logs.call(request.to_json)
+            Otlp::Response::Logs.build
+          }
+          server.handle(logs_handler)
+
+          metrics_handler = Otlp::ServiceHandler::Metrics.new
+          metrics_handler.callback = lambda { |request|
+            metrics.call(request.to_json)
+            Otlp::Response::Metrics.build
+          }
+          server.handle(metrics_handler)
+
+          traces_handler = Otlp::ServiceHandler::Traces.new
+          traces_handler.callback = lambda { |request|
+            traces.call(request.to_json)
+            Otlp::Response::Traces.build
+          }
+          server.handle(traces_handler)
+
+          server.run_till_terminated
+        end
+      end
+    end
+
     def start
       super
 
-      handler = HttpHandler.new
+      http_handler = HttpHandler.new
       http_server_create_http_server(:in_otel_http_server_helper, addr: @http_config.bind, port: @http_config.port, logger: log) do |serv|
         serv.post("/v1/logs") do |req|
-          handler.logs(req) { |record| router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_LOGS, message: record }) }
+          http_handler.logs(req) { |record| router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_LOGS, message: record }) }
         end
         serv.post("/v1/metrics") do |req|
-          handler.metrics(req) { |record| router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_METRICS, message: record }) }
+          http_handler.metrics(req) { |record| router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_METRICS, message: record }) }
         end
         serv.post("/v1/traces") do |req|
-          handler.traces(req) { |record| router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_TRACES, message: record }) }
+          http_handler.traces(req) { |record| router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_TRACES, message: record }) }
         end
       end
+
+      grpc_handler = GrpcHandler.new
+      grpc_handler.run(
+        logs: lambda { |record|
+          router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_LOGS, message: record })
+        },
+        metrics: lambda { |record|
+          router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_METRICS, message: record })
+        },
+        traces: lambda { |record|
+          router.emit(@tag, Fluent::EventTime.now, { type: Otlp::RECORD_TYPE_TRACES, message: record })
+        }
+      )
     end
   end
 end

lib/fluent/plugin/otlp/response.rb

@@ -20,56 +20,68 @@ def self.type(content_type)
   end
 
   class Logs
-    def initialize(rejected: 0, error: "")
-      @response = Opentelemetry::Proto::Collector::Logs::V1::ExportLogsServiceResponse.new(
+    def self.build(rejected: 0, error: "")
+      Opentelemetry::Proto::Collector::Logs::V1::ExportLogsServiceResponse.new(
         partial_success: Opentelemetry::Proto::Collector::Logs::V1::ExportLogsPartialSuccess.new(
           rejected_log_records: rejected,
           error_message: error
         )
       )
     end
 
+    def initialize(rejected: 0, error: "")
+      @response = Logs.build(rejected: rejected, error: error)
+    end
+
     def body(type:)
       if type == :protobuf
-        Opentelemetry::Proto::Collector::Logs::V1::ExportLogsServiceResponse.encode(@response)
+        @response.class.encode(@response)
       else
         @response.to_json
       end
     end
   end
 
   class Metrics
-    def initialize(rejected: 0, error: "")
-      @response = Opentelemetry::Proto::Collector::Metrics::V1::ExportMetricsServiceResponse.new(
+    def self.build(rejected: 0, error: "")
+      Opentelemetry::Proto::Collector::Metrics::V1::ExportMetricsServiceResponse.new(
         partial_success: Opentelemetry::Proto::Collector::Metrics::V1::ExportMetricsPartialSuccess.new(
           rejected_data_points: rejected,
           error_message: error
         )
       )
     end
 
+    def initialize(rejected: 0, error: "")
+      @response = Metrics.build(rejected: rejected, error: error)
+    end
+
     def body(type:)
       if type == :protobuf
-        Opentelemetry::Proto::Collector::Metrics::V1::ExportMetricsServiceResponse.encode(@response)
+        @response.class.encode(@response)
       else
         @response.to_json
       end
     end
   end
 
   class Traces
-    def initialize(rejected: 0, error: "")
-      @response = Opentelemetry::Proto::Collector::Trace::V1::ExportTraceServiceResponse.new(
+    def self.build(rejected: 0, error: "")
+      Opentelemetry::Proto::Collector::Trace::V1::ExportTraceServiceResponse.new(
         partial_success: Opentelemetry::Proto::Collector::Trace::V1::ExportTracePartialSuccess.new(
           rejected_spans: rejected,
           error_message: error
         )
       )
     end
 
+    def initialize(rejected: 0, error: "")
+      @response = Traces.build(rejected: rejected, error: error)
+    end
+
     def body(type:)
       if type == :protobuf
-        Opentelemetry::Proto::Collector::Trace::V1::ExportTraceServiceResponse.encode(@response)
+        @response.class.encode(@response)
       else
         @response.to_json
       end

lib/fluent/plugin/otlp/service_handler.rb

@@ -0,0 +1,41 @@
+# frozen_string_literal: true
+
+require "fluent/plugin/otlp/response"
+
+require "opentelemetry/proto/collector/logs/v1/logs_service_services_pb"
+require "opentelemetry/proto/collector/metrics/v1/metrics_service_services_pb"
+require "opentelemetry/proto/collector/trace/v1/trace_service_services_pb"
+
+require "grpc"
+
+class Fluent::Plugin::Otlp::ServiceHandler
+  class Logs < Opentelemetry::Proto::Collector::Logs::V1::LogsService::Service
+    def callback=(block)
+      @callback = block
+    end
+
+    def export(req)
+      @callback.call(req)
+    end
+  end
+
+  class Metrics < Opentelemetry::Proto::Collector::Metrics::V1::MetricsService::Service
+    def callback=(block)
+      @callback = block
+    end
+
+    def export(req, _call)
+      @callback.call(req)
+    end
+  end
+
+  class Traces < Opentelemetry::Proto::Collector::Trace::V1::TraceService::Service
+    def callback=(block)
+      @callback = block
+    end
+
+    def export(req)
+      @callback.call(req)
+    end
+  end
+end

lib/fluent/plugin/otlp/service.rb renamed to lib/fluent/plugin/otlp/service_stub.rb

@@ -9,7 +9,7 @@
 
 require "grpc"
 
-class Fluent::Plugin::Otlp::Service
+class Fluent::Plugin::Otlp::ServiceStub
   class Logs
     def initialize(host, creds, **kw)
       @stub = Opentelemetry::Proto::Collector::Logs::V1::LogsService::Stub.new(host, creds, **kw)
@@ -41,6 +41,5 @@ def export(json)
       message = Opentelemetry::Proto::Collector::Trace::V1::ExportTraceServiceRequest.decode_json(json)
       @stub.export(message)
     end
-
   end
-end
+end

lib/fluent/plugin/out_otlp.rb

@@ -2,7 +2,7 @@
 
 require "fluent/plugin/otlp/constant"
 require "fluent/plugin/otlp/request"
-require "fluent/plugin/otlp/service"
+require "fluent/plugin/otlp/service_stub"
 require "fluent/plugin/output"
 
 require "excon"
@@ -174,11 +174,11 @@ def export(chunk)
 
         case record["type"]
         when Otlp::RECORD_TYPE_LOGS
-          service = Otlp::Service::Logs.new(@grpc_config.endpoint, credential)
+          service = Otlp::ServiceStub::Logs.new(@grpc_config.endpoint, credential)
         when Otlp::RECORD_TYPE_METRICS
-          service = Otlp::Service::Metrics.new(@grpc_config.endpoint, credential)
+          service = Otlp::ServiceStub::Metrics.new(@grpc_config.endpoint, credential)
         when Otlp::RECORD_TYPE_TRACES
-          service = Otlp::Service::Traces.new(@grpc_config.endpoint, credential)
+          service = Otlp::ServiceStub::Traces.new(@grpc_config.endpoint, credential)
         end
 
         begin