feat: add possibility to ingest LineProtocol from Fluetd's event (#31)

Author: bednar

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Features
 1. [#30](https://github.com/influxdata/influxdb-plugin-fluent/pull/30): Field value for `time_key` can be formatted date (`2021-11-05 09:15:49.487727165 +0000`, `2021-11-05T10:04:43.617216Z`)
+1. [#31](https://github.com/influxdata/influxdb-plugin-fluent/pull/31): Add possibility to use LineProtocol from Fluetd's record.
 
 ### Dependencies
 1. [#30](https://github.com/influxdata/influxdb-plugin-fluent/pull/30): Update dependencies:

README.md

@@ -45,6 +45,7 @@ Store Fluentd event to InfluxDB 2 database.
 | field_cast_to_float | Turn on/off auto casting Integer value to Float. Helper to avoid mismatch error: 'series type mismatch: already Integer but got Float'. | bool | false |
 | time_precision | The time precision of timestamp. You should specify either second (s), millisecond (ms), microsecond (us), or nanosecond (ns). | String | ns |
 | time_key | A name of the record key that used as a 'timestamp' instead of event timestamp. If a record key doesn't exists or hasn't value then is used event timestamp. | String | nil |
+| line_protocol_key | A name of the record key that contains [LineProtocol](https://docs.influxdata.com/influxdb/v2.0/reference/syntax/line-protocol/). The value of this key is used for ingesting data into InfluxDB. If a record key doesn't exists or hasn't value then is used event timestamp. | String | nil |
 
 ##### Minimal configuration
 

lib/fluent/plugin/out_influxdb2.rb

@@ -73,6 +73,11 @@ class InfluxDBOutput < Fluent::Plugin::Output
   desc 'A name of the record key that used as a \'timestamp\' instead of event timestamp.' \
       'If a record key doesn\'t exists or hasn\'t value then is used event timestamp.'
 
+  config_param :line_protocol_key, :string, default: nil
+  desc 'A name of the record key that contains \'LineProtocol\'.' \
+      'The value of this key is used for ingesting data into InfluxDB.' \
+      'For more info see - https://docs.influxdata.com/influxdb/v2.0/reference/syntax/line-protocol/.'
+
   config_section :buffer do
     config_set_default :@type, DEFAULT_BUFFER_TYPE
     config_set_default :chunk_keys, ['tag']
@@ -121,22 +126,30 @@ def write(chunk)
     tag = chunk.metadata.tag
     bucket, measurement = expand_placeholders(chunk)
     chunk.msgpack_each do |time, record|
-      time_formatted = _format_time(time)
-      point = InfluxDB2::Point
-              .new(name: measurement)
-      record.each_pair do |k, v|
-        if k.eql?(@time_key)
-          time_formatted = _format_time(v)
-        else
-          _parse_field(k, v, point)
+      if @line_protocol_key
+        points << record[@line_protocol_key] if record.include?(@line_protocol_key)
+      else
+        time_formatted = _format_time(time)
+        point = InfluxDB2::Point
+                .new(name: measurement)
+        record.each_pair do |k, v|
+          if k.eql?(@time_key)
+            time_formatted = _format_time(v)
+          else
+            _parse_field(k, v, point)
+          end
+          point.add_tag('fluentd', tag) if @tag_fluentd
         end
-        point.add_tag('fluentd', tag) if @tag_fluentd
+        point.time(time_formatted, @precision)
+        points << point
       end
-      point.time(time_formatted, @precision)
-      points << point
     end
-    @write_api.write(data: points, bucket: bucket)
-    log.debug "Written points: #{points}"
+    if points.empty?
+      log.debug "Nothing to write for chunk: #{chunk.metadata}"
+    else
+      @write_api.write(data: points, bucket: bucket)
+      log.debug "Written points: #{points}"
+    end
   end
 
   private

test/influxdb/plugin/output_test.rb

@@ -586,4 +586,39 @@ def test_measurement_as_placeholder
     assert_requested(:post, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns',
                      times: 1, body: 'placeholder_h2o_tag level=2i,location="europe" 1444897215000000000')
   end
+
+  def test_line_protocol
+    stub_request(:any, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns')
+      .to_return(status: 204)
+    driver = create_driver(%(
+      @type influxdb2
+      token my-token
+      bucket my-bucket
+      org my-org
+      time_precision ns
+      line_protocol_key lp_key
+    ))
+    driver.run(default_tag: 'h2o_tag') do
+      emit_documents(driver, 'location' => 'europe', 'level' => 2, 'lp_key' => 'mem,tag=1 field=10 102030')
+    end
+    assert_requested(:post, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns',
+                     times: 1, body: 'mem,tag=1 field=10 102030')
+  end
+
+  def test_line_protocol_not_defined
+    stub_request(:any, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns')
+      .to_return(status: 204)
+    driver = create_driver(%(
+      @type influxdb2
+      token my-token
+      bucket my-bucket
+      org my-org
+      time_precision ns
+      line_protocol_key lp_key
+    ))
+    driver.run(default_tag: 'h2o_tag') do
+      emit_documents(driver)
+    end
+    assert_requested(:post, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns', times: 0)
+  end
 end