feat: field value for time_key can be formatted date (#30)

Author: bednar

.circleci/config.yml

@@ -41,9 +41,9 @@ commands:
       - restore_cache:
           name: Restoring Gem Cache
           keys:
-            - &cache-key gem-cache-v2-{{ checksum "fluent-plugin-influxdb-v2.gemspec" }}-<< parameters.ruby-image >>
-            - gem-cache-v2-{{ checksum "fluent-plugin-influxdb-v2.gemspec" }}
-            - gem-cache-v2-
+            - &cache-key gem-cache-v3-{{ checksum "fluent-plugin-influxdb-v2.gemspec" }}-<< parameters.ruby-image >>
+            - gem-cache-v3-{{ checksum "fluent-plugin-influxdb-v2.gemspec" }}
+            - gem-cache-v3-
       - run:
           name: Install dependencies
           command: |
@@ -120,6 +120,9 @@ workflows:
       - tests-ruby:
           name: ruby-2.7
           ruby-image: "cimg/ruby:2.7"
+      - tests-ruby:
+          name: ruby-3.0
+          ruby-image: "cimg/ruby:3.0"
       - tests-ruby:
           name: ruby-2.6
       - tests-ruby:

CHANGELOG.md

@@ -1,7 +1,15 @@
 ## 1.9.0 [unreleased]
 
+### 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`)
+
+### Dependencies
+1. [#30](https://github.com/influxdata/influxdb-plugin-fluent/pull/30): Update dependencies:
+    - influxdb-client to 2.1.0
+
 ### CI
 1. [#27](https://github.com/influxdata/influxdb-plugin-fluent/pull/27): Switch to next-gen CircleCI's convenience images
+1. [#30](https://github.com/influxdata/influxdb-plugin-fluent/pull/30): Add Ruby 3.0 into CI
 
 ## 1.8.0 [2021-08-20]
 

fluent-plugin-influxdb-v2.gemspec

@@ -47,7 +47,7 @@ Gem::Specification.new do |spec|
   spec.required_ruby_version = '>= 2.2.0'
 
   spec.add_runtime_dependency 'fluentd', '~> 1.8'
-  spec.add_runtime_dependency 'influxdb-client', '1.12.0'
+  spec.add_runtime_dependency 'influxdb-client', '2.1.0'
 
   spec.add_development_dependency 'bundler', '~> 2.0'
   spec.add_development_dependency 'codecov', '~> 0.1.16'

lib/fluent/plugin/out_influxdb2.rb

@@ -121,18 +121,12 @@ def write(chunk)
     tag = chunk.metadata.tag
     bucket, measurement = expand_placeholders(chunk)
     chunk.msgpack_each do |time, record|
-      if time.is_a?(Integer)
-        time_formatted = time
-      else
-        nano_seconds = time.sec * 1e9
-        nano_seconds += time.nsec
-        time_formatted = @precision_formatter.call(nano_seconds)
-      end
+      time_formatted = _format_time(time)
       point = InfluxDB2::Point
               .new(name: measurement)
       record.each_pair do |k, v|
         if k.eql?(@time_key)
-          time_formatted = v
+          time_formatted = _format_time(v)
         else
           _parse_field(k, v, point)
         end
@@ -157,6 +151,27 @@ def expand_placeholders(chunk)
     [bucket, measurement]
   end
 
+  def _format_time(time)
+    if time.is_a?(Integer)
+      time
+    elsif time.is_a?(Float)
+      time
+    elsif time.is_a?(Fluent::EventTime)
+      nano_seconds = time.sec * 1e9
+      nano_seconds += time.nsec
+      @precision_formatter.call(nano_seconds)
+    elsif time.is_a?(String)
+      begin
+        _format_time(Fluent::EventTime.parse(time))
+      rescue StandardError => e
+        log.debug "Cannot parse timestamp: #{time} due: #{e}"
+        time
+      end
+    else
+      time
+    end
+  end
+
   def _parse_field(key, value, point)
     if @tag_keys.include?(key)
       point.add_tag(key, value)

test/.rubocop.yml

@@ -0,0 +1,26 @@
+#
+# The MIT License
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+#
+
+inherit_from: ../.rubocop.yml
+Metrics/ClassLength:
+  Enabled: false
+

test/influxdb/plugin/output_test.rb

@@ -381,6 +381,82 @@ def test_time_key
                      times: 1, body: 'h2o_tag,version=v.10 level=2i,location="europe" 1544897215000000000')
   end
 
+  def test_time_float
+    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
+      tag_keys ["version"]
+      time_key time
+    ))
+    driver.run(default_tag: 'h2o_tag') do
+      emit_documents(driver, 'location' => 'europe', 'level' => 2, 'version' => 'v.10',
+                             'time' => 1_544_897_215_000_000_000.123)
+    end
+    assert_requested(:post, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns',
+                     times: 1, body: 'h2o_tag,version=v.10 level=2i,location="europe" 1544897215000000000')
+  end
+
+  def test_time_key_rfc_3339_format
+    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
+      tag_keys ["version"]
+      time_key time
+    ))
+    driver.run(default_tag: 'h2o_tag') do
+      emit_documents(driver, 'location' => 'europe', 'level' => 2, 'version' => 'v.10',
+                             'time' => '2021-11-05T10:04:43.617216Z')
+    end
+    assert_requested(:post, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns',
+                     times: 1, body: 'h2o_tag,version=v.10 level=2i,location="europe" 1636106683617216000')
+  end
+
+  def test_time_key_log_format
+    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
+      tag_keys ["version"]
+      time_key time
+    ))
+    driver.run(default_tag: 'h2o_tag') do
+      emit_documents(driver, 'location' => 'europe', 'level' => 2, 'version' => 'v.10',
+                             'time' => '2021-11-05 09:15:49.487727165 +0000')
+    end
+    assert_requested(:post, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns',
+                     times: 1, body: 'h2o_tag,version=v.10 level=2i,location="europe" 1636103749487727104')
+  end
+
+  def test_time_key_not_parseable
+    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
+      tag_keys ["version"]
+      time_key time
+    ))
+    driver.run(default_tag: 'h2o_tag') do
+      emit_documents(driver, 'location' => 'europe', 'level' => 2, 'version' => 'v.10',
+                             'time' => '1544897215000000000')
+    end
+    assert_requested(:post, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns',
+                     times: 1, body: 'h2o_tag,version=v.10 level=2i,location="europe" 1544897215000000000')
+  end
+
   def test_field_cast_to_float
     stub_request(:any, 'https://localhost:8086/api/v2/write?bucket=my-bucket&org=my-org&precision=ns')
       .to_return(status: 204)