# see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#monitoring defmodule Mongo.Monitor do @moduledoc """ Each server has a monitor process. The monitor process is created by the topology process. 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. 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. """ use GenServer # require Logger 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 def start_link(args) do GenServer.start_link(__MODULE__, args) end # We need to stop asynchronously because a Monitor can call the Topology # which may try to stop the same Monitor that called it. Ending in a timeout. # See issues #139 for some information. @doc """ Stop the monitor process. We need to stop asynchronously because a Monitor can call the Topology which may try to stop the same Monitor that called it. Ending in a timeout. """ def stop(pid) do GenServer.cast(pid, :stop) end @doc """ Try to update the server description. """ def force_check(pid) do GenServer.cast(pid, :update) end @doc """ Initialize the monitor process """ def init([address, topology_pid, heartbeat_frequency_ms, connection_opts]) do # monitors don't authenticate and use the "admin" database opts = connection_opts |> Keyword.put(:database, "admin") |> Keyword.put(:skip_auth, true) |> Keyword.put(:after_connect, {__MODULE__, :connected, [self(), topology_pid]}) |> Keyword.put(:backoff_min, heartbeat_frequency_ms) |> Keyword.put(:backoff_max, heartbeat_frequency_ms) |> Keyword.put(:backoff_type, :rand) |> Keyword.put(:connection_type, :monitor) |> Keyword.put(:topology_pid, topology_pid) |> Keyword.put(:pool_size, 1) |> Keyword.put(:idle_interval, 5_000) with {:ok, pid} <- DBConnection.start_link(Mongo.MongoDBConnection, opts) do {:ok, %{ connection_pid: pid, ## our connection pid to the mongodb server topology_pid: topology_pid, ## the topology_pid to which we report address: address, ## the address of the server, needed to make updates round_trip_time: nil, ## current round_trip_time, needed to make average value heartbeat_frequency_ms: heartbeat_frequency_ms, ## current heartbeat_frequency_ms opts: opts ## options }} end end @doc """ In this case we stop the DBConnection. """ def terminate(reason, state) do GenServer.stop(state.connection_pid, reason) end @doc """ Report the connection event, so the topology process can now create the connection pool. """ def connected(_connection, me, topology_pid) do Topology.monitor_connected(topology_pid, me) GenServer.cast(me, :update) end @doc """ Time to update the topology. Calling `isMaster` and updating the server description """ def handle_cast(:update, state) do new_state = update_server_description(state) # we return with heartbeat_frequency_ms, so after heartbeat_frequency_ms handle_info(:timeout...) gets called. {:noreply, new_state, new_state.heartbeat_frequency_ms} end def handle_cast(:stop, state) do exit(:normal) {:noreply, state} end @doc """ The `:timeout` call is the periodic call defined by the heartbeat frequency The ':update' call updates the server description for the topology process """ def handle_info(:timeout, state) do new_state = update_server_description(state) {:noreply, new_state, new_state.heartbeat_frequency_ms} end def handle_info(:update, state) do new_state = update_server_description(state) # we return with heartbeat_frequency_ms, so after heartbeat_frequency_ms handle_info(:timeout...) gets called. {:noreply, new_state, new_state.heartbeat_frequency_ms} end ## # Get a new server description from the server and send it to the Topology process. # defp update_server_description(%{topology_pid: topology_pid} = state) do %{:round_trip_time => round_trip_time} = server_description = get_server_description(state, 0) Topology.update_server_description(topology_pid, server_description) %{state | round_trip_time: round_trip_time} end # 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 # defp get_server_description(%{connection_pid: conn_pid, address: address, round_trip_time: last_rtt, opts: opts} = state, counter) do Mongo.Events.notify(%ServerHeartbeatStartedEvent{ connection_pid: conn_pid}) {result, finish_time, rtt} = call_is_master_command(conn_pid, opts) case result do {:ok, is_master_reply} -> notify_success(rtt, is_master_reply, conn_pid) ServerDescription.from_is_master(address, average_rtt(last_rtt, rtt), finish_time, is_master_reply) {:error, error} when counter == 1 -> notify_error(rtt, error, conn_pid) ServerDescription.from_is_master_error(address, error) {:error, _error} -> get_server_description(state, counter + 1) end end defp average_rtt(nil, rtt) do round(rtt) end defp average_rtt(last_rtt, rtt) do round(0.2 * rtt + 0.8 * last_rtt) end defp call_is_master_command(conn_pid, opts) do start_time = System.monotonic_time result = Mongo.exec_command(conn_pid, [isMaster: 1], opts) finish_time = System.monotonic_time rtt = System.convert_time_unit(finish_time - start_time, :native, :millisecond) finish_time = System.convert_time_unit(finish_time, :native, :millisecond) {result, finish_time, rtt} end defp notify_error(rtt, error, conn_pid) do :ok = Mongo.Events.notify(%ServerHeartbeatFailedEvent{ duration: rtt, failure: error, connection_pid: conn_pid }) end defp notify_success(rtt, reply, conn_pid) do :ok = Mongo.Events.notify(%ServerHeartbeatSucceededEvent{ duration: rtt, reply: reply, connection_pid: conn_pid }) end end