Merge pull request #10 from cosmo0920/add-arbitrary-precision-timestamp

Add an ability to specify arbitrary precision timestamp

Author: ecwws

lib/fluent/plugin/filter_elasticsearch_timestamp_check.rb

@@ -4,9 +4,13 @@ module Fluent::Plugin
   class ElasticsearchTimestampCheckFilter < Filter
     Fluent::Plugin.register_filter('elasticsearch_timestamp_check', self)
 
+    config_param :subsecond_precision, :integer, default: 3
+
     def configure(conf)
       super
       require 'date'
+      raise Fluent::ConfigError, "specify 1 or bigger number." if subsecond_precision < 1
+      @strftime_format = "%Y-%m-%dT%H:%M:%S.%#{@subsecond_precision}N%z".freeze
     end
 
     def start
@@ -32,13 +36,13 @@ def filter(tag, time, record)
             record['@timestamp'] = record['fluent_converted_timestamp'] =
               Time.at(
                 num / (10 ** ((Math.log10(num).to_i + 1) - 10))
-              ).strftime('%Y-%m-%dT%H:%M:%S.%L%z')
+              ).strftime(@strftime_format)
             break
           end
 
           # normal timestamp string processing
           record['@timestamp'] = record['fluent_converted_timestamp'] =
-            DateTime.parse(timestamp).strftime('%Y-%m-%dT%H:%M:%S.%L%z')
+            DateTime.parse(timestamp).strftime(@strftime_format)
           $log.debug("Timestamp parsed: #{record['@timestamp']}")
           break
         rescue ArgumentError
@@ -47,7 +51,7 @@ def filter(tag, time, record)
 
       unless record['fluent_converted_timestamp']
         record['@timestamp'] = record['fluent_added_timestamp'] =
-          Time.now.strftime('%Y-%m-%dT%H:%M:%S.%L%z')
+          Time.now.strftime(@strftime_format)
         $log.debug("Timestamp added: #{record['@timestamp']}")
       end
 

test/plugin/test_filter_elasticsearch_timestamp_check.rb

@@ -9,6 +9,18 @@ def create_driver(conf='')
     Fluent::Test::Driver::Filter.new(Fluent::Plugin::ElasticsearchTimestampCheckFilter).configure(conf)
   end
 
+  def test_configure
+    assert_raise(Fluent::ConfigError) do
+      create_driver(%[subsecond_precision -3])
+    end
+    assert_raise(Fluent::ConfigError) do
+      create_driver(%[subsecond_precision 0])
+    end
+    assert_nothing_raised do
+      create_driver(%[subsecond_precision 1])
+    end
+  end
+
   def test_added_timestamp
     d = create_driver
     d.run(default_tag: 'test') do
@@ -51,7 +63,29 @@ def test_timestamp_with_digit(data)
     num = timestamp.to_f
     formatted_time = Time.at(
       num / (10 ** ((Math.log10(num).to_i + 1) - 10))
-    ).strftime('%Y-%m-%dT%H:%M:%S.%L%z')
+    ).strftime('%Y-%m-%dT%H:%M:%S.%3N%z')
+    assert_true(filtered.key?("@timestamp"))
+    assert_true(filtered.key?("fluent_converted_timestamp"))
+    assert_equal(formatted_time, filtered["fluent_converted_timestamp"])
+  end
+
+  data('@timestamp'       => '@timestamp',
+       'timestamp'        => 'timestamp',
+       'time'             => 'time',
+       'syslog_timestamp' => 'syslog_timestamp')
+  def test_timestamp_with_digit_and_nano_precision(data)
+    timekey = data
+    precision = 9
+    d = create_driver(%[subsecond_precision #{precision}])
+    timestamp = '1505800348899'
+    d.run(default_tag: 'test') do
+      d.feed({'test' => 'notime'}.merge(timekey => timestamp))
+    end
+    filtered = d.filtered.map{|e| e.last}.first
+    num = timestamp.to_f
+    formatted_time = Time.at(
+      num / (10 ** ((Math.log10(num).to_i + 1) - 10))
+    ).strftime("%Y-%m-%dT%H:%M:%S.%#{precision}N%z")
     assert_true(filtered.key?("@timestamp"))
     assert_true(filtered.key?("fluent_converted_timestamp"))
     assert_equal(formatted_time, filtered["fluent_converted_timestamp"])