Detailed observability (#37)

This PR rename some existing observability metrics and introduce more
detailed metrics:

On message brokering:
- Rename `current_connections` to `concurrent_subscription` and split
subscriptions per issuer,
- Rename `connections[created / released]` to
`subscription_lifecycle[created/released]` and split per issuer,
- Add `subscription_duration_ms` to track the duration of SSE
subscriptions and split per issuer,
- Rename `messages[received/published]` by `messages[published/sent]`
and split per issuer (published = message published on the internal API,
sent = message sent through a SSE connection),
- Add `topic_count` to track how many topics are stored in the in-memory
message retention.

On system metrics:
- Add the boolean `stopping` that turned to `true` when a node is in the
graceful shutdown process.

Add http metrics:
- `http_request_count[public/internal, status]` count the number of http
request on each interfaces and per HTTP status, it is incremented at the
end of the request processing.
- `http_request_duration_ms[public/internal]` report the duration of
http request on each interfaces.

Author: achouippe

neurow/lib/neurow/application.ex

@@ -93,7 +93,7 @@ defmodule Neurow.Application do
         end
 
     MetricsPlugExporter.setup()
-    Stats.setup()
+    Neurow.Stats.setup()
     JOSE.json_module(:jiffy)
 
     opts = [strategy: :one_for_one, name: Neurow.Supervisor]

neurow/lib/neurow/broker/receiver_shard.ex

@@ -21,6 +21,10 @@ defmodule Neurow.Broker.ReceiverShard do
     GenServer.call(shard, {:flush_history})
   end
 
+  def topic_count(shard) do
+    GenServer.call(shard, {:topic_count})
+  end
+
   @impl true
   def init(shard) do
     :ok =
@@ -73,4 +77,9 @@ defmodule Neurow.Broker.ReceiverShard do
     :ets.delete_all_objects(table_1)
     {:reply, :ok, {table_0, table_1}}
   end
+
+  @impl true
+  def handle_call({:topic_count}, _from, {table_0, table_1}) do
+    {:reply, :ets.info(table_0)[:size] + :ets.info(table_1)[:size], {table_0, table_1}}
+  end
 end

neurow/lib/neurow/broker/receiver_shard_manager.ex

@@ -35,20 +35,27 @@ defmodule Neurow.Broker.ReceiverShardManager do
     {:noreply, state}
   end
 
-  def all_pids(fun) do
+  def receiver_shards() do
     Enum.map(0..(@shards - 1), fn shard ->
-      fun.({shard, Neurow.Broker.ReceiverShard.build_name(shard)})
+      {shard, Neurow.Broker.ReceiverShard.build_name(shard)}
     end)
   end
 
   def rotate do
-    Stats.inc_history_rotate()
+    Neurow.Stats.MessageBroker.inc_history_rotate()
 
-    all_pids(fn {_, pid} ->
+    Enum.each(receiver_shards(), fn {_shard, pid} ->
       send(pid, {:rotate})
     end)
   end
 
+  def topic_count do
+    receiver_shards()
+    |> Enum.map(fn {_shard, pid} -> pid end)
+    |> Enum.map(&Neurow.Broker.ReceiverShard.topic_count/1)
+    |> Enum.sum()
+  end
+
   @impl true
   def handle_call({:rotate}, _, state) do
     rotate()
@@ -57,7 +64,7 @@ defmodule Neurow.Broker.ReceiverShardManager do
 
   @impl true
   def handle_call({:flush_history}, _from, state) do
-    all_pids(fn {_, pid} ->
+    Enum.each(receiver_shards(), fn {_shard, pid} ->
       pid |> Neurow.Broker.ReceiverShard.flush_history()
     end)
 
@@ -74,7 +81,7 @@ defmodule Neurow.Broker.ReceiverShardManager do
   end
 
   def create_receivers() do
-    all_pids(fn {shard, pid} ->
+    Enum.map(receiver_shards(), fn {shard, pid} ->
       Supervisor.child_spec({Neurow.Broker.ReceiverShard, shard}, id: pid)
     end)
   end

neurow/lib/neurow/configuration.ex

@@ -5,6 +5,11 @@ defmodule Neurow.Configuration do
     GenServer.start_link(__MODULE__, default, name: __MODULE__)
   end
 
+  def issuers() do
+    Application.fetch_env!(:neurow, :public_api_authentication)[:issuers]
+    |> Map.keys()
+  end
+
   def public_api_issuer_jwks(issuer_name) do
     GenServer.call(__MODULE__, {:public_api_issuer_jwks, issuer_name})
   end

neurow/lib/neurow/internal_api/endpoint.ex

@@ -15,7 +15,7 @@ defmodule Neurow.InternalApi.Endpoint do
       &Neurow.Configuration.internal_api_verbose_authentication_errors/0,
     max_lifetime: &Neurow.Configuration.internal_api_jwt_max_lifetime/0,
     send_forbidden: &Neurow.InternalApi.Endpoint.send_forbidden/3,
-    inc_error_callback: &Stats.inc_jwt_errors_internal/0,
+    inc_error_callback: &Neurow.Stats.Security.inc_jwt_errors_internal/0,
     exclude_path_prefixes: ["/ping", "/nodes", "/cluster_size_above", "/history"]
   )
 
@@ -78,7 +78,7 @@ defmodule Neurow.InternalApi.Endpoint do
 
   post "/v1/publish" do
     case extract_params(conn) do
-      {:ok, messages, topics} ->
+      {:ok, issuer, messages, topics} ->
         publish_timestamp = :os.system_time(:millisecond)
 
         nb_publish =
@@ -96,7 +96,7 @@ defmodule Neurow.InternalApi.Endpoint do
           end)
         end)
 
-        Stats.inc_msg_received()
+        Neurow.Stats.MessageBroker.inc_message_published(issuer)
 
         conn
         |> put_resp_header("content-type", "application/json")
@@ -126,7 +126,7 @@ defmodule Neurow.InternalApi.Endpoint do
       full_topics =
         Enum.map(PublishRequest.topics(publish_request), fn topic -> "#{issuer}-#{topic}" end)
 
-      {:ok, PublishRequest.messages(publish_request), full_topics}
+      {:ok, issuer, PublishRequest.messages(publish_request), full_topics}
     else
       error ->
         error

neurow/lib/neurow/public_api/endpoint.ex

@@ -2,9 +2,6 @@ defmodule Neurow.PublicApi.Endpoint do
   require Logger
   import Plug.Conn
   use Plug.Router
-
-  plug(:monitor_sse)
-
   plug(:preflight_request)
 
   plug(Neurow.JwtAuthPlug,
@@ -14,7 +11,7 @@ defmodule Neurow.PublicApi.Endpoint do
     verbose_authentication_errors:
       &Neurow.Configuration.public_api_verbose_authentication_errors/0,
     max_lifetime: &Neurow.Configuration.public_api_jwt_max_lifetime/0,
-    inc_error_callback: &Stats.inc_jwt_errors_public/0
+    inc_error_callback: &Neurow.Stats.Security.inc_jwt_errors_public/0
   )
 
   plug(:match)
@@ -52,6 +49,9 @@ defmodule Neurow.PublicApi.Endpoint do
         :ok = Neurow.StopListener.subscribe()
         :ok = Phoenix.PubSub.subscribe(Neurow.PubSub, topic)
 
+        {:ok, _pid} =
+          Neurow.PublicApi.SSEMonitor.start_link(issuer)
+
         last_event_id = extract_last_event_id(conn)
 
         case last_event_id do
@@ -71,7 +71,7 @@ defmodule Neurow.PublicApi.Endpoint do
             conn
             |> send_chunked(200)
             |> import_history(topic, last_event_id)
-            |> loop(timeout_ms, keep_alive_ms, now_ms, now_ms, exp)
+            |> loop(timeout_ms, keep_alive_ms, now_ms, now_ms, exp, issuer)
         end
 
       _ ->
@@ -186,7 +186,7 @@ defmodule Neurow.PublicApi.Endpoint do
     {conn, sent}
   end
 
-  def loop(conn, sse_timeout_ms, keep_alive_ms, last_message_ms, last_ping_ms, jwt_exp_s) do
+  def loop(conn, sse_timeout_ms, keep_alive_ms, last_message_ms, last_ping_ms, jwt_exp_s, issuer) do
     now_ms = :os.system_time(:millisecond)
 
     cond do
@@ -204,7 +204,8 @@ defmodule Neurow.PublicApi.Endpoint do
           keep_alive_ms,
           last_message_ms,
           now_ms,
-          jwt_exp_s
+          jwt_exp_s,
+          issuer
         )
 
       # JWT token expired, send a credentials_expired event and stop the connection
