🎉 introduce :filter and :update_expressions

This commit introduces :filter (instead of :query-key and :query-value) for
defining the query that we pass down to Mongo. It is a hash and will allow us
to filter by a composite key.

In addition to that :update_expressions is now another optional hash that can
be added to *replace* the default $set operator.

The hash is a set of Mongo Update
Expressions (https://docs.mongodb.com/manual/reference/operator/update/#id1)
and the values are also substituted. Note that pipeline (Mongo >= 4.2) support
is not there yet.

Finally, action is now fully dynamic and expanded via sprintf as in
logstash-output-elasticsearch.

Author: arichiardi

.gitignore

@@ -2,3 +2,4 @@
 Gemfile.lock
 .bundle
 vendor
+logs

README.md

@@ -1,6 +1,10 @@
-# Logstash Plugin
+# Logstash Mongo Output Plugin
 
-[![Travis Build Status](https://travis-ci.org/logstash-plugins/logstash-output-mongodb.svg)](https://travis-ci.org/logstash-plugins/logstash-output-mongodb)
+---
+This is a fork of [logstash-plugins/logstash-output-mongodb](https://github.com/logstash-plugins/logstash-output-mongodb).
+
+It adds the :action, :filter, :update_expressions and :upsert parameters
+---
 
 This is a plugin for [Logstash](https://github.com/elastic/logstash).
 
@@ -21,20 +25,17 @@ Need help? Try #logstash on freenode IRC or the https://discuss.elastic.co/c/log
 
 ### 1. Plugin Developement and Testing
 
-#### Code
-- To get started, you'll need JRuby with the Bundler gem installed.
+For developing this plugin we use the wonderful work of [cameronkerrnz/logstash-plugin-dev](https://github.com/cameronkerrnz/logstash-plugin-dev):
 
-- Create a new plugin or clone and existing from the GitHub [logstash-plugins](https://github.com/logstash-plugins) organization. We also provide [example plugins](https://github.com/logstash-plugins?query=example).
+To start an interactive environment run:
 
-- Install dependencies
-```sh
-bundle install
+``` sh
+docker run --rm -it -v ${PWD}:/work cameronkerrnz/logstash-plugin-dev:7.9
 ```
 
-#### Test
-
-- Update your dependencies
+After that you can run the usual suspects:
 
+- Install/Update dependencies
 ```sh
 bundle install
 ```
@@ -95,4 +96,4 @@ Programming is not a required skill. Whatever you've seen about open source and
 
 It is more important to the community that you are able to contribute.
 
-For more information about contributing, see the [CONTRIBUTING](https://github.com/elastic/logstash/blob/master/CONTRIBUTING.md) file.
+For more information about contributing, see the [CONTRIBUTING](https://github.com/elastic/logstash/blob/master/CONTRIBUTING.md) file.

lib/logstash/outputs/mongodb.rb

@@ -34,22 +34,55 @@ class LogStash::Outputs::Mongodb < LogStash::Outputs::Base
   # "_id" field in the event.
   config :generateId, :validate => :boolean, :default => false
 
-
   # Bulk insert flag, set to true to allow bulk insertion, else it will insert events one by one.
   config :bulk, :validate => :boolean, :default => false
+
   # Bulk interval, Used to insert events periodically if the "bulk" flag is activated.
   config :bulk_interval, :validate => :number, :default => 2
+
   # Bulk events number, if the number of events to insert into a collection raise that limit, it will be bulk inserted
   # whatever the bulk interval value (mongodb hard limit is 1000).
   config :bulk_size, :validate => :number, :default => 900, :maximum => 999, :min => 2
 
-  # The method used to write processed events to MongoDB.
-  # Possible values are `insert`, `update` and `replace`.
-  config :action, :validate => :string, :required => true
-  # The key of the query to find the document to update or replace.
-  config :query_key, :validate => :string, :required => false, :default => "_id"
-  # The value of the query to find the document to update or replace. This can be dynamic using the `%{foo}` syntax.
-  config :query_value, :validate => :string, :required => false
+  # The Mongo DB action to perform. Valid actions are:
+  #
+  # - insert: inserts a document, fails if a document the document already exists.
+  # - update: updates a document given a `filter`. You can also upsert a document, see the `upsert` option.
+  # - delete: *Not Supported* at the moment
+  #
+  # A sprintf-able string is allowed to change the action based on the content
+  # of the event. The value `%{[foo]}` would use the foo field for the action.
+  #
+  # For more details on actions, check out the https://docs.mongodb.com/ruby-driver/v2.6/tutorials/ruby-driver-bulk-operations[Mongo Ruby Driver documentation]
+  config :action, :validate => :string, :default => "insert"
+
+  # The :filter clause for an update or replace.
+  #
+  # A sprintf-able string is allowed for keys: the value `my_%{[foo]}` would
+  # use the foo field instead *always coerced to a string*.
+  #
+  # Hovewever, the
+  # https://www.elastic.co/guide/en/logstash/current/field-references-deepdive.html[Field
+  # Reference Syntax] is required for values - these preserve type (integer,
+  # float, ...).
+  config :filter, :validate => :hash, :required => false, :default => {}
+
+  # The hash in :update_expressions will be used *instead* of the default
+  # '$set'. This option is useful for using alternative operators like '$inc'.
+  #
+  # A sprintf-able string is allowed for keys: the value `my_%{[foo]}` would
+  # use the foo field instead *always coerced to a string*.
+  #
+  # Hovewever, the
+  # https://www.elastic.co/guide/en/logstash/current/field-references-deepdive.html[Field
+  # Reference Syntax] is required for values - these preserve type (integer,
+  # float, ...).
+  #
+  # Keys must start with `$`, see the https://docs.mongodb.com/manual/reference/operator/update/#id1[Mongo DB Update Operators] for reference.
+  #
+  # Note that pipeline support (Mongo >= 4.2) is not there yet.
+  config :update_expressions, :validate => :hash, :required => false, :default => nil
+
   # If true, a new document is created if no document exists in DB with given `document_id`.
   # Only applies if action is `update` or `replace`.
   config :upsert, :validate => :boolean, :required => false, :default => false
@@ -88,20 +121,37 @@ def validate_config
     if @bulk_size > 1000
       raise LogStash::ConfigurationError, "Bulk size must be lower than '1000', currently '#{@bulk_size}'"
     end
-    if @action != "insert" && @action != "update" && @action != "replace"
-      raise LogStash::ConfigurationError, "Only insert, update and replace are valid for 'action' setting."
+    if !@update_expressions.nil?
+      @update_expressions.keys.each { |k|
+        if !is_update_operator(k)
+          raise LogStash::ConfigurationError, "The :update_expressions option contains '#{k}', which is not an Update expression."
+          break
+        end
+      }
     end
-    if (@action == "update" || @action == "replace") && (@query_value.nil? || @query_value.empty?)
-      raise LogStash::ConfigurationError, "If action is update or replace, query_value must be set."
+  end
+
+  def validate_action(action, filter, update_expressions)
+    if action != "insert" && action != "update" && action != "replace"
+      raise LogStash::ConfigurationError, "Only insert, update and replace are supported Mongo actions, got '#{action}'."
+    end
+    if (action == "update" || action == "replace") && (filter.nil? || filter.empty?)
+      raise LogStash::ConfigurationError, "If action is update or replace, filter must be set."
+    end
+    if action != "update" && !(update_expressions.nil? || update_expressions.empty?)
+      raise LogStash::ConfigurationError, "The :update_expressions only makes sense if the action is an update."
     end
   end
 
   def receive(event)
+    action = event.sprintf(@action)
+
+    validate_action(action, @filter, @update_expressions)
+
     begin
       # Our timestamp object now has a to_bson method, using it here
       # {}.merge(other) so we don't taint the event hash innards
       document = {}.merge(event.to_hash)
-
       if !@isodate
         timestamp = event.timestamp
         if timestamp
@@ -117,9 +167,19 @@ def receive(event)
       end
 
       collection = event.sprintf(@collection)
-      if @action == "update" or @action == "replace"
-        document["metadata_mongodb_output_query_value"] = event.sprintf(@query_value)
+      if action == "update" or action == "replace"
+        document["metadata_mongodb_output_filter"] = apply_event_to_hash(event, @filter)
       end
+
+      if action == "update" and !(@update_expressions.nil? || @update_expressions.empty?)
+        # we only expand the values cause keys are update expressions
+        expressions_hash = {}
+        @update_expressions.each do |k, v|
+          expressions_hash[k] = apply_event_to_hash(event, v)
+        end
+        document["metadata_mongodb_output_update_expressions"] = expressions_hash
+      end
+
       if @bulk
         @@mutex.synchronize do
           if(!@documents[collection])
@@ -133,7 +193,8 @@ def receive(event)
           end
         end
       else
-        write_to_mongodb(collection, [document])
+        result = write_to_mongodb(collection, [document])
+        @logger.debug("Bulk write result", :result => result)
       end
     rescue => e
       if e.message =~ /^E11000/
@@ -144,7 +205,7 @@ def receive(event)
         # to fix the issue.
         @logger.warn("Skipping insert because of a duplicate key error", :event => event, :exception => e)
       else
-        @logger.warn("Failed to send event to MongoDB, retrying in #{@retry_delay.to_s} seconds", :event => event, :exception => e)
+        @logger.warn("Failed to send event to MongoDB retrying in #{@retry_delay.to_s} seconds", :result=> e.result, :message => e.message)
         sleep(@retry_delay)
         retry
       end
@@ -153,25 +214,61 @@ def receive(event)
 
   def write_to_mongodb(collection, documents)
     ops = get_write_ops(documents)
+    @logger.debug("Sending", :ops => ops)
     @db[collection].bulk_write(ops)
   end
 
   def get_write_ops(documents)
     ops = []
     documents.each do |doc|
-      replaced_query_value = doc["metadata_mongodb_output_query_value"]
-      doc.delete("metadata_mongodb_output_query_value")
-      if @action == "insert"
+      filter = doc["metadata_mongodb_output_filter"]
+      doc.delete("metadata_mongodb_output_filter")
+
+      update_expressions = doc["metadata_mongodb_output_update_expressions"]
+      doc.delete("metadata_mongodb_output_update_expressions")
+
+      # TODO: support multiple expressions as pipeline for Mongo >= 4.2
+      update = if !update_expressions.nil?
+                 update_expressions
+               else
+                 {'$set' => to_dotted_hash(doc)}
+               end
+
+      if action == "insert"
         ops << {:insert_one => doc}
-      elsif @action == "update"
-        ops << {:update_one => {:filter => {@query_key => replaced_query_value}, :update => {'$set' => to_dotted_hash(doc)}, :upsert => @upsert}}
-      elsif @action == "replace"
-        ops << {:replace_one => {:filter => {@query_key => replaced_query_value}, :replacement => doc, :upsert => @upsert}}
+      elsif action == "update"
+        ops << {:update_one => {:filter => filter, :update => update, :upsert => @upsert}}
+      elsif action == "replace"
+        ops << {:replace_one => {:filter => filter, :replacement => doc, :upsert => @upsert}}
       end
     end
     ops
   end
 
+  def is_update_operator(string)
+    string.start_with?("$")
+  end
+
+  # Apply the event to the input hash keys and values.
+  #
+  # This function is recursive.
+  #
+  # It uses event.sprintf for keys but event.get for values because it looks
+  # like event.sprintf always returns a string and we don't want to always
+  # coerce.
+  #
+  # See  https://github.com/elastic/logstash/issues/5114
+  def apply_event_to_hash(event, hash)
+    hash.clone.each_with_object({}) do |(k, v), ret|
+      if v.is_a? Hash
+        ret[event.sprintf(k)] = apply_event_to_hash(event, v)
+      else
+        event_value = event.get(v)
+        ret[event.sprintf(k)] = event_value.nil? ? v : event_value
+      end
+    end
+  end
+
   def to_dotted_hash(hash, recursive_key = "")
     hash.each_with_object({}) do |(k, v), ret|
       key = recursive_key + k.to_s

logstash-output-mongodb.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-mongodb'
-  s.version         = '3.2.0'
+  s.version         = '3.2.1'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Writes events to MongoDB"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/integration/mongodb_spec.rb

@@ -6,8 +6,7 @@
   let(:uri)        { 'mongodb://localhost:27017' }
   let(:database)   { 'logstash' }
   let(:collection) { 'logs' }
-  let(:uuid)       { SecureRandom.uuid }
-  let(:action) { 'insert' }
+  let(:action)     { 'insert' }
 
   let(:config) do
     { "uri" => uri, "database" => database,
@@ -19,8 +18,10 @@
     subject { LogStash::Outputs::Mongodb.new(config) }
 
     let(:properties) { { "message" => "This is a message!",
-                         "uuid" => uuid, "number" => BigDecimal.new("4321.1234"),
-                         "utf8" => "żółć", "int" => 42,
+                         "uuid" => "00000000-0000-0000-0000-000000000000",
+                         "number" => BigDecimal.new("4321.1234"),
+                         "utf8" => "żółć",
+                         "int" => 42,
                          "arry" => [42, "string", 4321.1234]} }
     let(:event)      { LogStash::Event.new(properties) }