feat: add /metrics/:region endpoint (#1656)

Author: edgurgel

lib/realtime_web/controllers/metrics_controller.ex

@@ -6,26 +6,36 @@ defmodule RealtimeWeb.MetricsController do
 
   # We give more memory and time to collect metrics from all nodes as this is a lot of work
   def index(conn, _) do
-    {time, metrics} = :timer.tc(&cluster_metrics/0, :millisecond)
+    {time, metrics} = :timer.tc(fn -> metrics([Node.self() | Node.list()]) end, :millisecond)
     Logger.info("Collected cluster metrics in #{time} milliseconds")
 
     conn
     |> put_resp_content_type("text/plain")
     |> send_resp(200, metrics)
   end
 
-  defp cluster_metrics() do
+  def region(conn, %{"region" => region}) do
+    nodes = Realtime.Nodes.region_nodes(region)
+    {time, metrics} = :timer.tc(fn -> metrics(nodes) end, :millisecond)
+    Logger.info("Collected metrics for region #{region} in #{time} milliseconds")
+
+    conn
+    |> put_resp_content_type("text/plain")
+    |> send_resp(200, metrics)
+  end
+
+  defp metrics(nodes) do
     bump_max_heap_size()
     timeout = Application.fetch_env!(:realtime, :metrics_rpc_timeout)
 
-    Node.list()
+    nodes
     |> Task.async_stream(
       fn node ->
         {node, GenRpc.call(node, __MODULE__, :get_metrics, [], timeout: timeout)}
       end,
       timeout: :infinity
     )
-    |> Enum.reduce([PromEx.get_metrics()], fn {_, {node, response}}, acc ->
+    |> Enum.reduce([], fn {_, {node, response}}, acc ->
       case response do
         {:error, :rpc_error, reason} ->
           Logger.error("Cannot fetch metrics from the node #{inspect(node)} because #{inspect(reason)}")

lib/realtime_web/router.ex

@@ -76,6 +76,7 @@ defmodule RealtimeWeb.Router do
     pipe_through(:metrics)
 
     get("/", MetricsController, :index)
+    get("/:region", MetricsController, :region)
   end
 
   scope "/api" do

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.67.8",
+      version: "2.68.0",
       elixir: "~> 1.18",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/realtime_web/controllers/metrics_controller_test.exs

@@ -2,8 +2,10 @@ defmodule RealtimeWeb.MetricsControllerTest do
   # Usage of Clustered
   # Also changing Application env
   use RealtimeWeb.ConnCase, async: false
+  alias Realtime.GenRpc
 
   import ExUnit.CaptureLog
+  use Mimic
 
   setup_all do
     metrics_tags = %{
@@ -45,9 +47,13 @@ defmodule RealtimeWeb.MetricsControllerTest do
     end
 
     test "returns 200 and log on timeout", %{conn: conn} do
-      current_value = Application.get_env(:realtime, :metrics_rpc_timeout)
-      on_exit(fn -> Application.put_env(:realtime, :metrics_rpc_timeout, current_value) end)
-      Application.put_env(:realtime, :metrics_rpc_timeout, 0)
+      Mimic.stub(GenRpc, :call, fn node, mod, func, args, opts ->
+        if node != node() do
+          {:error, :rpc_error, :timeout}
+        else
+          call_original(GenRpc, :call, [node, mod, func, args, opts])
+        end
+      end)
 
       log =
         capture_log(fn ->
@@ -84,4 +90,64 @@ defmodule RealtimeWeb.MetricsControllerTest do
                |> response(403)
     end
   end
+
+  describe "GET /metrics/:region" do
+    setup %{conn: conn} do
+      # The metrics pipeline requires authentication
+      jwt_secret = Application.fetch_env!(:realtime, :metrics_jwt_secret)
+      token = generate_jwt_token(jwt_secret, %{})
+      authenticated_conn = put_req_header(conn, "authorization", "Bearer #{token}")
+
+      {:ok, conn: authenticated_conn}
+    end
+
+    test "returns 200", %{conn: conn} do
+      assert response =
+               conn
+               |> get(~p"/metrics/ap-southeast-2")
+               |> text_response(200)
+
+      # Check prometheus like metrics
+      assert response =~
+               "# HELP beam_system_schedulers_online_info The number of scheduler threads that are online."
+
+      assert response =~ "region=\"ap-southeast-2\""
+      refute response =~ "region=\"us-east-1\""
+    end
+
+    test "returns 200 and log on timeout", %{conn: conn} do
+      Mimic.stub(GenRpc, :call, fn _node, _mod, _func, _args, _opts ->
+        {:error, :rpc_error, :timeout}
+      end)
+
+      log =
+        capture_log(fn ->
+          assert response =
+                   conn
+                   |> get(~p"/metrics/ap-southeast-2")
+                   |> text_response(200)
+
+          assert response == ""
+        end)
+
+      assert log =~ "Cannot fetch metrics from the node"
+    end
+
+    test "returns 403 when authorization header is missing", %{conn: conn} do
+      assert conn
+             |> delete_req_header("authorization")
+             |> get(~p"/metrics/ap-southeast-2")
+             |> response(403)
+    end
+
+    test "returns 403 when authorization header is wrong", %{conn: conn} do
+      token = generate_jwt_token("bad_secret", %{})
+
+      assert _ =
+               conn
+               |> put_req_header("authorization", "Bearer #{token}")
+               |> get(~p"/metrics/ap-southeast-2")
+               |> response(403)
+    end
+  end
 end