@@ -226,7 +227,8 @@ defmodule Neurow.PublicApi.Endpoint do
         receive do
           {:pubsub_message, message} ->
             conn = write_chunk(conn, message)
-            Stats.inc_msg_published()
+
+            Neurow.Stats.MessageBroker.inc_message_sent(issuer)
             new_last_message_ms = :os.system_time(:millisecond)
 
             conn
@@ -235,7 +237,8 @@ defmodule Neurow.PublicApi.Endpoint do
               keep_alive_ms,
               new_last_message_ms,
               new_last_message_ms,
-              jwt_exp_s
+              jwt_exp_s,
+              issuer
             )
 
           :shutdown ->
@@ -244,10 +247,26 @@ defmodule Neurow.PublicApi.Endpoint do
 
           # Consume useless messages to avoid memory overflow
           _ ->
-            conn |> loop(sse_timeout_ms, keep_alive_ms, last_message_ms, last_ping_ms, jwt_exp_s)
+            loop(
+              conn,
+              sse_timeout_ms,
+              keep_alive_ms,
+              last_message_ms,
+              last_ping_ms,
+              jwt_exp_s,
+              issuer
+            )
         after
           next_tick_ms ->
-            conn |> loop(sse_timeout_ms, keep_alive_ms, last_message_ms, last_ping_ms, jwt_exp_s)
+            loop(
+              conn,
+              sse_timeout_ms,
+              keep_alive_ms,
+              last_message_ms,
+              last_ping_ms,
+              jwt_exp_s,
+              issuer
+            )
         end
     end
   end
