feat: add placeholder support for bucket & measurement fields (#26)

Author: HongChenTW

CHANGELOG.md

@@ -1,12 +1,15 @@
 ## 1.8.0 [unreleased]
 
+### Features
+1. [#26](https://github.com/influxdata/influxdb-plugin-fluent/pull/26): Add placeholder support for bucket & measurement fields
+
 ## 1.7.0 [2021-03-05]
 
 ### Features
-1. [#23](https://github.com/influxdata/influxdb-plugin-fluent/pull/23): Added possibility to specify the certification verification behaviour
+1. [#23](https://github.com/influxdata/influxdb-plugin-fluent/pull/23): Add possibility to specify the certification verification behaviour
 
 ### CI
-1. [#24](https://github.com/influxdata/influxdb-plugin-fluent/pull/24): Updated stable image to `influxdb:latest` and nightly to `quay.io/influxdb/influxdb:nightly`
+1. [#24](https://github.com/influxdata/influxdb-plugin-fluent/pull/24): Updat stable image to `influxdb:latest` and nightly to `quay.io/influxdb/influxdb:nightly`
 
 ## 1.6.0 [2020-10-02]
 
@@ -18,15 +21,15 @@
 
 ## 1.5.0 [2020-07-17]
 
-1. [#12](https://github.com/influxdata/influxdb-plugin-fluent/pull/12): Renamed gem to `fluent-plugin-influxdb-v2`
+1. [#12](https://github.com/influxdata/influxdb-plugin-fluent/pull/12): Rename gem to `fluent-plugin-influxdb-v2`
 
 ### Dependencies
 1. [#11](https://github.com/influxdata/influxdb-plugin-fluent/pull/11): Upgrade InfluxDB client to 1.6.0
 
 ## 1.4.0 [2020-06-19]
 
 ### Features
-1. [#8](https://github.com/influxdata/influxdb-plugin-fluent/pull/8): Added support for nested fields
+1. [#8](https://github.com/influxdata/influxdb-plugin-fluent/pull/8): Add support for nested fields
 
 ### Dependencies
 1. [#9](https://github.com/influxdata/influxdb-plugin-fluent/pull/9): Upgrade InfluxDB client to 1.5.0

lib/fluent/plugin/out_influxdb2.rb

@@ -119,7 +119,7 @@ def multi_workers_ready?
   def write(chunk)
     points = []
     tag = chunk.metadata.tag
-    measurement = @measurement || tag
+    bucket, measurement = expand_placeholders(chunk)
     chunk.msgpack_each do |time, record|
       if time.is_a?(Integer)
         time_formatted = time
@@ -141,12 +141,22 @@ def write(chunk)
       point.time(time_formatted, @precision)
       points << point
     end
-    @write_api.write(data: points)
+    @write_api.write(data: points, bucket: bucket)
     log.debug "Written points: #{points}"
   end
 
   private
 
+  def expand_placeholders(chunk)
+    bucket = extract_placeholders(@bucket, chunk)
+    measurement = if @measurement
+                    extract_placeholders(@measurement, chunk)
+                  else
+                    chunk.metadata.tag
+                  end
+    [bucket, measurement]
+  end
+
   def _parse_field(key, value, point)
     if @tag_keys.include?(key)
       point.add_tag(key, value)

test/influxdb/plugin/output_test.rb

@@ -475,4 +475,39 @@ def emit_documents(driver, data = { 'location' => 'europe', 'level' => 2 })
     driver.feed(time, data)
     time
   end
+
+  def test_bucket_as_placeholder
+    stub_request(:any, 'https://localhost:8086/api/v2/write?bucket=placeholder_h2o_tag&org=my-org&precision=ns')
+      .to_return(status: 204)
+    driver = create_driver(%(
+      @type influxdb2
+      token my-token
+      bucket placeholder_${tag}
+      org my-org
+      time_precision ns
+    ))
+    driver.run(default_tag: 'h2o_tag') do
+      emit_documents(driver)
+    end
+    assert_requested(:post, 'https://localhost:8086/api/v2/write?bucket=placeholder_h2o_tag&org=my-org&precision=ns',
+                     times: 1, body: 'h2o_tag level=2i,location="europe" 1444897215000000000')
+  end
+
+  def test_measurement_as_placeholder
+    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
+      measurement placeholder_${tag}
+    ))
+    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: 1, body: 'placeholder_h2o_tag level=2i,location="europe" 1444897215000000000')
+  end
 end