Merge pull request #52 from zookzook/command-monitoring

Command monitoring

Author: zookzook

.gitignore

@@ -10,3 +10,4 @@ plt_core_path/*.plt
 .idea/
 *.iml
 *.code-workspace
+data/

.travis.yml

@@ -1,27 +1,27 @@
+dist: bionic
 sudo: required
 language: elixir
 
-elixir:
-  - 1.8
-  - 1.9
+elixir: 1.10.2
+otp_release: 22.0
+
+addons:
+  apt:
+    packages:
+      - python3
+      - python3-pip
 
 install:
-  - wget http://fastdl.mongodb.org/linux/mongodb-linux-x86_64-${MONGODB}.tgz
+  - wget https://fastdl.mongodb.org/linux/mongodb-linux-x86_64-ubuntu1804-4.4.0-rc0.tgz
   - mkdir ${PWD}/mongodb
-  - tar xzf mongodb-linux-x86_64-${MONGODB}.tgz --strip-components=1 -C ${PWD}/mongodb
+  - tar xzf mongodb-linux-x86_64-ubuntu1804-4.4.0-rc0.tgz --strip-components=1 -C ${PWD}/mongodb
 
 before_script:
+  - pyenv global 3.6
+  - pip3 install --upgrade pip
+  - pip3 install mtools[all]
   - export PATH=${PWD}/mongodb/bin/:$PATH
-  - bash ./start_mongo.bash
-  - mkdir ${PWD}/mongodb/data
-  - ${PWD}/mongodb/bin/mongod --dbpath ${PWD}/mongodb/data --logpath ${PWD}/mongodb/mongodb.log --fork
-
-env:
-  matrix:
-    - MONGODB=3.2.20
-    - MONGODB=3.4.16
-    - MONGODB=3.6.6
-    - MONGODB=ubuntu1604-v4.2-latest
+  - mlaunch init --setParameter enableTestCommands=1 --replicaset --name "rs_1"
 
 script:
   - mix local.hex --force

CHANGELOG.md

@@ -1,3 +1,12 @@
+## 0.7.0
+
+* Enhancements
+    * refactored event notification system
+    * added support for retryable reads and writes
+    * refactored the test cases
+    * now using mtools for a MongoDB deployment in the travis ci environment
+    * travis ci uses only the latest MongoDB version [The failCommand](https://github.com/mongodb/mongo/wiki/The-%22failCommand%22-fail-point) 
+    
 ## 0.6.5
 
 * Enhancements

README.md

@@ -8,7 +8,7 @@
 
 ## Features
 
-  * Supports MongoDB versions 3.2, 3.4, 3.6, 4.0, 4.2
+  * Supports MongoDB versions 3.2, 3.4, 3.6, 4.0, 4.2, 4.4
   * Connection pooling ([through DBConnection 2.x](https://github.com/elixir-ecto/db_connection))
   * Streaming cursors
   * Performant ObjectID generation
@@ -20,6 +20,9 @@
   * Support for bulk writes ([See](https://github.com/mongodb/specifications/blob/master/source/crud/crud.rst#write))
   * support for driver sessions ([See](https://github.com/mongodb/specifications/blob/master/source/sessions/driver-sessions.rst))
   * support for driver transactions ([See](https://github.com/mongodb/specifications/blob/master/source/transactions/transactions.rst))
+  * support for command monitoring ([See](https://github.com/mongodb/specifications/blob/master/source/command-monitoring/command-monitoring.rst))
+  * support for retryable reads ([See](https://github.com/mongodb/specifications/blob/master/source/retryable-reads/retryable-reads.rst))
+  * support for retryable writes ([See](https://github.com/mongodb/specifications/blob/master/source/retryable-writes/retryable-writes.rst))
 
 ## Data representation
 
@@ -368,7 +371,21 @@ Using `$in`
 Mongo.find(:mongo, "users", %{email: %{"$in" => ["[email protected]", "[email protected]"]}})
 ```
 
-## Testing
+## Testing and Travis CI
+
+The `travis.yml` file uses only the latest MongoDB. It creates a replica set of three nodes and disables the SSL test case. If you want to
+run the test cases against other MongoDB deployments or older versions, you can use the [mtools](https://github.com/rueckstiess/mtools) for deployment and run the test cases locally:
+
+### Example
+
+```bash
+pyenv global 3.6
+pip3 install --upgrade pip
+pip3 install mtools[all]
+export PATH=to-your-mongodb/bin/:$PATH
+mlaunch init --setParameter enableTestCommands=1 --replicaset --name "rs_1"
+mix test --exclude ssl --exclude socket
+``` 
 
 The SSL test suite is enabled by default. You have two options. Either exclude
 the SSL tests or enable SSL on your Mongo server.

lib/mongo.ex

@@ -50,8 +50,6 @@ defmodule Mongo do
   import Mongo.Utils
   import Mongo.WriteConcern
 
-  require Logger
-
   use Bitwise
   use Mongo.Messages
   alias Mongo.Query
@@ -60,6 +58,10 @@ defmodule Mongo do
   alias Mongo.UrlParser
   alias Mongo.Session
   alias Mongo.ReadPreference
+  alias Mongo.Events
+  alias Mongo.Events.CommandSucceededEvent
+  alias Mongo.Events.CommandFailedEvent
+  alias Mongo.Error
 
   @timeout 15000 # 5000
 
@@ -379,26 +381,52 @@ defmodule Mongo do
 
   end
 
+  def admin_command(topology_pid, cmd) do
+    with {:ok, doc} <- issue_command(topology_pid, cmd, :write, database: "admin", retryable_writes: false) do
+      {:ok, doc}
+    end
+  end
+
   @doc """
   This function is very fundamental.
   """
-  def issue_command(topology_pid, cmd, type, opts) do
+  def issue_command(topology_pid, cmd, :read, opts) do
 
-    new_cmd = case type do
-      :read  -> ReadPreference.add_read_preference(cmd, opts)
-      :write -> cmd
-    end
+    new_cmd = ReadPreference.add_read_preference(cmd, opts)
 
-    Logger.debug("issue_command: #{inspect type} #{inspect new_cmd}")
+    ## check, if retryable reads are enabled
+    opts = Mongo.retryable_reads(opts)
 
-    with {:ok, session} <- Session.start_implicit_session(topology_pid, type, opts),
+    with {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
          result <- exec_command_session(session, new_cmd, opts),
          :ok <- Session.end_implict_session(topology_pid, session) do
+      case result do
+        {:error, error} ->
+          case Error.should_retry_read(error, cmd, opts) do
+            true  -> issue_command(topology_pid, cmd, :read, Keyword.put(opts, :read_counter, 2))
+            false -> {:error, error}
+          end
+        _other        -> result
+      end
+    else
+      {:new_connection, _server} ->
+        :timer.sleep(1000)
+        issue_command(topology_pid, cmd, :read, opts)
+    end
+  end
+  def issue_command(topology_pid, cmd, :write, opts) do
+
+    ## check, if retryable reads are enabled
+    opts = Mongo.retryable_writes(opts, acknowledged?(cmd[:writeConcerns]))
+
+    with {:ok, session} <- Session.start_implicit_session(topology_pid, :write, opts),
+         result         <- exec_command_session(session, cmd, opts),
+         :ok            <- Session.end_implict_session(topology_pid, session) do
       result
     else
       {:new_connection, _server} ->
         :timer.sleep(1000)
-        issue_command(topology_pid, cmd, type, opts)
+        issue_command(topology_pid, cmd, :write, opts)
     end
   end
 
@@ -688,33 +716,61 @@ defmodule Mongo do
   @doc false
   @spec exec_command_session(GenServer.server, BSON.document, Keyword.t) :: {:ok, BSON.document | nil} | {:error, Mongo.Error.t}
   def exec_command_session(session, cmd, opts) do
-
-    Logger.debug("Executing cmd with session: #{inspect cmd}")
-
-    with {:ok, conn, cmd} <- Session.bind_session(session, cmd),
-         {:ok, _cmd, doc} <- DBConnection.execute(conn, %Query{action: :command}, [cmd], defaults(opts)),
-         doc              <- Session.update_session(session, doc, opts),
-         {:ok, doc}       <- check_for_error(doc) do
+    with {:ok, conn, new_cmd}      <- Session.bind_session(session, cmd),
+         {:ok, _cmd, {doc, event}} <- DBConnection.execute(conn, %Query{action: :command}, [new_cmd], defaults(opts)),
+         doc                       <- Session.update_session(session, doc, opts),
+         {:ok, doc}                <- check_for_error(doc, event) do
       {:ok, doc}
+    else
+      {:error, error} ->
+      ## todo update Topology
+        case Error.should_retry_write(error, cmd, opts) do
+          true  ->
+          with :ok <- Session.select_server(session, opts) do
+           exec_command_session(session, cmd, Keyword.put(opts, :write_counter, 2))
+          end
+          false -> {:error, error}
+        end
     end
 
   end
 
   @doc false
   @spec exec_command(GenServer.server, BSON.document, Keyword.t) :: {:ok, BSON.document | nil} | {:error, Mongo.Error.t}
   def exec_command(conn, cmd, opts) do
-
-    Logger.debug("Executing cmd: #{inspect cmd}")
-
-    with {:ok, _cmd, doc} <- DBConnection.execute(conn, %Query{action: :command}, [cmd], defaults(opts)),
-         {:ok, doc} <- check_for_error(doc) do
+    with {:ok, _cmd, {doc, event}} <- DBConnection.execute(conn, %Query{action: :command}, [cmd], defaults(opts)),
+         {:ok, doc} <- check_for_error(doc, event) do
       {:ok, doc}
     end
 
   end
 
-  defp check_for_error(%{"ok" => ok} = response) when ok == 1, do: {:ok, response}
-  defp check_for_error(doc), do: {:error, Mongo.Error.exception(doc)}
+  defp check_for_error(%{"ok" => ok} = response, {event, duration}) when ok == 1 do
+    Events.notify(%CommandSucceededEvent{
+      reply: response,
+      duration: duration,
+      command_name: event.command_name,
+      request_id: event.request_id,
+      operation_id: event.operation_id,
+      connection_id: event.connection_id
+    }, :commands)
+    {:ok, response}
+  end
+  defp check_for_error(doc, {event, duration}) do
+
+    error = Mongo.Error.exception(doc)
+
+    Events.notify(%CommandFailedEvent{
+      failure: error,
+      duration: duration,
+      command_name: event.command_name,
+      request_id: event.request_id,
+      operation_id: event.operation_id,
+      connection_id: event.connection_id
+    }, :commands)
+
+    {:error, error}
+  end
 
   @doc """
   Returns the wire version of the database
@@ -1149,6 +1205,22 @@ defmodule Mongo do
     end
   end
 
+  @doc """
+  Convenient function that drops the database `name`.
+  """
+  @spec drop_database(GenServer.server, String.t) :: :ok | {:error, Mongo.Error.t}
+  def drop_database(topology_pid, name \\ nil)
+  def drop_database(topology_pid, nil) do
+    with {:ok, _} <- Mongo.issue_command(topology_pid, [dropDatabase: 1], :write, []) do
+      :ok
+    end
+  end
+  def drop_database(topology_pid, name) do
+    with {:ok, _} <- Mongo.issue_command(topology_pid, [dropDatabase: 1], :write, [database: name]) do
+      :ok
+    end
+  end
+
   @doc """
   Getting Collection Names
   """
@@ -1168,6 +1240,63 @@ defmodule Mongo do
     |> Stream.map(fn coll -> coll["name"] end)
   end
 
+  @doc """
+  In case of retryable reads are enabled, the keyword `:read_counter` is added with the value of 1.
+
+  In other cases like
+
+  * `:retryable_reads` is false or nil
+  * `:session` is nil
+  * `:read_counter` is nil
+
+  the `opts` is unchanged
+
+  ## Example
+
+    iex> Mongo.retryable_reads([retryable_reads: true])
+    [retryable_reads: true, read_counter: 1]
+
+  """
+  def retryable_reads(opts) do
+    case opts[:read_counter] do
+      nil -> case opts[:retryable_reads] == true && opts[:session] == nil do
+              true -> opts ++ [read_counter: 1]
+              false -> opts
+             end
+      _other -> opts
+    end
+  end
+
+  @doc """
+  In case of retryable writes are enabled, the keyword `:write_counter` is added with the value of 1.
+
+  In other cases like
+
+  * `:retryable_writes` is false or nil
+  * `:session` is nil
+  * `:write_counter` is nil
+
+  the `opts` is unchanged
+
+  ## Example
+
+    iex> Mongo.retryable_writes([retryable_writes: true], true)
+    [retryable_writes: true, write_counter: 1]
+
+  """
+  def retryable_writes(opts, true) do
+    case opts[:write_counter] do
+      nil -> case Keyword.get(opts, :retryable_writes, true) == true && opts[:session] == nil do
+               true  -> opts ++ [write_counter: 1]
+               false -> opts
+             end
+      _other -> opts
+    end
+  end
+  def retryable_writes(opts, false) do
+    Keyword.put(opts, :retryable_writes, false)
+  end
+
   defp get_stream(topology_pid, cmd, opts) do
     Mongo.Stream.new(topology_pid, cmd, opts)
   end

lib/mongo/app.ex

@@ -7,7 +7,7 @@ defmodule Mongo.App do
     children = [
       worker(Mongo.IdServer, []),
       worker(Mongo.PBKDF2Cache, []),
-      worker(:gen_event, [local: Mongo.Events])
+      supervisor(Registry, [:duplicate, :events_registry])
     ]
 
     opts = [strategy: :one_for_one, name: Mongo.Supervisor]