Transform all time object from RubyTime to LogStash::Timestamp when reading from a DB

Fixes #83

Author: Pere Urbon-Bayes

lib/logstash/plugin_mixins/jdbc.rb

@@ -143,14 +143,12 @@ def execute_statement(statement, parameters)
       if @jdbc_paging_enabled
         query.each_page(@jdbc_page_size) do |paged_dataset|
           paged_dataset.each do |row|
-            #Stringify row keys
-            yield Hash[row.map { |k, v| [k.to_s, v] }]
+            yield extract_values_from(row)
           end
         end
       else
         query.each do |row|
-          #Stringify row keys
-          yield Hash[row.map { |k, v| [k.to_s, v] }]
+          yield extract_values_from(row)
         end
       end
       success = true
@@ -173,4 +171,20 @@ def symbolized_params(parameters)
       hash
     end
   end
+
+  private
+  #Stringify row keys and decorate values when necessary
+  def extract_values_from(row)
+    Hash[row.map { |k, v| [k.to_s, decorate_value(v)] }]
+  end
+
+  private
+  def decorate_value(value)
+    if value.is_a?(Time)
+      # transform it to LogStash::Timestamp as required by LS
+      LogStash::Timestamp.new(value)
+    else
+      value  # no-op
+    end
+  end
 end

spec/inputs/jdbc_spec.rb

@@ -19,7 +19,8 @@
     Jdbc::Derby.load_driver
     db.create_table :test_table do
       DateTime :created_at
-      Integer :num
+      Integer  :num
+      DateTime :custom_time
     end
   end
 
@@ -96,7 +97,7 @@
     let(:settings) do
       {
         "statement" => "SELECT :num_param as num_param FROM SYSIBM.SYSDUMMY1",
-        "parameters" => {"num_param" => 10}
+        "parameters" => { "num_param" => 10}
       }
     end
 
@@ -159,7 +160,7 @@
 
     it "should fetch all rows" do
       num_rows.times do
-        db[:test_table].insert(:num => 1, :created_at => Time.now.utc)
+        db[:test_table].insert(:num => 1, :custom_time => Time.now.utc, :created_at => Time.now.utc)
       end
 
       plugin.run(queue)
@@ -169,6 +170,35 @@
 
   end
 
+  context "when fetching time data" do
+
+    let(:settings) do
+      {
+        "statement" => "SELECT * from test_table",
+      }
+    end
+
+    let(:num_rows) { 10 }
+
+    before do
+      num_rows.times do
+        db[:test_table].insert(:num => 1, :custom_time => Time.now.utc, :created_at => Time.now.utc)
+      end
+
+      plugin.register
+    end
+
+    after do
+      plugin.stop
+    end
+
+    it "should convert it to LogStash::Timestamp " do
+      plugin.run(queue)
+      event = queue.pop
+      expect(event["custom_time"]).to be_a(LogStash::Timestamp)
+    end
+  end
+
   context "when iteratively running plugin#run" do
     let(:settings) do
       {"statement" => "SELECT num, created_at FROM test_table WHERE created_at > :sql_last_start"}
@@ -289,7 +319,7 @@
 
     before do
       num_rows.times do
-        db[:test_table].insert(:num => 1, :created_at => Time.now.utc)
+        db[:test_table].insert(:num => 1, :custom_time => Time.now.utc, :created_at => Time.now.utc)
       end
 
       plugin.register