Add "timestamp_key", prefer log message when merging

README.md

@@ -31,7 +31,8 @@ Configuration options for fluent.conf are:
   * json_merge - Same as json but merge content of `log_key` into the top level and strip `log_key`
 * `log_key` - Used to specify the key when merging json or sending logs in text format (default `message`)
 * `open_timeout` - Set timeout seconds to wait until connection is opened.
-* `add_timestamp` - Add `timestamp` field to logs before sending to sumologic (default `true`)
+* `add_timestamp` - Add `timestamp` (or `timestamp_key`) field to logs before sending to sumologic (default `true`)
+* `timestamp_key` - Field name when `add_timestamp` is on (default `timestamp`)
 * `proxy_uri` - Add the `uri` of the `proxy` environment if present.
 * `metric_data_format` - The format of metrics you will be sending, either `graphite` or `carbon2` (Default is `graphite `)
 * `disable_cookies` - Option to disable cookies on the HTTP Client. (Default is `false `)

fluent-plugin-sumologic_output.gemspec

@@ -20,7 +20,7 @@ Gem::Specification.new do |gem|
 
   gem.required_ruby_version = '>= 2.0.0'
 
-  gem.add_development_dependency "bundler", "~> 1.3"
+  gem.add_development_dependency "bundler", "~> 2"
   gem.add_development_dependency "rake"
   gem.add_development_dependency 'test-unit', '~> 3.1.0'
   gem.add_development_dependency "codecov", ">= 0.1.10"

lib/fluent/plugin/out_sumologic.rb

@@ -80,6 +80,7 @@ class Fluent::Plugin::Sumologic < Fluent::Plugin::Output
   config_param :delimiter, :string, :default => "."
   config_param :open_timeout, :integer, :default => 60
   config_param :add_timestamp, :bool, :default => true
+  config_param :timestamp_key, :string, :default => 'timestamp'
   config_param :proxy_uri, :string, :default => nil
   config_param :disable_cookies, :bool, :default => false
 
@@ -145,7 +146,7 @@ def merge_json(record)
       log = record[@log_key].strip
       if log[0].eql?('{') && log[-1].eql?('}')
         begin
-          record = JSON.parse(log).merge(record)
+          record = record.merge(JSON.parse(log))
           record.delete(@log_key)
         rescue JSON::ParserError
           # do nothing, ignore
@@ -255,12 +256,12 @@ def write(chunk)
           end
         when 'json_merge'
           if @add_timestamp
-            record = { :timestamp => sumo_timestamp(time) }.merge(record)
+            record = { @timestamp_key => sumo_timestamp(time) }.merge(record)
           end
           log = dump_log(merge_json(record))
         else
           if @add_timestamp
-            record = { :timestamp => sumo_timestamp(time) }.merge(record)
+            record = { @timestamp_key => sumo_timestamp(time) }.merge(record)
           end
           log = dump_log(record)
         end

test/plugin/test_out_sumologic.rb

@@ -69,6 +69,7 @@ def test_default_configure
     assert_equal instance.delimiter, '.'
     assert_equal instance.open_timeout, 60
     assert_equal instance.add_timestamp, true
+    assert_equal instance.timestamp_key, 'timestamp'
     assert_equal instance.proxy_uri, nil
     assert_equal instance.disable_cookies, false
   end
@@ -174,7 +175,28 @@ def test_emit_json_merge
     end
     assert_requested :post, "https://collectors.sumologic.com/v1/receivers/http/1234",
                      headers: {'X-Sumo-Category'=>'test', 'X-Sumo-Client'=>'fluentd-output', 'X-Sumo-Host'=>'test', 'X-Sumo-Name'=>'test'},
-                     body: /\A{"foo2":"bar2","timestamp":\d+,"foo":"bar"}\z/,
+                     body: /\A{"timestamp":\d+,"foo":"bar","foo2":"bar2"}\z/,
+                     times:1
+  end
+
+  def test_emit_json_merge_timestamp
+    config = %{
+      endpoint        https://collectors.sumologic.com/v1/receivers/http/1234
+      log_format      json_merge
+      source_category test
+      source_host     test
+      source_name     test
+
+    }
+    driver = create_driver(config)
+    time = event_time
+    stub_request(:post, 'https://collectors.sumologic.com/v1/receivers/http/1234')
+    driver.run do
+      driver.feed("output.test", time, {'message' => '{"timestamp":123}'})
+    end
+    assert_requested :post, "https://collectors.sumologic.com/v1/receivers/http/1234",
+                     headers: {'X-Sumo-Category'=>'test', 'X-Sumo-Client'=>'fluentd-output', 'X-Sumo-Host'=>'test', 'X-Sumo-Name'=>'test'},
+                     body: /\A{"timestamp":123}\z/,
                      times:1
   end
 
@@ -221,6 +243,27 @@ def test_emit_json_no_timestamp
                      times:1
   end
 
+  def test_emit_json_timestamp_key
+    config = %{
+      endpoint        https://collectors.sumologic.com/v1/receivers/http/1234
+      log_format      json
+      source_category test
+      source_host     test
+      source_name     test
+      timestamp_key   ts
+    }
+    driver = create_driver(config)
+    time = event_time
+    stub_request(:post, 'https://collectors.sumologic.com/v1/receivers/http/1234')
+    driver.run do
+      driver.feed("output.test", time, {'message' => 'test'})
+    end
+    assert_requested :post, "https://collectors.sumologic.com/v1/receivers/http/1234",
+                     headers: {'X-Sumo-Category'=>'test', 'X-Sumo-Client'=>'fluentd-output', 'X-Sumo-Host'=>'test', 'X-Sumo-Name'=>'test'},
+                     body: /\A{"ts":\d+.,"message":"test"}\z/,
+                     times:1
+  end
+
   def test_emit_graphite
     config = %{
       endpoint            https://collectors.sumologic.com/v1/receivers/http/1234