feat: block queue metrics and stalled submission alarm

apps/omg/lib/omg/state.ex

@@ -261,10 +261,10 @@ defmodule OMG.State do
     - pushes the new block to subscribers of `"blocks"` internal event bus topic
   """
   def handle_cast(:form_block, state) do
-    _ = Logger.debug("Forming new block...")
+    _ = Logger.info("Forming new block...")
     state = Core.claim_fees(state)
     {:ok, {%Block{number: blknum} = block, db_updates}, new_state} = Core.form_block(state)
-    _ = Logger.debug("Formed new block ##{blknum}")
+    _ = Logger.info("Formed new block ##{blknum}")
 
     # persistence is required to be here, since propagating the block onwards requires restartability including the
     # new block

apps/omg_child_chain/lib/omg_child_chain/block_queue.ex

@@ -194,6 +194,11 @@ defmodule OMG.ChildChain.BlockQueue do
     {:noreply, state1}
   end
 
+  @doc false
+  # Ignore unrelated events
+  def handle_info({:internal_event_bus, :block_submitting, _}, state), do: {:noreply, state}
+  def handle_info({:internal_event_bus, :block_submitted, _}, state), do: {:noreply, state}
+
   # private (server)
 
   @spec submit_blocks(Core.t()) :: :ok
@@ -205,6 +210,7 @@ defmodule OMG.ChildChain.BlockQueue do
 
   defp submit(submission) do
     _ = Logger.info("Submitting: #{inspect(submission)}")
+    _ = publish_block_submitting_event(submission.num)
 
     submit_result = Eth.submit_block(submission.hash, submission.nonce, submission.gas_price)
     newest_mined_blknum = RootChain.get_mined_child_block()
@@ -218,6 +224,7 @@ defmodule OMG.ChildChain.BlockQueue do
           error
 
         {:ok, txhash} ->
+          _ = publish_block_submitted_event(submission.num)
           _ = GasAnalyzer.enqueue(txhash)
           _ = Balance.check()
           :ok
@@ -229,6 +236,34 @@ defmodule OMG.ChildChain.BlockQueue do
     :ok = final_result
   end
 
+  # Publishes each time a block is being submitted, regardless of failing or being successful.
+  # This differs from `:enqueue_block` which publishes only once when the block is formed.
+  # For example, when a block is re-submitted 3 times before it got accepted, there would be
+  # 1 x `:enqueue_block` and 3 x `:block_submitting` events published.
+  #
+  # The telemetry event is emitted for raw metrics propagation. The bus event is published so
+  # a consumer, potentially a GenServer or other processes can perform extra operations on the data.
+  defp publish_block_submitting_event(blknum) do
+    _ = :telemetry.execute([:blknum_submitting, __MODULE__], %{blknum: blknum})
+
+    {:child_chain, "blocks"}
+    |> OMG.Bus.Event.new(:block_submitting, blknum)
+    |> OMG.Bus.direct_local_broadcast()
+  end
+
+  # Publishes when a block is successfully submitted. Only 1 `block_submitted` event will ever
+  # be published for each block.
+  #
+  # The telemetry event is emitted for raw metrics propagation. The bus event is published so
+  # a consumer, potentially a GenServer or other processes can perform extra operations on the data.
+  defp publish_block_submitted_event(blknum) do
+    _ = :telemetry.execute([:blknum_submitted, __MODULE__], %{blknum: blknum})
+
+    {:child_chain, "blocks"}
+    |> OMG.Bus.Event.new(:block_submitted, blknum)
+    |> OMG.Bus.direct_local_broadcast()
+  end
+
   defp log_init_error(fields) do
     fields = Keyword.update!(fields, :known_hashes, fn hashes -> Enum.map(hashes, &Encoding.to_hex/1) end)
 

apps/omg_child_chain/lib/omg_child_chain/block_queue/measure.ex

@@ -27,16 +27,23 @@ defmodule OMG.ChildChain.BlockQueue.Measure do
   alias OMG.ChildChain.BlockQueue
   alias OMG.ChildChain.BlockQueue.Balance
   alias OMG.ChildChain.BlockQueue.GasAnalyzer
+  alias OMG.ChildChain.BlockQueue.SubmissionMonitor
   alias OMG.Status.Metric.Datadog
 
   @supported_events [
     [:process, BlockQueue],
+    # Events triggered when a block is submitting or submitted
+    [:blknum_submitting, BlockQueue],
+    [:blknum_submitted, BlockQueue],
+    # Events providing collections of blocks being submitted or stalled
+    [:blocks_submitting, SubmissionMonitor],
+    [:blocks_stalled, SubmissionMonitor],
     [:gas, GasAnalyzer],
     [:authority_balance, Balance]
   ]
   def supported_events(), do: @supported_events
 
-  def handle_event([:process, BlockQueue], _, _state, _config) do
+  def handle_event([:process, BlockQueue], _measurements, _metadata, _config) do
     value =
       self()
       |> Process.info(:message_queue_len)
@@ -47,11 +54,25 @@ defmodule OMG.ChildChain.BlockQueue.Measure do
 
   def handle_event([:gas, GasAnalyzer], %{gas: gas}, _, _config) do
     gwei = div(gas, 1_000_000_000)
-    _ = Datadog.gauge(name(:block_submission), gwei)
+    _ = Datadog.gauge(name(:block_submission_gas), gwei)
   end
 
   def handle_event([:authority_balance, Balance], %{authority_balance: authority_balance}, _, _config) do
     gwei = div(authority_balance, 1_000_000_000)
     _ = Datadog.gauge(name(:authority_balance), gwei)
   end
+
+  def handle_event([:blknum_submitting, BlockQueue], %{blknum: blknum}, _, _config) do
+    _ = Datadog.gauge(name(:block_queue_blknum_submitting), blknum)
+    _ = Datadog.increment(name(:block_submission_attempt), 1)
+  end
+
+  def handle_event([:blknum_submitted, BlockQueue], %{blknum: blknum}, _, _config) do
+    _ = Datadog.gauge(name(:block_queue_blknum_submitted), blknum)
+    _ = Datadog.increment(name(:block_submission_success), 1)
+  end
+
+  def handle_event([:blocks_stalled, SubmissionMonitor], %{blocks: blocks}, _, _config) do
+    _ = Datadog.gauge(name(:block_queue_num_blocks_stalled), length(blocks))
+  end
 end

apps/omg_child_chain/lib/omg_child_chain/block_queue/monitor/alarm_handler.ex

@@ -0,0 +1,48 @@
+# Copyright 2019-2020 OmiseGO Pte Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+defmodule OMG.ChildChain.BlockQueue.Monitor.AlarmHandler do
+  @moduledoc """
+  Listens for :block_submit_stalled alarms and reflect the alarm's state back to the monitor.
+  """
+  require Logger
+
+  # The alarm reporter and monitor happen to be the same module here because we are just
+  # reflecting the alarm's state back to the reporter.
+  @reporter OMG.ChildChain.BlockQueue.Monitor
+  @monitor OMG.ChildChain.BlockQueue.Monitor
+
+  def init(_args) do
+    {:ok, %{}}
+  end
+
+  def handle_call(_request, state), do: {:ok, :ok, state}
+
+  def handle_event({:set_alarm, {:block_submit_stalled, %{reporter: @reporter}}}, state) do
+    _ = Logger.warn(":block_submit_stalled alarm raised.")
+    :ok = GenServer.cast(@monitor, {:set_alarm, :block_submit_stalled})
+    {:ok, state}
+  end
+
+  def handle_event({:clear_alarm, {:block_submit_stalled, %{reporter: @reporter}}}, state) do
+    _ = Logger.warn(":block_submit_stalled alarm cleared.")
+    :ok = GenServer.cast(@monitor, {:clear_alarm, :block_submit_stalled})
+    {:ok, state}
+  end
+
+  def handle_event(event, state) do
+    _ = Logger.info("#{__MODULE__} got event: #{inspect(event)}. Ignoring.")
+    {:ok, state}
+  end
+end

apps/omg_child_chain/lib/omg_child_chain/block_queue/submission_monitor.ex

@@ -0,0 +1,172 @@
+# Copyright 2019-2020 OmiseGO Pte Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+defmodule OMG.ChildChain.BlockQueue.SubmissionMonitor do
+  @moduledoc """
+  Listens to block events and raises :block_submit_stalled alarm when a pending block
+  doesn't get successfully submitted within the specified time threshold.
+  """
+  use GenServer
+  require Logger
+
+  defstruct pending_blocks: [],
+            root_chain_height: 0,
+            stall_threshold_blocks: 4,
+            alarm_module: nil,
+            alarm_raised: false
+
+  @typep blknum() :: pos_integer()
+  @typep pending_block() :: {blknum :: blknum(), first_submit_height :: pos_integer()}
+
+  @type t() :: %__MODULE__{
+          pending_blocks: [pending_block()],
+          root_chain_height: non_neg_integer(),
+          stall_threshold_blocks: pos_integer(),
+          alarm_module: module(),
+          alarm_raised: boolean()
+        }
+
+  #
+  # GenServer APIs
+  #
+
+  def start_link(args) do
+    GenServer.start_link(__MODULE__, args, name: __MODULE__)
+  end
+
+  #
+  # GenServer behaviors
+  #
+
+  def init(opts) do
+    _ = Logger.info("Starting #{__MODULE__}")
+    _ = install_alarm_handler()
+    event_bus = Keyword.fetch!(opts, :event_bus_module)
+    check_interval_ms = Keyword.fetch!(opts, :check_interval_ms)
+
+    state = %__MODULE__{
+      pending_blocks: [],
+      stall_threshold_blocks: Keyword.fetch!(opts, :stall_threshold_blocks),
+      alarm_module: Keyword.fetch!(opts, :alarm_module),
+      alarm_raised: false
+    }
+
+    :ok = event_bus.subscribe({:child_chain, "blocks"}, link: true)
+    :ok = event_bus.subscribe({:root_chain, "ethereum_new_height"}, link: true)
+
+    {:ok, _} = :timer.send_interval(check_interval_ms, self(), :check_stall)
+    {:ok, state}
+  end
+
+  def handle_info(:check_stall, state) do
+    stalled_blocks =
+      Enum.filter(state.pending_blocks, fn {_blknum, first_submit_height} ->
+        state.root_chain_height - first_submit_height >= state.stall_threshold_blocks
+      end)
+
+    _ = :telemetry.execute([:blocks_stalled, __MODULE__], %{blocks: stalled_blocks})
+    _ = log_stalled_blocks(stalled_blocks, state.root_chain_height)
+    _ = trigger_alarm(state.alarm_module, state.alarm_raised, stalled_blocks)
+
+    {:noreply, state}
+  end
+
+  # Keeps track of the latest root chain height
+  def handle_info({:internal_event_bus, :ethereum_new_height, new_height}, state) do
+    {:noreply, %{state | root_chain_height: new_height}}
+  end
+
+  # Listens for a block being submitted and add it to monitoring if it hasn't been tracked
+  def handle_info({:internal_event_bus, :block_submitting, blknum}, state) do
+    pending_blocks = add_new_blknum(state.pending_blocks, blknum, state.root_chain_height)
+    {:noreply, %{state | pending_blocks: pending_blocks}}
+  end
+
+  # Listens for a block that got submitted and drop it from monitoring
+  def handle_info({:internal_event_bus, :block_submitted, blknum}, state) do
+    pending_blocks = remove_blknum(state.pending_blocks, blknum)
+    {:noreply, %{state | pending_blocks: pending_blocks}}
+  end
+
+  # Ignore unrelated events
+  def handle_info({:internal_event_bus, :enqueue_block, _}, state) do
+    {:noreply, state}
+  end
+
+  #
+  # Handle incoming alarms
+  #
+  # These functions are called by the AlarmHandler so that this monitor process can update
+  # its internal state according to the raised alarms.
+  #
+  def handle_cast({:set_alarm, :block_submit_stalled}, state) do
+    {:noreply, %{state | alarm_raised: true}}
+  end
+
+  def handle_cast({:clear_alarm, :block_submit_stalled}, state) do
+    {:noreply, %{state | alarm_raised: false}}
+  end
+
+  #
+  # Private functions
+  #
+
+  # Add the blknum to tracking only if it is not already tracked
+  @spec add_new_blknum([{blknum(), any()}], blknum(), non_neg_integer()) :: [pending_block()]
+  defp add_new_blknum(pending_blocks, blknum, root_chain_height) do
+    case Enum.any?(pending_blocks, fn {pending_blknum, _} -> pending_blknum == blknum end) do
+      true -> pending_blocks
+      false -> [{blknum, root_chain_height} | pending_blocks]
+    end
+  end
+
+  @spec remove_blknum([{blknum(), any()}], blknum()) :: [pending_block()]
+  defp remove_blknum(pending_blocks, blknum) do
+    Enum.reject(pending_blocks, fn {pending_blknum, _} -> pending_blknum == blknum end)
+  end
+
+  #
+  # Alarms management
+  #
+
+  defp install_alarm_handler() do
+    case Enum.member?(:gen_event.which_handlers(:alarm_handler), __MODULE__.AlarmHandler) do
+      true -> :ok
+      _ -> :alarm_handler.add_alarm_handler(__MODULE__.AlarmHandler)
+    end
+  end
+
+  @spec trigger_alarm(alarm :: module(), alarm_raised :: boolean(), stalled_blocks :: [blknum()]) :: :ok
+  defp trigger_alarm(alarm_module, false, stalled_blocks) when length(stalled_blocks) >= 1 do
+    alarm_module.set(alarm_module.block_submit_stalled(__MODULE__))
+  end
+
+  defp trigger_alarm(alarm_module, true, []) do
+    alarm_module.clear(alarm_module.block_submit_stalled(__MODULE__))
+  end
+
+  defp trigger_alarm(_alarm_module, _alarm_raised, _stalled_blocks), do: :ok
+
+  #
+  # Logging
+  #
+  defp log_stalled_blocks([], _), do: :ok
+
+  defp log_stalled_blocks(stalled_blocks, root_chain_height) do
+    Logger.warn(
+      "#{__MODULE__}: Stalled blocks: #{inspect(stalled_blocks)}. " <>
+        "Current height: #{inspect(root_chain_height)}"
+    )
+  end
+end