@@ -301,9 +320,4 @@ defmodule Neurow.PublicApi.Endpoint do
       String.match?(origin, allowed_origin)
     end)
   end
-
-  defp monitor_sse(conn, _) do
-    {:ok, _pid} = SSEMonitor.start_link(conn)
-    conn
-  end
 end

neurow/lib/neurow/public_api/sse_monitor.ex

@@ -0,0 +1,31 @@
+defmodule Neurow.PublicApi.SSEMonitor do
+  require Logger
+  use GenServer
+
+  def start_link(issuer) do
+    GenServer.start_link(__MODULE__, {issuer, :os.system_time()})
+  end
+
+  @impl true
+  def init({issuer, start_time}) do
+    Neurow.Stats.MessageBroker.inc_subscriptions(issuer)
+    Process.flag(:trap_exit, true)
+    {:ok, {issuer, start_time}}
+  end
+
+  @impl true
+  def terminate(:normal, {issuer, start_time}) do
+    track_subscription_end(issuer, start_time)
+    Logger.debug("SSE connection end")
+  end
+
+  def terminate(reason, {issuer, start_time}) do
+    track_subscription_end(issuer, start_time)
+    Logger.debug("SSE connection terminated: #{inspect(reason)}")
+  end
+
+  defp track_subscription_end(issuer, start_time) do
+    duration_ms = System.convert_time_unit(:os.system_time() - start_time, :native, :millisecond)
+    Neurow.Stats.MessageBroker.dec_subscriptions(issuer, duration_ms)
+  end
+end