in_opentelemetry: add ${type} placeholder support in tag parameter (#8)

* in_opentelemetry: add ${type} placeholder support in tag parameter

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

* Update example

---------

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

Author: Watson1978

README.md

@@ -34,6 +34,33 @@ To receive data, this plugin requires `<http>` or `<grpc>` section, or both.
 |-----------|--------|----------------------|----------|
 | tag       | string | The tag of the event | required |
 
+`tag` parameter supports `${type}` placeholder and it will be expanded with data type.
+
+* log data
+  * `${type}` will be replaced to `logs`
+* metric data
+  * `${type}` will be replaced to `metrics`
+* trace data
+  * `${type}` will be replaced to `traces`
+
+This can be used to change the output destination for each message type.
+
+Example:
+
+```
+<source>
+  @type opentelemetry
+
+  # Expand to opentelemetry.logs, opentelemetry.metrics, opentelemetry.traces according to received data.
+  tag opentelemetry.${type}
+
+  <http>
+    bind 127.0.0.1
+    port 4318
+  </http>
+</source>
+```
+
 #### `<http>` section
 
 This requires to receive data via HTTP/HTTPS.

example/README.md

@@ -15,10 +15,9 @@ flowchart LR
     A[app/metrics.js] -->|metric data / otlp| C[Fluentd]
     B[app/tracing.js] -->|trace data / otlp| C[Fluentd]
     C[Fluentd] -->|metric data / otlp| D[Otel Collector]
-    C[Fluentd] -->|trace data / otlp| D[Otel Collector]
+    C[Fluentd] -->|trace data / otlp| F[Jaeger]
     C[Fluentd] -->|sample log / Fluentd Forward Protocol| D[Otel Collector]
     D[Otel Collector] -->|metric data| E[Prometheus / Grafana]
-    D[Otel Collector] -->|trace data| F[Jaeger]
     D[Otel Collector] -->|sample log| G[Elasticsearch / Kibana]
 ```
 

example/collector/otel-collector-config.yaml

@@ -13,18 +13,18 @@ exporters:
     endpoint: http://elasticsearch:9200
   prometheus:
     endpoint: "0.0.0.0:8889"
-  otlp:
-    endpoint: "jaeger:4317"
-    tls:
-      insecure: true
+  # otlp:
+  #   endpoint: "jaeger:4317"
+  #   tls:
+  #     insecure: true
   debug:
     verbosity: detailed
 
 service:
   pipelines:
-    traces:
-      receivers: [otlp]
-      exporters: [debug, otlp]
+    # traces:
+    #   receivers: [otlp]
+    #   exporters: [debug, otlp]
     metrics:
       receivers: [otlp]
       exporters: [debug, prometheus]

example/fluentd/fluent.conf

@@ -1,6 +1,6 @@
 <source>
   @type opentelemetry
-  tag opentelemetry
+  tag opentelemetry.${type}
 
   # Experimental
   # <grpc>
@@ -14,6 +14,21 @@
   </http>
 </source>
 
+# Send traces data to jaeger directly using OpenTelemetry Protocol
+<match opentelemetry.traces>
+  @type opentelemetry
+  <buffer []>
+    @type memory
+    flush_mode immediate
+    chunk_limit_size 1m
+  </buffer>
+
+  <http>
+    endpoint "http://jaeger:4318"
+  </http>
+</match>
+
+# Send other OpenTelemetry data to OpenTelemetry Collector
 <match opentelemetry.**>
   @type opentelemetry
   <buffer []>

lib/fluent/plugin/in_opentelemetry.rb

@@ -43,6 +43,8 @@ class OpentelemetryInput < Input
     def configure(conf)
       super
 
+      expand_tag_placeholders(@tag)
+
       if @grpc_config && !defined?(GRPC)
         raise Fluent::ConfigError, "To use gRPC feature, please install grpc gem such as 'fluent-gem install grpc'."
       end
@@ -59,13 +61,13 @@ def start
         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 }) }
+            http_handler.logs(req) { |record| router.emit(tag_for(Opentelemetry::RECORD_TYPE_LOGS), Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_LOGS, "message" => record }) }
           end
           serv.post("/v1/metrics") do |req|
-            http_handler.metrics(req) { |record| router.emit(@tag, Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_METRICS, "message" => record }) }
+            http_handler.metrics(req) { |record| router.emit(tag_for(Opentelemetry::RECORD_TYPE_METRICS), Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_METRICS, "message" => record }) }
           end
           serv.post("/v1/traces") do |req|
-            http_handler.traces(req) { |record| router.emit(@tag, Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_TRACES, "message" => record }) }
+            http_handler.traces(req) { |record| router.emit(tag_for(Opentelemetry::RECORD_TYPE_TRACES), Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_TRACES, "message" => record }) }
           end
         end
       end
@@ -75,17 +77,31 @@ def start
           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 })
+              router.emit(tag_for(Opentelemetry::RECORD_TYPE_LOGS), Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_LOGS, "message" => record })
             },
             metrics: lambda { |record|
-              router.emit(@tag, Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_METRICS, "message" => record })
+              router.emit(tag_for(Opentelemetry::RECORD_TYPE_METRICS), Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_METRICS, "message" => record })
             },
             traces: lambda { |record|
-              router.emit(@tag, Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_TRACES, "message" => record })
+              router.emit(tag_for(Opentelemetry::RECORD_TYPE_TRACES), Fluent::EventTime.now, { "type" => Opentelemetry::RECORD_TYPE_TRACES, "message" => record })
             }
           )
         end
       end
     end
+
+    private
+
+    def expand_tag_placeholders(tag)
+      @expand_tag_placeholders ||= {
+        Opentelemetry::RECORD_TYPE_LOGS => tag.gsub(Opentelemetry::PLACEHOLDER_TYPE, Opentelemetry::PLACEHOLDER_TYPE_LOGS),
+        Opentelemetry::RECORD_TYPE_METRICS => tag.gsub(Opentelemetry::PLACEHOLDER_TYPE, Opentelemetry::PLACEHOLDER_TYPE_METRICS),
+        Opentelemetry::RECORD_TYPE_TRACES => tag.gsub(Opentelemetry::PLACEHOLDER_TYPE, Opentelemetry::PLACEHOLDER_TYPE_TRACES)
+      }
+    end
+
+    def tag_for(type)
+      @expand_tag_placeholders[type]
+    end
   end
 end

lib/fluent/plugin/opentelemetry/constant.rb

@@ -16,6 +16,11 @@ module Fluent::Plugin::Opentelemetry
   RECORD_TYPE_METRICS = "opentelemetry_metrics"
   RECORD_TYPE_TRACES = "opentelemetry_traces"
 
+  PLACEHOLDER_TYPE = "${type}"
+  PLACEHOLDER_TYPE_LOGS = "logs"
+  PLACEHOLDER_TYPE_METRICS = "metrics"
+  PLACEHOLDER_TYPE_TRACES = "traces"
+
   TLS_VERSIONS_MAP =
     begin
       map = {

test/fluent/plugin/test_in_opentelemetry.rb

@@ -287,6 +287,50 @@ def test_https_receive_protocol_buffers
     end
   end
 
+  sub_test_case "Placeholder" do
+    def config
+      <<~"CONFIG"
+        tag opentelemetry.${type}
+        <http>
+          bind 127.0.0.1
+          port #{@port}
+        </http>
+      CONFIG
+    end
+
+    data("metrics" => {
+           request_path: "/v1/metrics",
+           request_data: TestData::JSON::METRICS,
+           record_type: Fluent::Plugin::Opentelemetry::RECORD_TYPE_METRICS,
+           record_data: TestData::JSON::METRICS,
+           expanded_tag: "opentelemetry.metrics"
+         },
+         "traces" => {
+           request_path: "/v1/traces",
+           request_data: TestData::JSON::TRACES,
+           record_type: Fluent::Plugin::Opentelemetry::RECORD_TYPE_TRACES,
+           record_data: TestData::JSON::TRACES,
+           expanded_tag: "opentelemetry.traces"
+         },
+         "logs" => {
+           request_path: "/v1/logs",
+           request_data: TestData::JSON::LOGS,
+           record_type: Fluent::Plugin::Opentelemetry::RECORD_TYPE_LOGS,
+           record_data: TestData::JSON::LOGS,
+           expanded_tag: "opentelemetry.logs"
+         })
+    def test_type_placeholder(data)
+      d = create_driver
+      res = d.run(expect_records: 1) do
+        post_json(data[:request_path], data[:request_data])
+      end
+
+      expected_events = [[data[:expanded_tag], @event_time, { "type" => data[:record_type], "message" => data[:record_data] }]]
+      assert_equal(200, res.status)
+      assert_equal(expected_events, d.events)
+    end
+  end
+
   def compress(data)
     gz = Zlib::GzipWriter.new(StringIO.new)
     gz << data