lib/logstash_writer.rb in logstash_writer-0.0.5 vs lib/logstash_writer.rb in logstash_writer-0.0.6

@@ -6,12 +6,12 @@
 
 # Write messages to a logstash server.
 #
 # Flings events, represented as JSON objects, to logstash using the
 # `json_lines` codec (over TCP).  Doesn't do any munging or modification of
-# the event data given to it, other than adding `@timestamp` and `_id`
-# fields if they do not already exist.
+# the event data given to it, other than adding `@timestamp` and
+# `[@metadata][_id]` fields if they do not already exist.
 #
 # We support highly-available logstash installations by means of multiple
 # address records, or via SRV records.  See the docs for .new for details
 # as to the valid formats for the server.
 #
@@ -102,12 +102,12 @@
     @worker_mutex = Mutex.new
   end
 
   # Add an event to the queue, to be sent to logstash.  Actual event
   # delivery will happen in a worker thread that is started with
-  # #run.  If the event does not have a `@timestamp` or `_id` element, they
-  # will be added with appropriate values.
+  # #run.  If the event does not have a `@timestamp` or `[@metadata][_id]`
+  # element, they will be added with appropriate values.
   #
   # @param e [Hash] the event data to be sent.
   #
   # @return [NilClass]
   #
@@ -118,15 +118,23 @@
 
     unless e.has_key?(:@timestamp) || e.has_key?("@timestamp")
       e[:@timestamp] = Time.now.utc.strftime("%FT%T.%NZ")
     end
 
-    unless e.has_key?(:_id) || e.has_key?("_id")
+    if e.has_key?("@metadata")
+      e[:@metadata] = (e[:@metadata] || {}).merge(e["@metadata"])
+    end
+
+    unless e.has_key?(:@metadata)
+      e[:@metadata] = {}
+    end
+
+    unless e[:@metadata].has_key?(:_id) || e.has_key?("_id")
       # This is the quickest way I've found to get a long, random string.
       # We don't need any sort of cryptographic or unpredictability
       # guarantees for what we're doing here, so SecureRandom is unnecessary
       # overhead.
-      e[:_id] = rand(0x1000_0000_0000_0000_0000_0000_0000_0000).to_s(36)
+      e[:@metadata][:_id] = rand(0x1000_0000_0000_0000_0000_0000_0000_0000).to_s(36)
     end
 
     @queue_mutex.synchronize do
       @queue << { content: e, arrival_timestamp: Time.now }
       while @queue.length > @backlog