out_mongo.rb in fluent-plugin-mongo-1.4.0

- old
+ new

@@ -36,10 +36,14 @@
     desc "Replace dot with specified string"
     config_param :replace_dot_in_key_with, :string, default: nil
     desc "Replace dollar with specified string"
     config_param :replace_dollar_in_key_with, :string, default: nil
 
+    # Additional date field to be used to Date object
+    desc "Specify keys to use MongoDB's Date. Supported value types are Integer/Float/EventTime/String"
+    config_param :date_keys, :array, default: nil
+
     # tag mapping mode
     desc "Use tag_mapped mode"
     config_param :tag_mapped, :bool, default: false,
                  deprecated: "use '${tag}' placeholder in collection parameter."
     desc "Remove tag prefix"
@@ -52,10 +56,11 @@
     config_param :ssl_key, :string, default: nil
     config_param :ssl_key_pass_phrase, :string, default: nil, secret: true
     config_param :ssl_verify, :bool, default: false
     config_param :ssl_ca_cert, :string, default: nil
 
+
     config_section :buffer do
       config_set_default :@type, DEFAULT_BUFFER_TYPE
       config_set_default :chunk_keys, ['tag']
     end
 
@@ -196,14 +201,43 @@
     end
 
     def collect_records(chunk)
       records = []
       time_key = @inject_config.time_key if @inject_config
+      date_keys = @date_keys
+
       tag = chunk.metadata.tag
       chunk.msgpack_each {|time, record|
         record = inject_values_to_record(tag, time, record)
         # MongoDB uses BSON's Date for time.
         record[time_key] = Time.at(time || record[time_key]) if time_key
+
+        if date_keys
+          date_keys.each { |date_key|
+            begin
+              date_value = record[date_key]
+              case date_value
+              when Fluent::EventTime
+                record[date_key] = date_value.to_time
+              when Integer
+                record[date_key] = if date_value > 9999999999
+                                     # epoch with milliseconds: e.g. javascript
+                                     Time.at(date_value / 1000.0)
+                                   else
+                                     # epoch with seconds: e.g. ruby
+                                     Time.at(date_value)
+                                   end
+              when Float
+                record[date_key] = Time.at(date_value)
+              else
+                record[date_key] = Time.parse(date_value)
+              end
+            rescue ArgumentError
+              log.warn "Failed to parse '#{date_key}' field. Expected date types are Integer/Float/String/EventTime: #{record[date_key]}"
+              record[date_key] = nil
+            end
+          }
+        end
         records << record
       }
       records
     end
 