#106: work in progress: better error handling if monitor lost the connection

Author: zookzook

insights/config/dev.exs

@@ -55,7 +55,9 @@ config :insights, InsightsWeb.Endpoint,
   ]
 
 # Do not include metadata nor timestamps in development logs
-config :logger, :console, format: "[$level] $message\n"
+config :logger, :console,
+       format: "$time [$level] $message ($metadata)\n",
+       metadata: [:module, :function, :line]
 
 # Set a higher stacktrace during development. Avoid configuring such
 # in production as building large stacktraces may be expensive.

lib/mongo.ex

@@ -841,7 +841,7 @@ defmodule Mongo do
       %CommandSucceededEvent{
         reply: response,
         duration: duration,
-        command_name: event.command_name,
+        command_name: event.command_name, ## todo
         #request_id: event.request_id,
         #operation_id: event.operation_id,
         #connection_id: event.connection_id

lib/mongo/mongo_db_connection.ex

@@ -8,8 +8,6 @@ defmodule Mongo.MongoDBConnection do
   use DBConnection
   use Mongo.Messages
 
-  ## todo import Keywords
-
   alias Mongo.Events
   alias Mongo.Events.CommandStartedEvent
   alias Mongo.MongoDBConnection.Utils
@@ -223,7 +221,7 @@ defmodule Mongo.MongoDBConnection do
   @impl true
   def ping(%{wire_version: _wire_version} = state) do
     ###with {:ok, %{wire_version: ^wire_version}} <- wire_version(state), do: {:ok, state}
-    ## Logger.info("Ignoring ping")
+    ## todo Logger.info("Ignoring ping")
     {:ok, state}
   end
 
@@ -268,11 +266,9 @@ defmodule Mongo.MongoDBConnection do
     Events.notify(event, :commands)
 
     with {duration, {:ok, flags, doc}} <- :timer.tc(fn -> Utils.get_response(state.request_id, %{state | timeout: timeout}) end)do
-      Logger.info("more_to_come-Response: flags: #{inspect  Integer.to_string(flags, 2)} and docs #{inspect doc}")
       {:ok, {doc, event, flags, duration}, state}
     else
       {_duration, error} ->
-        Logger.info("More to come error")
         error
     end
   end
@@ -305,11 +301,9 @@ defmodule Mongo.MongoDBConnection do
 
     with {duration, {:ok, flags, doc}} <- :timer.tc(fn -> Utils.post_request(op, state.request_id, %{state | timeout: timeout}) end),
          state = %{state | request_id: state.request_id + 1} do
-      Logger.info("Post-Request ok")
       {:ok, {doc, event, flags, duration}, state}
     else
       {_duration, error} ->
-      Logger.info("Post-Request error")
         error
     end
   end

lib/mongo/monitor.ex

@@ -108,6 +108,7 @@ defmodule Mongo.Monitor do
 
     info("Stopping streaming")
     GenServer.stop(streaming_pid, reason)
+    info("stopped streaming")
   end
 
   @doc """
@@ -189,7 +190,7 @@ defmodule Mongo.Monitor do
       %{state | round_trip_time: round_trip_time}
     else
       error ->
-        Logger.warn("Unable to update server description because of #{inspect error}")
+        Logger.warn("Unable to round trip time because of #{inspect error}")
         state
     end
   end

lib/mongo/streaming_hello_monitor.ex

@@ -1,57 +1,44 @@
-# see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#monitoring
 defmodule Mongo.StreamingHelloMonitor do
   @moduledoc """
-  Each server has a monitor process. The monitor process is created by the topology process.
+   See https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-monitoring.rst#streaming-protocol
 
-  If the network connection is working, then the monitor process reports this and the topology process starts the
-  connection pool. Per server we get 1 + pool size connections to each server.
+  The streaming protocol is used to monitor MongoDB 4.4+ servers and optimally reduces the time it takes for a client to discover server state changes.
+  Multi-threaded or asynchronous drivers MUST use the streaming protocol when connected to a server that supports the awaitable hello or legacy hello commands.
+  This protocol requires an extra thread and an extra socket for each monitor to perform RTT calculations.
 
-  After waiting for `heartbeat_frequency_ms` milliseconds, the monitor process calls `isMaster` command and
-  reports the result to the topology process.
-
-  The result of the `isMaster` command is mapped the `ServerDescription` structure and sent to the topology process, which
-  updates it internal data structure.
+  This module implements the streaming protocol. The GenServer is started and maintained by the Monitor process. The streaming hello monitor uses the
+  more to come flag while updating the current server description.
   """
   require Logger
 
   use GenServer
   use Bitwise
 
+  alias Mongo.Events.ServerHeartbeatFailedEvent
+  alias Mongo.Events.ServerHeartbeatStartedEvent
+  alias Mongo.Events.ServerHeartbeatSucceededEvent
+  alias Mongo.ServerDescription
   alias Mongo.StreamingHelloMonitor
   alias Mongo.Topology
-  alias Mongo.ServerDescription
-  alias Mongo.Events.{ServerHeartbeatStartedEvent, ServerHeartbeatFailedEvent,ServerHeartbeatSucceededEvent}
 
-  # this is not configurable because the specification says so
-  # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#minheartbeatfrequencyms
-  # not used @min_heartbeat_frequency_ms 500
+  @more_to_come_mask 0x2
 
   def start_link(args) do
     GenServer.start_link(StreamingHelloMonitor, args)
   end
 
-  @doc """
-  Try to update the server description.
-  """
-  def force_check(pid) do
-    GenServer.cast(pid, :update)
-  end
-
-  def set_heartbeat_frequency_ms(pid, heartbeat_frequency_ms) do
-    GenServer.cast(pid, {:update, heartbeat_frequency_ms})
-  end
-
   @doc """
   Initialize the monitor process
   """
   def init([topology_pid, address, heartbeat_frequency_ms, opts]) do
 
+    heartbeat_frequency_ms = 10_000
+
     opts = opts
            |> Keyword.drop([:after_connect])
-           |> Keyword.put(:after_connect, {__MODULE__, :connected, [self()]})
+           |> Keyword.put(:after_connect, {StreamingHelloMonitor, :connected, [self()]})
            |> Keyword.put(:connection_type, :stream_monitor)
 
-
     info("Starting stream hello monitor with options #{inspect(opts, pretty: true)}")
 
     with {:ok, pid} <- DBConnection.start_link(Mongo.MongoDBConnection, opts) do
@@ -60,7 +47,7 @@ defmodule Mongo.StreamingHelloMonitor do
         topology_pid: topology_pid,                     ## the topology_pid to which we report
         address: address,                               ## the address of the server, needed to make updates
         heartbeat_frequency_ms: heartbeat_frequency_ms, ## current heartbeat_frequency_ms
-        max_await_time_ms: 10000,
+        max_await_time_ms: heartbeat_frequency_ms,
         more_to_come: false,
         topology_version: nil, # {processId: <ObjectId>, counter: <int64>},
         opts: opts ## options
@@ -72,8 +59,8 @@ defmodule Mongo.StreamingHelloMonitor do
   @doc """
   In this case we stop the DBConnection.
   """
-  def terminate(reason, %{connection_pid: connection_pid} = state) do
-    info("Terminating streaming hello monitor for reason #{inspect reason}")
+  def terminate(reason, %{connection_pid: connection_pid}) do
+    ## debug info("Terminating streaming hello monitor for reason #{inspect reason}")
     GenServer.stop(connection_pid, reason)
   end
 
@@ -93,7 +80,7 @@ defmodule Mongo.StreamingHelloMonitor do
 
   def handle_info(:update, state) do
     new_state = update_server_description(state)
-    Process.send_after(self(), :update, new_state.heartbeat_frequency_ms)
+    send(self(), :update)
     {:noreply, new_state}
   end
 
@@ -102,12 +89,12 @@ defmodule Mongo.StreamingHelloMonitor do
   #
   defp update_server_description(%{topology_pid: topology_pid} = state) do
     with {topology_version, flags, server_description} <- get_server_description(state) do
+      info("Updating server description")
       Topology.update_server_description(topology_pid, server_description)
       state = %{state | topology_version: topology_version}
 
-      case flags &&& 0x2 do
-        0x2 ->
-          info("More to come")
+      case flags &&& @more_to_come_mask do
+        @more_to_come_mask ->
           state = %{state | more_to_come: true}
           update_server_description(state)
         _other ->
@@ -119,39 +106,32 @@ defmodule Mongo.StreamingHelloMonitor do
 
   # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#network-error-when-calling-ismaster
   ##
-  # Calls isMaster and parses the result to update the server description
-  # In case of one network error, the function tries one more time to call isMaster command
+  # Calls hello command and parses the result to update the server description
   #
-  defp get_server_description(%{connection_pid: conn_pid, address: address, topology_version: topology_version, opts: opts} = state) do
+  defp get_server_description(%{connection_pid: conn_pid, address: address, topology_version: topology_version, max_await_time_ms: max_await_time_ms, opts: opts} = state) do
 
     Mongo.Events.notify(%ServerHeartbeatStartedEvent{ connection_pid: conn_pid})
 
     {duration, result} = case state do
       %{more_to_come: true} ->
-        :timer.tc(fn ->
-
-          info("Calling more to come")
-          result = Mongo.exec_more_to_come(conn_pid, opts)
-          info("End of more to come")
-          result
-        end)
+        :timer.tc(fn -> Mongo.exec_more_to_come(conn_pid, opts) end)
 
       _other ->
+        opts = Keyword.put(opts, :max_await_time_ms, max_await_time_ms)
         :timer.tc(fn -> hello_command(conn_pid, topology_version, opts) end)
     end
 
     case result do
       {:ok, {flags, hello_doc}} ->
 
-        info("Got flags: #{inspect flags}")
-
         server_description = hello_doc
                              |> ServerDescription.parse_hello_response()
                              |> Map.put(:address, address)
                              |> Map.put(:last_update_time, DateTime.utc_now())
                              |> Map.put(:error, nil)
 
         notify_success(duration, hello_doc, conn_pid)
+
         {hello_doc["topologyVersion"], flags, server_description}
 
       {:error, error} ->
@@ -168,7 +148,13 @@ defmodule Mongo.StreamingHelloMonitor do
 
   defp hello_command(conn_pid, %{"counter" => counter, "processId" => process_id}, opts) do
     opts = Keyword.merge(opts, [flags: [:exhaust_allowed]])
-    Mongo.exec_command(conn_pid, [isMaster: 1, maxAwaitTimeMS: 10_000, topologyVersion: %{counter: %BSON.LongNumber{value: counter}, processId: process_id}], opts)
+    cmd = [isMaster: 1,
+      maxAwaitTimeMS: Keyword.get(opts, :max_await_time_ms, 10_000),
+      topologyVersion: %{
+        counter: %BSON.LongNumber{value: counter},
+        processId: process_id}
+    ]
+    Mongo.exec_command(conn_pid, cmd, opts)
   end
 
   defp hello_command(conn_pid, _topology_version, opts) do
@@ -183,7 +169,7 @@ defmodule Mongo.StreamingHelloMonitor do
     Mongo.Events.notify(%ServerHeartbeatSucceededEvent{duration: duration, reply: reply, connection_pid: conn_pid})
   end
 
-  defp info(message) do
+  def info(message) do
     Logger.info(IO.ANSI.format([:blue, :bright, message]))
   end
 

lib/mongo/topology.ex

@@ -113,11 +113,8 @@ defmodule Mongo.Topology do
   ## GenServer Callbacks
 
   # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#configuration
-  @doc false
   def init(opts) do
 
-    Logger.info("Starting topology!!")
-
     seeds              = Keyword.get(opts, :seeds, [seed(opts)])
     type               = Keyword.get(opts, :type, :unknown)
     set_name           = Keyword.get(opts, :set_name, nil)
@@ -158,9 +155,6 @@ defmodule Mongo.Topology do
   end
 
   def terminate(_reason, state) do
-
-    Logger.info("Terminating Topology")
-
     case state.opts[:pw_safe] do
        nil -> nil
        pid -> GenServer.stop(pid)
@@ -195,18 +189,27 @@ defmodule Mongo.Topology do
     {:noreply, new_state}
   end
 
-  def handle_cast({:disconnect, kind, host}, state) do
+  ##
+  # In case of :monitor or :stream_monitor we mark the server description of the address as unknown
+  ##
+  def handle_cast({:disconnect, kind, address}, state) when kind in [:monitor, :stream_monitor] do
+    server_description = ServerDescription.parse_hello_response(address, "#{inspect kind} disconnected")
 
-    Logger.info("Disconnection by #{inspect kind}")
-    new_state = remove_address(host, state)
-    maybe_reinit(new_state)
-    {:noreply, new_state}
+    new_state = address
+                |> remove_address(state)
+                |> maybe_reinit()
+
+    handle_cast({:server_description, server_description}, new_state)
   end
 
-  def handle_cast({:disconnect, :client, _host}, state) do
+  def handle_cast({:disconnect, _kind, _host}, state) do
     {:noreply, state}
   end
 
+  ##
+  # After the monitor is connected to the server, the connection pool is started and
+  # the "waiting pids" are informed to call the command again
+  ##
   def handle_cast({:connected, monitor_pid}, state) do
     monitor = Enum.find(state.monitors, fn {_key, value} -> value == monitor_pid end)
     new_state = case monitor do
@@ -475,8 +478,9 @@ defmodule Mongo.Topology do
     state
   end
 
-  defp maybe_reinit(%{monitors: monitors} = state) when map_size(monitors) > 0,
-    do: state
+  defp maybe_reinit(%{monitors: monitors} = state) when map_size(monitors) > 0 do
+    state
+  end
   defp maybe_reinit(state) do
     servers = servers_from_seeds(state.seeds)
 
@@ -493,17 +497,13 @@ defmodule Mongo.Topology do
 
   defp remove_address(address, state) do
 
-    IO.inspect(Process.info(self(), :current_stacktrace), label: "STACKTRACE")
+    Mongo.Events.notify(%ServerClosedEvent{address: address, topology_pid: self()})
+    GenServer.stop(state.monitors[address])
 
-    Logger.info("Removing address #{inspect address}")
-
-    :ok = Mongo.Events.notify(%ServerClosedEvent{address: address, topology_pid: self()})
-    :ok = GenServer.stop(state.monitors[address])
-
-    :ok = case state.connection_pools[address] do
-            nil -> :ok
-            pid -> GenServer.stop(pid)
-          end
+    case state.connection_pools[address] do
+      nil -> :ok
+      pid -> GenServer.stop(pid)
+    end
 
     %{state | monitors: Map.delete(state.monitors, address),
       connection_pools: Map.delete(state.connection_pools, address)}

lib/mongo/topology_description.ex

@@ -67,10 +67,7 @@ defmodule Mongo.TopologyDescription do
   end
 
   @doc """
-  Returns a tuple of three values:
-  * servers: possible list of servers, maybe []
-  * slave_ok:
-  * mongod?:
+  Selects the next possible server from the current topology.
   """
   def select_servers(topology, type, opts \\ [])
   def select_servers(%{:compatible => false}, _type, _opts) do
@@ -85,6 +82,7 @@ defmodule Mongo.TopologyDescription do
     end
 
     addr = servers
+           |> Enum.filter(fn {_, %{type: type}} -> type != :unknown end) ## only valid servers
            |> Enum.map(fn {server, _} -> server end)
            |> Enum.take_random(1)